使用 Apache Spark 和 kafka 进行实时数据库流式传输

Question

我正在使用 Kafka 设计一个 spark 流应用程序。我有几个问题如下： 我正在将 RDBMS 表中的数据流式传输到 kafka，并使用 Spark 消费者使用 Spark - SQL 来消费消息和处理

问题： 1. 我将数据从表中流式传输到 kafka（键作为表名，值作为 JSON 记录形式的表数据）——这是正确的架构吗？

1. 在 spark 消费者中，我正在尝试使用 DStream.foreachRDD(x => 转换为 x RDD) 来使用数据——我遇到了这个问题（它说转换中的转换错误不允许......我正在尝试在 foreachRDD 函数中提取键以获取表名并使用 map 函数转换 x.values 以从 JSON 转换回普通字符串，然后将每条记录保存到 Spark-sql )

这种用于数据库流式传输的架构和设计是否可行？如何解决转换问题中的转换？

问候， 皮尤什·坎萨尔

Answer 1

我有一个类似的用例。

我使用 Nifi 从 RDBMS 视图中获取数据并放入 Kafka 主题。 我为具有多个分区的 Oracle 数据库中的每个视图都有一个主题。 使用 Nifi 将数据转换为 JSON 格式并放入 Kafka。

是否有任何要求对所有表数据使用相同的 kafka 主题？

以下代码将用于将数据持久保存到 Cassandra。

> val msg = KafkaUtils.createDirectStream[String, String, StringDecoder,
> StringDecoder](ssc, kafkaParams, topicsSet)
>   
>     /* Process Records for each RDD */   Holder.log.info("Spark foreach Starts")
>        val data = msg.map(_._2) 
>        data.foreachRDD(rdd =>{
>        if(rdd.toLocalIterator.nonEmpty)    {
>     
>     
>       val messageDfRdd = sqlContext.read.json(rdd)
var data2=messageDfRdd .map(p => employee(p.getLong(1),p.getString(4),p.getString(0),p.getString(2),p.getString(3),p.getString(5)));

>  //code to save to Cassandra.   
>            }