Работает ли spark-streaming-kafka-0-10_2.10 с Python? - PullRequest
Новая
       43

Работает ли spark-streaming-kafka-0-10_2.10 с Python?

0 голосов
/ 10 июля 2019

Я не смог найти документацию, рассказывающую, как использовать spark-streaming-kafka-0-10_2.10 с Python для интеграции Kafka в качестве источника ввода для Spark (https://spark.apache.org/docs/latest/streaming-kafka-integration.html). Не поддерживается Python?

Заранее спасибо.

1 Ответ

0 голосов
/ 10 июля 2019

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

Пожалуйста, взгляните на

  1. Документация pyspark
  2. интеграция с потоковым потоком / kafka
  3. как развернуть для python (Kafka 0.10)

Пример добавления JAR в сеанс PySpark 

    from pyspark.sql import SparkSession

    spark = SparkSession.builder.appName('test') \
        .config('spark.jars.packages', 'org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.0') \
        .getOrCreate()

Тогда как обычно 

    import random

    from pyspark import SparkContext
    from pyspark.streaming import StreamingContext
    from pyspark.streaming.kafka import KafkaUtils

    sc = SparkContext(appName='testIntegration')
    ssc = StreamingContext(sc, 2)

    topic = "topic-%d" % random.randint(0, 10000)
    brokers = {"metadata.broker.list": "123.43.54.231:9092,123.43.54.235:9092,123.43.54.239:9092"}
    stream = KafkaUtils.createDirectStream(ssc, [topic], brokers)

    ...

    ssc.start()
    ssc.awaitTermination()
...