使用ApacheSpark和kfaka的实时数据库流

Question

我正在用Kafka设计一个spark流媒体应用程序。我有以下几个问题：我正在将数据从RDBMS表流式传输到kafka，并使用Spark consumer来使用消息，并使用Spark-SQL进行处理

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

这种数据库流的架构和设计是否正常，我如何解决转换问题中的转换？

你好Piyush Kansal

Answer 1

我有一个类似的用例。

我使用Nifi从RDBMS视图中获取数据并放入Kafka Topic。我对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.   
>            }