无法将Pyspark数据帧发送到Kafka主题

Question

我试图使用pyspark将每日批次的数据发送到Kafka主题，但我当前收到以下错误：

Traceback（最近的最后一次调用）： File"， line 5， in File"/usr/local/rms/lib/hdp26_c5000/park2/python/pyspark/sql/readwriter.py"， line 548， in保存自己。_jwrite.save（）File"/usr/local/rms/lib/hdp26_c5000/park2/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py"， line 1133， in call File"/usr/local/rms/lib/hdp26_c5000/park2/python/pyspark/sql/utils.py"， line 71， in deco引发Analysis Exception（s.split (': ', 1）[1]， stackTrace）pyspark.sql.utils.Analysis Exception： u"对未解析对象上的属性的无效调用，tree： un分辨率vedalias（'shop_id，无）"

我使用的代码如下：

from pyspark.sql import SparkSession
from pyspark.sql import functions

spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config("spark.debug.maxToStringFields", 100000) \
.getOrCreate()

df = spark.sql('''select distinct shop_id, item_id 
from sale.data
''')

df.selectExpr("shop_id", "item_id") \
.write \
.format("kafka") \
.option("kafka.bootstrap.servers", "myserver.local:443") \
.option("topic","test_topic_01") \
.save()

目前使用的版本有：

-火花2.1.1.2.6.2.0-205

-Kafka经纪人0.11

Answer 1

Kafka期望在其主题中写入一个键和值。尽管键不是强制性的。它通过查看应该是“键”和“值”的数据帧列的名称来做到这一点。

在查询中，您只选择了“shop_id”列，因此不存在键或值列。错误消息：“UnsolvedAlias（'shop\u id，None）”告诉您，“shop\u id”列被选为键（因为它是第一列），但没有任何内容被解释为强制值。

您可以通过将列重命名为“value”来解决问题，例如：

df = spark.sql('''select distinct shop_id, item_id 
from sale.data
''')

df.withColumn("value", col("shop_id").cast(StringType)) \
.write \
.format("kafka") \
.option("kafka.bootstrap.servers", "myserver.local:443") \
.option("topic","test_topic_01") \
.save()