例外：“writeStream”只能在流数据集/数据帧上调用

Question

尝试为spark数据流的writeStream函数创建一个测试，如下所示：

SparkSession spark = SparkSession.builder().master("local").appName("spark 
session").getOrCreate()

val lakeDF = spark.createDF(List(("hi")), List(("word", StringType, true)))

lakeDF.writeStream
  .trigger(Trigger.Once)
  .format("parquet")
  .option("checkpointLocation", checkpointPath)
  .start(dataPath)

但我收到以下异常： org.apache.spark.sql.AnalysisException: 'writeStream' 只能在流数据集/DataFrame 上调用；

我是 Spark 流式传输的新手，请告诉我如何创建流式传输数据帧/将上述常规数据帧转换为我的测试套件的流式传输数据帧。

Answer 1

在 Spark 结构化流中，数据帧/数据集是使用 SparkSession 上的 readStream 创建的。如果数据帧/数据集不是使用流创建的，则不允许使用 writeStream 进行存储。

所以使用 readStream 创建数据帧/数据集并使用 writeStream 存储数据帧/数据集

 val kafkaStream = sparkSession.readStream.format("kafka")
.option("kafka.bootstrap.servers", "kafka-broker-hostname:port")
.option("subscribe", "topicname")
.load()

Answer 2

考虑以下示例。

第一个适用于非流式传输的 df。喜欢你的，例如由列表/序列组成的文件或 df。

// Write key-value data from a DataFrame to a specific Kafka topic specified in an option
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .write
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "topic1")
  .save()

第二个适用于流式传输的数据帧。例如从 kafka 或其他流媒体源读取。

// Write key-value data from a DataFrame to a specific Kafka topic specified in an option
val ds = df
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "topic1")
  .start()

要么使用流式数据帧，要么像第一个示例一样更改代码。