访问 Kafka 源后，Spark 数据帧失去流式处理能力

Question

我使用 Spark 2.4.3 和 Kafka 2.3.0。我想使用从 Kafka 到 Spark 的数据进行 Spark 结构化流式传输。一般来说，它确实在测试模式下工作，但由于我必须对数据进行一些处理（并且不知道另一种方法），Spark 数据帧不再具有流式传输功能。

#!/usr/bin/env python3

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json
from pyspark.sql.types import StructField, StructType, StringType, DoubleType

# create schema for data
schema = StructType([StructField("Signal", StringType()),StructField("Value", DoubleType())])

# create spark session
spark = SparkSession.builder.appName("streamer").getOrCreate()

# create DataFrame representing the stream
dsraw = spark.readStream \
  .format("kafka").option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "test")
print("dsraw.isStreaming: ", dsraw.isStreaming)

# Convert Kafka stream to something readable
ds = dsraw.selectExpr("CAST(value AS STRING)")
print("ds.isStreaming: ", ds.isStreaming)

# Do query on the converted data
dsQuery = ds.writeStream.queryName("ds_query").format("memory").start()
df1 = spark.sql("select * from ds_query")
print("df1.isStreaming: ", df1.isStreaming)

# convert json into spark dataframe cols
df2 = df1.withColumn("value", from_json("value", schema))
print("df2.isStreaming: ", df2.isStreaming)

输出为：

dsraw.isStreaming:  True
ds.isStreaming:  True
df1.isStreaming:  False
df2.isStreaming:  False

所以我在创建第一个数据帧时失去了流式处理能力。我怎样才能避免它？如何从流中获取流式 Spark 数据帧？

Answer 1

不建议将内存接收器用于生产应用程序，因为所有数据都将存储在驱动程序中。

也没有理由这样做，除非出于调试目的，因为您可以像处理 'normal' 数据帧一样处理流式数据帧。例如：

import pyspark.sql.functions as F

lines = spark.readStream.format("socket").option("host", "XXX.XXX.XXX.XXX").option("port", XXXXX).load()

words = lines.select(lines.value)

words = words.filter(words.value.startswith('h'))

wordCounts = words.groupBy("value").count()

wordCounts = wordCounts.withColumn('count', F.col('count') + 2)

query = wordCounts.writeStream.queryName("test").outputMode("complete").format("memory").start()

如果您仍然想采用您的方法：即使 df.isStreaming 告诉您它不是流式数据帧（这是正确的），底层数据源是一个流，因此数据帧将随着每个处理的批次。

访问 Kafka 源后，Spark 数据帧失去流式处理能力

Spark dataframe lose streaming capability after accessing Kafka source

apache-kafka

apache-spark

spark-streaming

apache-spark-sql

pyspark