使用来自 pyspark 2.4.4 的火花流

Question

我在 k8s 容器中设置了 spark 2.4.4 版本。我正在尝试为使用这样的火花流编写一个简单的 hello world：

from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
spark = SparkSession.builder.appName("pyspark-kafka").getOrCreate()
sc.setLogLevel("WARN")

ssc = StreamingContext(sc, 60)
kafkaStream = KafkaUtils.createDirectStream(ssc, ['users-update'], {"metadata.broker.list":'pubsub-0.pubsub:9092,pubsub-1.pubsub:9092,pubsub-2.pubsub:9092'})

请注意，pubsub-x.pubsub 是我的容器可见的 kafka 代理。 （在我的最后一行 pyspark 代码中，一个简单的 python 程序直接使用带有代理和主题的 kafka-python 客户端工作得很好。）

我收到此错误消息：

________________________________________________________________________________________________

  Spark Streaming's Kafka libraries not found in class path. Try one of the following.

  1. Include the Kafka library and its dependencies with in the
     spark-submit command as

     $ bin/spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8:2.4.4 ...

  2. Download the JAR of the artifact from Maven Central http://search.maven.org/,
     Group Id = org.apache.spark, Artifact Id = spark-streaming-kafka-0-8-assembly, Version = 2.4.4.
     Then, include the jar in the spark-submit command as

     $ bin/spark-submit --jars <spark-streaming-kafka-0-8-assembly.jar> ...

________________________________________________________________________________________________

maven 上的任何地方都没有 2.4.4 版本的 kafka 库。 https://search.maven.org/search?q=spark%20kafka 显示最后发布的 jar 是 2.10 或 2.11 版本。

我的 pyspark 安装中确实有一个 spark-streaming_2.12-2.4.4.jar jar，但它似乎没有正确的 kafka 类。

感谢您的任何指点！ --斯里达尔

Answer 1

Spark v2.4.4 是使用 scala v2.11 预构建的。从火花下载页面：

请注意，Spark 是使用 Scala 2.11 预构建的，但版本 2.4.2 是使用 Scala 2.12 预构建的。

所以，基本上2.10 和 2.11 是构建 spark 的 scala 版本，你应该下载 spark-streaming-kafka jar，它是在你的情况下使用相同版本的 scala 构建的 2.11。

我已经检查了 spark 2.4.4 中的 jars 文件夹，并且那里存在 spark-streaming_2.11-2.4.4.jar jar。因此，如果您已将 spark-streaming_2.12-2.4.4.jar 从外部添加到类路径中，则应删除它，否则您将得到版本不匹配。

您可以从here 下载spark-streaming-kafka-0-8-assembly.jar 而且我认为您还需要从here 添加kafka-clients jar。

Answer 2

我的 pyspark 安装中确实有一个 spark-streaming_2.12-2.4.4.jar jar，但它似乎没有正确的 kafka 类。

这只是 Spark 的基本 Streaming 包。 Spark 不附带 Kafka 类

Spark Streaming 已被弃用，取而代之的是 Spark Structured Streaming

你想要这个包用于带有 Scala 2.12 的 Spark

'org.apache.spark:spark-sql-kafka-0-10_2.12:2.4.4'

你会像这样开始，包括引导服务器的选项

df = spark.readStream().format("kafka")