apache-spark - 如何根据数据大小重新分区rdd

Question

我正在开发 Spark Streaming 项目，该项目从 Kafka 获取数据并应用一些规则并将数据保存在 Hive 中。我的问题是数据摄取率不固定。 60 秒可能是 100 万条消息到来，也可能是 1 条。我想在 Dstream 上添加重新分区。因为Dstream只有3个分区，无法在一分钟内处理百万条记录。重新分区在少于 20 条记录时会出现问题。它在 Hive 中创建多个小文件。

dataStream.map(_._2).repartition(20)

我的问题是如何根据 rdd 大小对 rdd 进行重新分区。这样它就可以处理一条消息或 100 万条消息。

Answer 1

你无法以任何有用的方式做到这一点。即使您决定使用 transform

 stream.transform { rdd => {
   val n = rdd.count
   rdd.repartition(getNumParttitions(n))
 }}

这超出了整个操作的目的，因为您必须在重新分区和初始分发之前扫描所有数据，这仍然是瓶颈。

相反，我建议根据 spark.streaming.kafka.maxRatePerPartition(旧 API)或配置背压(spark.streaming.backpressure.enabled，spark.streaming.backpressure.initialRate，较新的 API)