Spark структурированный поток исполнителей странное поведение - PullRequest
Новая
       36

Spark структурированный поток исполнителей странное поведение

0 голосов
/ 27 апреля 2018

Использование структурированного потока Spark с решением Cloudera Я использую 3 исполнителя, но когда я запускаю приложение, используемый исполнитель только один. Как я могу использовать несколько исполнителей?

Позвольте мне дать вам больше информации. Это мои параметры:

Запуск команды: 

spark2-submit --master yarn \
--deploy-mode cluster \
--conf spark.ui.port=4042 \
--conf spark.eventLog.enabled=false \
--conf spark.dynamicAllocation.enabled=false \
--conf spark.streaming.backpressure.enabled=true \
--conf spark.streaming.kafka.consumer.poll.ms=512 \
--num-executors 3 \
--executor-cores 3 \
--executor-memory 2g \
--jars /data/test/spark-avro_2.11-3.2.0.jar,/data/test/spark-streaming-kafka-0-10_2.11-2.1.0.cloudera1.jar,/data/test/spark-sql-kafka-0-10_2.11-2.1.0.cloudera1.jar \
--class com.test.Hello /data/test/Hello.jar

Код: 

val lines = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", <topic_list:9092>)
      .option("subscribe", <topic_name>)
      .option("group.id", <consumer_group_id>)
      .load()
      .select($"value".as[Array[Byte]], $"timestamp")
      .map((c) => { .... })

val query = lines
      .writeStream
      .format("csv")
      .option("path", <outputPath>)
      .option("checkpointLocation", <checkpointLocationPath>)
      .start()
query.awaitTermination()

Результат в SparkUI: Изображение SparkUI

То, что я ожидал, что все исполнители работали.

Есть предложения?

Спасибо Paolo

1 Ответ

0 голосов
/ 28 апреля 2018

Похоже, в вашей конфигурации нет ничего плохого, просто используемые вами разделы могут быть только одним. Вам нужно увеличить разделы у вашего производителя кафки. Обычно количество разделов примерно в 3-4 раза превышает количество исполнителей.

Если вы не хотите прикасаться к коду производителя, вы можете обойти это, выполнив repartition (3) перед применением метода map, поэтому каждый исполнитель работает на своем собственном логическом разделе.

Если вы все еще хотите, чтобы вы явно упоминали работу, которую получает каждый исполнитель, вы можете воспользоваться методом mapPerPartion.

...