2

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

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

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

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

Spark 文档并没有真正的帮助 - 它说： ... groupId = org.apache.spark artifactId = spark-sql-kafka-0-10_2.12 version = 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”。如此回答

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

版本：

  • 卡夫卡 2.13-2.8.1
  • 火花3.1.2
  • 爪哇 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 集成指南”的部署部分部署应用程序。

编辑 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'

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

4

1 回答 1

4

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

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

版本：

  • 火花3.1.2

您是否尝试过查看特定于版本的文档？

换句话说，您需要spark-sql-kafka3.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 术语。在本地运行仍然是一个“部署”

于 2021-12-16T15:55:34.837 回答