无法重新分区 DStream

Question

val sparkConf = new SparkConf().setMaster("yarn-cluster")
                               .setAppName("SparkJob")
                               .set("spark.executor.memory","2G")
                               .set("spark.dynamicAllocation.executorIdleTimeout","5")

val streamingContext = new StreamingContext(sparkConf, Minutes(1))

var historyRdd: RDD[(String, ArrayList[String])] = streamingContext.sparkContext.emptyRDD

var historyRdd_2: RDD[(String, ArrayList[String])] = streamingContext.sparkContext.emptyRDD

val stream_1 = KafkaUtils.createDirectStream[String, GenericData.Record, StringDecoder, GenericDataRecordDecoder](streamingContext, kafkaParams ,  Set(inputTopic_1))
val dstream_2 = KafkaUtils.createDirectStream[String, GenericData.Record, StringDecoder, GenericDataRecordDecoder](streamingContext, kafkaParams ,  Set(inputTopic_2))

val dstream_2 = stream_2.map((r: Tuple2[String, GenericData.Record]) => 
{
    //some mapping
}

dstream_1.foreachRDD(r => r.repartition(500))
val historyDStream = dstream_1.transform(rdd => rdd.union(historyRdd))
dstream_2.foreachRDD(r => r.repartition(500))
val historyDStream_2 = dstream_2.transform(rdd => rdd.union(historyRdd_2))
val fullJoinResult = historyDStream.fullOuterJoin(historyDStream_2)

val filtered = fullJoinResult.filter(r => r._2._1.isEmpty)


filtered.foreachRDD{rdd =>
    val formatted = rdd.map(r  => (r._1 , r._2._2.get)) 
    historyRdd_2.unpersist(false) // unpersist the 'old' history RDD
    historyRdd_2 = formatted // assign the new history
    historyRdd_2.persist(StorageLevel.MEMORY_AND_DISK) // cache the computation
}

val filteredStream = fullJoinResult.filter(r => r._2._2.isEmpty)

filteredStream.foreachRDD{rdd =>
    val formatted = rdd.map(r => (r._1 , r._2._1.get)) 
    historyRdd.unpersist(false) // unpersist the 'old' history RDD
    historyRdd = formatted // assign the new history
    historyRdd.persist(StorageLevel.MEMORY_AND_DISK) // cache the computation
}

streamingContext.start()
streamingContext.awaitTermination()
}
}

我无法使用上面的代码对 DStream 进行重新分区，我的输入得到了 128 个分区，这是没有的。 Kafka 分区，并且由于 Join 我需要随机读取和写入数据，所以我想通过增加分区数来增加并行度。但是分区保持不变。为什么会这样？

Answer 1

就像map 或filter 一样，repartition 是 Spark 中的转换，意味着 3 件事：

• 它返回另一个不可变的 RDD
• 很懒
• 它需要通过一些动作来实现

考虑这段代码：

dstream_1.foreachRDD(r => r.repartition(500))

在foreachRDD 中使用repartition 作为副作用没有任何作用。生成的 RDD 永远不会被使用，因此重新分区永远不会发生。

我们应该将这种转换与作业中的其他操作“链接”起来。在这种情况下，实现此目的的一种简单方法是改用transform：

val repartitionedDStream = dstream_1.transform(rdd => rdd.repartition(500))
... use repartitionedDStream further on ...