我将抄表记录以 JSON 形式从 kafka_2.11-0.10.0.1 流式传输到 Spark 2.1 中。我改用结构化流媒体;尽管kafka消费者确认传入数据,但我控制台和writeStream不动。我正在测试使用
pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
My code:
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSession \
.builder \
.appName("interval") \
.master("local[4]") \
.getOrCreate()
schema = StructType().add("customer_id", StringType())
df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "xx.xxx.xx.xxx:9092") \
.option("subscribe", "test") \
.option("startingOffsets", "earliest") \
.load() \
.select(from_json(col("value").cast("string"), schema).alias("parsed_value"))
query = df.writeStream \
.option("checkpointLocation", "/user/XX/checkpoint5") \
.format("parquet") \
.start("/user/XX/interval5")
它使用 388 字节 parquet 文件创建检查点和数据目录。然而,没有写入任何流数据。
$ hdfs dfs -ls interval5
drwxr-xr-x ... interval5/_spark_metadata
-rw-r--r-- ... interval5/part-00000-0b2eb00a-c361-4dfe-a24e-9589d150a911.snappy.parquet
-rw-r--r-- ... interval5/part-00000-e0cb12d1-9c29-4eb0-92a8-688f468a42ce.snappy.parquet
kafka-consumer 确认数据正在发送:
{"customer_id":"customer_736"}
{"customer_id":"customer_995"}
{"customer_id":"customer_1899"}
{"customer_id":"customer_35"}
kafka-consumer 显示流数据。
我认为我错过了出列和保存流式传输行的重要步骤 - 一天的 stackoverflow 搜索并没有帮助。
(编辑以删除对控制台的引用;因为它不相关)。