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

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

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

除了调试目的之外，也没有理由这样做，因为您可以像处理"普通"数据帧一样处理流数据帧。例如：

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 告诉您它不是流式处理数据帧(这是正确的(，基础数据源也是一个流，因此数据帧将随着每个处理的批处理而增长。