首页 文章

如何将RDD数量的DStream转换为单个RDD

提问于
浏览 次
1

基本上我使用单个Spark Streaming消费者[直接方法]从多个kafka主题消费数据 .

val dStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet).map(_._2)

批处理间隔为 30 Seconds .

我在这里有几个问题 .

  • 当我在DStream上调用foreachRDD时，DStream是否包含多个RDD而不是单个RDD？每个主题会创建单独的RDD ??

  • 如果是，我想将所有RDD联合到单个RDD，然后处理数据 . 我怎么做？

  • 如果我的处理时间超过批处理间隔，DStream是否会包含多个RDD？

我尝试使用以下方法将DStream RDD联合到单个RDD . 首先是我的理解正确吗？如果DStream总是返回单个RDD，则不需要以下代码 .

Sample Code:

var dStreamRDDList = new ListBuffer[RDD[String]]
dStream.foreachRDD(rdd =>
        {
            dStreamRDDList += rdd
        })
val joinedRDD = ssc.sparkContext.union(dStreamRDDList).cache()

//THEN PROCESS USING joinedRDD
//Convert joinedRDD to DF, then apply aggregate operations using DF API.
scala spark-streaming kafka-consumer-api

1 回答

  • 1

    当我在DStream上调用foreachRDD时，DStream是否包含多个RDD而不是单个RDD？每个主题会创建单独的RDD吗？

    不会 . 即使您有多个主题，您也可以在任何给定的批处理间隔内拥有一个RDD .

    如果我的处理时间超过批处理间隔，DStream是否会包含多个RDD？

    不，如果您的处理时间比批处理间隔长，那么所有要做的就是读取主题偏移量 . 只有在上一个作业完成后才开始处理下一批 .

    作为附注，请确保您确实需要使用 foreachRDD ，或者如果可能you're misusing the DStream API（免责声明：我是该帖子的作者）

    回复于

相关问题