将 Pyspark 与 Kafka 连接起来

Question

我在理解如何连接 Kafka 和 PySpark 时遇到问题。

我在 Windows 10 上安装了 kafka，主题很好地流式传输数据。 我已经安装了运行正常的 pyspark——我能够毫无问题地创建测试 DataFrame。

但是当我尝试连接到 Kafka 流时，它给了我错误：

AnalysisException：找不到数据源：kafka。请部署 根据“结构化流式传输”的部署部分应用程序- Kafka 集成指南”。

Spark 文档并没有真正的帮助 - 它说： ... groupId = org.apache.spark artifactId = spark-sql-kafka-0-10_2.12 版本 = 3.2.0 ...

对于 Python 应用程序，您需要在部署应用程序时添加上述库及其依赖项。请参阅下面的部署小节。

然后当您转到部署部分时，它会说：

与任何 Spark 应用程序一样，spark-submit 用于启动您的应用程序。 spark-sql-kafka-0-10_2.12 及其依赖可以直接使用 --packages 添加到 spark-submit 中，例如， ./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0 ...

我正在开发应用程序，我不想部署它。 如果我正在开发 pyspark 应用程序，在哪里以及如何添加这些依赖项？

尝试了几个教程最终变得更加困惑。

看到回答说

"您需要将 kafka-clients JAR 添加到您的 --packages".so-answer

更多的步骤可能会有用，因为对于新手来说这是不清楚的。

版本：

• 卡夫卡2.13-2.8.1
• 火花3.1.2
• java 11.0.12

所有环境变量和路径都设置正确。

编辑

我已经加载了：

   os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2,org.apache.kafka:kafka-clients:2.8.1'

按照建议，但仍然出现相同的错误。 我已经三次检查了 kafka、scala 和 spark 版本并尝试了各种组合，但没有成功，我仍然遇到同样的错误：

AnalysisException：找不到数据源：kafka。请部署 按照“Structured Streaming-Kafka Integration Guide”的部署部分应用程序。

编辑 2

我安装了最新的 Spark 3.2.0 和 Hadoop 3.3.1 和 kafka 版本 kafka_2.12-2.8.1。更改了所有环境变量，测试了 Spark 和 Kafka - 工作正常。

我的环境变量现在看起来像这样：

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0,org.apache.kafka:kafka-clients:2.8.1'

仍然没有运气，我得到同样的错误:(

Answer 1

Spark 文档并没有真正的帮助 - 它说 ... artifactId = spark-sql-kafka-0-10_2.12 version = 3.2.0 ...

是的，这是正确的......但对于最新版本的 Spark

版本：

• 火花3.1.2

你试过查看version specific docs吗？

换句话说，您需要匹配的spark-sql-kafka 3.1.2 版本。

bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2

或者在 Python 中，

import os

spark_version = '3.1.2'
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:{}'.format(spark_version)

# init spark here

需要在上面添加这个库及其依赖

正如您在我之前的回答中找到的，还使用逗号分隔列表附加 kafka-clients 包。

--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2,org.apache.kafka:kafka-clients:2.8.1

---

我正在开发应用程序，我不想部署它。

“部署”是 Spark 术语。本地运行还是“部署”