我有五个执行程序为这个spark作业运行，但来自dstream的RDD只分配在两个分区之间 . 如果我做 repartion(5) 然后它仍然分布在2个分区中，但是当我通过执行 val newrdd= sparkcontext.parallelize(rdd.take(rdd.count())) 创建一个新的RDD时，它会正确地分布在5个分区中 . 但是在并行化现有RDD之后创建一个新的RDD并不是一个好主意，所以我不想这样做 .
我在这里错过了什么吗？
码：
val ssc = new StreamingContext(sparksession.sparkContext, Seconds(batchDuration.toLong))
val inputDirectStream = EventHubsUtils.createDirectStreams(
ssc,
eventHubNamespace,
progressDir,
Map(eventHubName -> eventhubParameters))
inputDirectStream.foreachRDD { rdd =>
println(rdd.partitions.size)//it prints 2
rdd.repartition(5)
println(rdd.partitions.size)//it also prints 2
var newrdd = sparksession.sparkContext.parallelize(rdd.take(rdd.count().toInt))
println(newrdd.partitions.size)//it prints 5
}
我正在运行我的火花流工作如下：
spark-submit --class“com.mycomp.Main”--executor-memory 1g --executor-cores 1 --num-executors 5 --conf“spark.streaming.stopGracefullyOnShutdown = true”--master yarn - jars /tmp/jobs/supporting.jar /tmp/jobs/cdc.jar false> / tmp / jobs / output 2>＆1
有关如何使RDD在5个分区之间分配的任何建议（取决于执行程序和核心的数量） .
1 回答
调用
repartition将返回需要使用的新RDD（已更改分区） . 换句话说，您需要将重新分区调用的返回值分配给新变量，否则您只需使用旧分区处理旧RDD . 更改为val rdd2 = rdd.repartition(5)并在此之后使用rdd2.Note ：由于Scala在设计上是一种惰性语言，因此在对数据执行操作之前不会进行实际的重新分区 . 例如，您可以对数据运行
first或count以使其重新发布 . 但是，仍然可以正确反映使用rdd.partitions.size检查分区 .