Почему я получаю следующие журналы при запуске моего кода зажигания через spark-submit

Question

Я запускаю скрипт с помощью этой команды

spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 direct_kafka_wordcount.py localhost 9092 

Я не могу подключиться к своей теме Kafka и получить информацию.Я перепробовал все, но не повезло.Я использую этот простой код wordcount моего живого потока Kafka.

Кэш Ivy Default установлен в: /home/sagar/.ivy2/cache Банки для пакетов, хранящихся в: / home / sagar/.ivy2/jars :: загрузка настроек :: url = jar: file: /usr/local/spark-2.4.3-bin-hadoop2.7/jars/ivy-2.4.0.jar! / org / apache / ivy/core/settings/ivysettings.xml org.apache.spark # spark-streaming-kafka-0-10_2.11 добавлен как зависимость :: разрешение зависимостей :: org.apache.spark # spark-submit-parent-be411cc2-fb3f-4049-b222-e3eca55e020b; конф. 1.0: [по умолчанию] найдено org.apache.spark # spark-streaming-kafka-0-10_2.11; 2.2.0 в центральном найденном org.apache.kafka # kafka_2.11; 0.10.0.1 в центральном найденном com.101tec # zkclient; 0.8 в центральном найденном org.slf4j # slf4j-api; 1.7.16 в центральном найденном org.slf4j # slf4j-log4j12; 1.7.16 в центральном найденном log4j # log4j; 1.2.17 вцентральный найденный com.yammer.metrics # metrics-core; 2.2.0 в центральном найденном org.scala-lang.modules # scala-parser-combinators_2.11; 1.0.4 в центральном найденном org.apache.kafka # kafka-clients;0,10.0.1 в центральном найденном net.jpountz.lz4 # lz4; 1.3.0 в центральном найденном org.xerial.snappy # snappy-java; 1.1.2.6 в центральном найденном org.spark-project.spark # не используется; 1.0.0 в центральном:: отчет о разрешении :: разрешение 1491 мс :: артефакты dl 9 мс :: используемые модули: com.101tec # zkclient; 0.8 от центрального в [по умолчанию] com.yammer.metrics # metrics-core; 2.2.0 из центрального в [по умолчанию]log4j # log4j; 1.2.17 из центрального в [по умолчанию] net.jpountz.lz4 # lz4; 1.3.0 из центрального в [по умолчанию] org.apache.kafka # kafka-клиентов; 0.10.0.1 из центрального в [по умолчанию] org.apache.kafka # kafka_2.11; 0.10.0.1 из центрального в [по умолчанию] org.apache.spark # spark-streaming-kafka-0-10_2.11; 2.2.0 из центрального в [по умолчанию] org.scala-lang.modules # scala-parser-combinators_2.11; 1.0.4 из центрального в [по умолчанию] org.slf4j # slf4j-api; 1.7.16 из центрального в [по умолчанию] org.slf4j # slf4j-log4j12; 1.7.16 из центральногов [по умолчанию] org.spark-project.spark # не используется; 1.0.0 из центрального в [по умолчанию] org.xerial.snappy # snappy-java; 1.1.2.6 из центрального в [по умолчанию] --------------------------------------------------------------------- ||модули ||артефакты ||конф |номер |поиск | dwnlded | выселили ||номер | dwnlded |--------------------------------------------------------------------- |по умолчанию |12 |1 |1 |0 ||12 |0 |--------------------------------------------------------------------- :: retrieving :: org.apache.spark # spark-submit-parent-be411cc2-fb3f-4049-b222-e3eca55e020b confs: [по умолчанию]0 артефактов скопировано, 12 уже получено (0 КБ / 8 мс) 19/07/09 14:28:08 ПРЕДУПРЕЖДЕНИЕ NativeCodeLoader: невозможно загрузить библиотеку native-hadoop для вашей платформы ... с использованием встроенных классов java, где это применимо, Traceback (последний вызовпоследний): файл "/usr/local/spark-2.4.3-bin-hadoop2.7/examples/src/main/python/streaming/direct_kafka_wordcount.py", строка 48, в kvs = KafkaUtils.createDirectStream (ssc, [topic], {"metadata.broker.list": brokers}) Файл "/usr/local/spark/python/lib/pyspark.zip/pyspark/streaming/kafka.py", строка 146, в файле createDirectStream "/ usr/local/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py ", строка 1257, в call File" / usr / local / spark / python / lib /py4j-0.10.7-src.zip/py4j/protocol.py ", строка 328, в get_return_value py4j.protocol.Py4JJavaError: Произошла ошибка при вызове o26.createDirectStreamWithoutMessageHandler.: org.apache.spark.SparkException: посредник не в правильном формате: [localhost] в org.apache.spark.streaming.kafka.KafkaCluster $ SimpleConsumerConfig $$ anonfun $ 7.apply (KafkaCluster.scala: 390) в орг.apache.spark.streaming.kafka.KafkaCluster $ SimpleConsumerConfig $$ anonfun $ 7.apply (KafkaCluster.scala: 387) в scala.collection.TraversableLike $$ anonfun $ map $ 1.apply (TraversableLike.scala: 234) в scala.collection.TraversableLike $$ anonfun $ map $ 1.apply (TraversableLike.scala: 234) в scala.collection.IndexedSeqOptimized $ class.foreach (IndexedSeqOptimized.scala: 33) в scala.collection.mutable.ArrayOps $ ofRef.foreach (ArrayOps.186) в scala.collection.TraversableLike $ class.map (TraversableLike.scala: 234) в scala.collection.mutable.ArrayOps $ ofRef.map (ArrayOps.scala: 186) в org.apache.spark.streaming.kafka.KafkaCluster$ SimpleConsumerConfig. (KafkaCluster.scala: 387) в org.apache.spark.streaming.kafka.KafkaCluster $ SimpleConsumerConfig $ .apply (KafkaCluster.scala: 422) в org.apache.spark.streaming.kafka.KafkaCluster.config (KafkaCluster.scala: 53) в org.apache.spark.streaming.kafka.KafkaCluster.getPartitionMetadata (KafkaCluster.scala: 130) в org.apache.spark.streaming.kafka.KafkaCluster.CafkaCluster.getscala: 119) в org.apache.spark.streaming.kafka.KafkaUtils $ .getFromOffsets (KafkaUtils.scala: 211) в org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper.createDirectStream (KafkaUtils.scala: 720)apache.spark.streaming.kafka.DelegatingMethodAccessorImpl.invoke (DelegatingMethodAccessorImpl.java:43) по адресу java.lang.reflect.Method.invoke (Method.java:498) по адресу py4j.reflection.MethodInvoker.invoke (MethodInvoker.java:24reflection на(ReflectionEngine.java:357) в py4j.Gateway.invoke (Gateway.java:282) в py4j.commands.AbstractCommand.invokeMethod (AbstractCommand.java:132) в py4j.commands.CallCommand.execute (CallCommand.java:79) в py4j.GatewayConnection.run (GatewayConnection.java:238) в java.lang.Thread.run.java: 748)

Answer 1

Неверный синтаксис, попробуйте это ( проверьте часть узла брокера kafka ):

spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 
direct_kafka_wordcount.py localhost:9092

В общих чертах, подключение к серверам начальной загрузки kafka всегда требует синтаксиса host:port.