如何以编程方式将 Kafka 主题加载和流式传输到 PySpark 数据帧

Question

有很多方法可以将 spark 数据帧读/写到 kafka。我正在尝试从 kafka 主题中读取消息并从中创建一个数据框。我能够从主题中获取消息，但无法将其转换为数据名。任何建议都会有所帮助。

import pyspark
from pyspark.sql import SparkSession, Row
from pyspark.context import SparkContext
from kafka import KafkaConsumer

sc = SparkContext.getOrCreate()
spark = SparkSession(sc)

consumer = KafkaConsumer('Jim_Topic')

for message in consumer:
    data = message
    print(data) # Printing the messages properly
    df = data.map # am unable to convert it to a dataframe.

下面的方法我也试过了，

df = spark \
  .read \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "Jim_Topic") \
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

遇到错误，

pyspark.sql.utils.AnalysisException：找不到数据源：kafka。请按照《Structured Streaming + Kafka Integration Guide》的部署部分部署应用。;

Answer 1

根据您的用例，您可以

1. create a Kafka source for streaming queries
2. 或create a Kafka source of batch queries

---

用于流式查询

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "Jim_Topic")
  .load()

# Query data
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
  .as[(String, String)]

---

批量查询

val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "Jim_Topic")
  .load()

# Query data
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
  .as[(String, String)]

---

确保也添加所需的依赖项：

org.apache.spark:spark-sql-kafka-0-10_2.11:2.0.2

（替换为您的 Spark 版本 - 以上指的是 Spark 版本2.0.2）