Is it good practice to query a database when streaming with Spark

user_13862723 0
Viewed 1101

I am trying to make a simple streaming app with Spark (and also with Flink). Input is a Kafka topic that holds some product info, and before I write this product info to somewhere else, I want to get category info of product and add to my model during streaming. I am planning to store my product-category mapping data in a PostgreSQL table, or in a Couchbase bucket. So I need to query one of these for every stream data. What I want to learn is:

  • Is it applicable in Spark/Flink?

  • Is it a good practice for streaming apps? If not, how can I achieve doing this with a big-data way? Should I store my mapping data somewhere else, or join that mapping data with streaming data in another way?

spark-streaming flink-streaming apache-flink apache-spark
2 Answers

This is quite common practice for at least Spark, but it may depend more on the database selection. Like Cassandra is very fast when doing lookup by full primary key (if you don't have a lot of data, enabling row cache may also help), I think that Couchbase may also be a good choice (I didn't work with it for a long time). Code may look something like this (full code is in this Zeppelin Notebook. This code also requires Spark Cassandra Connector 2.5.0, that has support for "direct join with Cassandra" - see the blog post on that release):

val streamingInputDF = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "10.101.36.9:9092")
      .option("subscribe", "tweets2")
      .load()

val tweetDF = streamingInputDF.selectExpr("CAST(value AS STRING)")
  .select(from_json($"value", schema).as("tweet"))
  .select($"tweet.payload.created_at".as("created_at").cast(TimestampType),
    $"tweet.payload.lang".as("language"))

val streamingCountsDF =  tweetDF
  .where(col("language").isNotNull)
  .groupBy($"language", window($"created_at", "1 minutes"))
  .count()
  .select($"language", $"window.start".as("ts"), $"count")

import org.apache.spark.sql.cassandra._
val lang_details = spark.read.cassandraFormat("languages", "zep").load()

val joined = streamingCountsDF.join(lang_details, 
       lang_details("id") === streamingCountsDF("language"), "left_outer")
  .select($"language", $"native_name".as("lang_name"), $"ts", $"count")

...

Also, you need to take into account other requirements - how often you'll get updates, how fast these updates need to be propagated to streaming job, etc. For Spark, you may, for example, have a separate dataframe for data that is stored in the database, and you cache this data for faster joining, but refresh dataframe every N minutes, to get latest updates from DB. (you can find source code for that in the Stream Processing with Apache Spark book)

Alex_Ott_186270

This is a common pattern in Flink applications, and there are a few approaches to consider:

(1) You can use an external database to do lookup joins. The Table API has built-in support for doing this with JDBC databases (including PostgreSQL). For example here is an enrichment join of a Kafka stream with a lookup table in MySQL, where the MySQL table is being accessed via a Hive catalog:

SELECT
  l_proctime AS `querytime`,
  l_orderkey AS `order`,
  l_linenumber AS `linenumber`,
  l_currency AS `currency`,
  rs_rate AS `cur_rate`, 
  (l_extendedprice * (1 - l_discount) * (1 + l_tax)) / rs_rate AS `open_in_euro`
FROM prod_lineitem
JOIN hive.`default`.prod_rates FOR SYSTEM_TIME AS OF l_proctime ON rs_symbol = l_currency
WHERE
  l_linestatus = 'O';

Documentation.

(2) For other (non-JDBC) data sources, you can implement your own enrichment with external services/databases using an async function.

(3) Flink 1.11 adds support for ingesting Debezium CDC (change data capture) streams, making it easier to maintain synchronized views of external databases that are materialized in Flink state. MySQL, PostgreSQL, Oracle, Microsoft SQL Server and many other databases are supported. By mirroring the external data source into Flink, you will get higher throughput and lower latency, and less load on the external database.

user_20008230
Related