У меня есть потоковое приложение, которое выглядит так:

val message = KafkaUtils.createStream(...).map(_._2)

message.foreachRDD( rdd => {

  if (!rdd.isEmpty){
    val kafkaDF = sqlContext.read.json(rdd)

    kafkaDF.foreachPartition(
      i =>{
        createConnection()
        i.foreach(
          row =>{
            connection.sendToTable()
          }
        )
        closeConnection()
      }
    )

И я запускаю его на кластере пряжи с помощью

spark-submit --master yarn-cluster --num-executors 3 --driver-memory 2g --executor-memory 2g --executor-cores 5....

Когда я пытаюсь войтиkafkaDF.rdd.partitions.sizeв результате получается в основном «1» или «5». Я запутался, возможно ли контролировать количество разделов моего DataFrame?KafkaUtils.createStream похоже, не принимает никаких параметров, связанных с количеством разделов, которые я хочу для rdd. Я старалсяkafkaDF.rdd.repartition( int ), но, похоже, тоже не работает.

Как я могу добиться большего параллелизма в моем коде? Если мой подход неверен, как правильно его достичь?