Spark 3.x 在 Python 中与 Kafka 的集成

Posted 2023-03-10

【中文标题】Spark 3.x 在 Python 中与 Kafka 的集成

【英文标题】：Spark 3.x Integration with Kafka in Python

【发布时间】：2020-09-05 13:26:34

【问题描述】：

使用 spark-streaming 的 Kafka 会引发错误：

from pyspark.streaming.kafka import KafkaUtils ImportError: No module named kafka

我已经设置了一个 kafka 代理和一个包含一个 master 和一个 worker 的工作 spark 环境。

import os

os.environ['PYSPARK_PYTHON'] = '/usr/bin/python2.7'
import findspark
findspark.init('/usr/spark/spark-3.0.0-preview2-bin-hadoop2.7')
import pyspark
import sys
from pyspark import SparkConf,SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils



if __name__=="__main__":
    sc = SparkContext(appName="SparkStreamAISfromKAFKA")
    sc.setLogLevel("WARN")
    ssc = StreamingContext(sc,1)
    kvs = KafkaUtils.createStream(ssc,"my-kafka-broker","raw-event-streaming-consumer",'enriched_ais_messages':1)
    lines = kvs.map(lambda x: x[1])
    lines.count().map(lambda x: 'Messages AIS: %s' % x).pprint()
    ssc.start()
    ssc.awaitTermination()

我假设错误是缺少与 kafka ans 相关的内容，特别是版本。有人可以帮忙吗？

spark-version：版本 3.0.0-preview2

我执行：

/usr/spark/spark-3.0.0-preview2-bin-hadoop2.7/bin/spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.1 --jars spark-streaming-kafka-0-10_2.11 spark_streamer.py spark://mysparkip:7077

【参考方案1】：

根据Spark Streaming + Kafka Integration Guide：

“自 Spark 2.3.0 起，Kafka 0.8 支持已弃用。”

此外，下面的屏幕截图显示，Kafka 0.10（及更高版本）不支持 Python。

在您的情况下，您必须使用 Spark 2.4 才能运行您的代码。

PySpark 支持结构化流媒体

如果您打算使用最新版本的 Spark（例如 3.x）并且仍想在 Python 中将 Spark 与 Kafka 集成，您可以使用结构化流式处理。您将在Structured Streaming + Kafka Integration Guide (Kafka broker version 0.10.0 or higher) 中找到有关如何使用 Python API 的详细说明：

从 Kafka 读取数据

# Subscribe to 1 topic
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
  .option("subscribe", "topic1") \
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

向 Kafka 写入数据

# Write key-value data from a DataFrame to a specific Kafka topic specified in an option
ds = df \
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
  .writeStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
  .option("topic", "topic1") \
  .start()