为什么我的查询因 AnalysisException 而失败？答案

【问题标题】：Why does my query fail with AnalysisException?为什么我的查询因 AnalysisException 而失败？
【发布时间】：2018-10-27 21:21:28
【问题描述】：

我是 Spark 流媒体的新手。我正在尝试使用本地 csv 文件进行结构化 Spark 流式传输。我在处理时遇到以下异常。

Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
FileSource[file:///home/Teju/Desktop/SparkInputFiles/*.csv]

这是我的代码。

val df = spark
  .readStream
  .format("csv")
  .option("header", "false") // Use first line of all files as header
  .option("delimiter", ":") // Specifying the delimiter of the input file
  .schema(inputdata_schema) // Specifying the schema for the input file
  .load("file:///home/Teju/Desktop/SparkInputFiles/*.csv")

val filterop = spark.sql("select tagShortID,Timestamp,ListenerShortID,rootOrgID,subOrgID,first(rssi_weightage(RSSI)) as RSSI_Weight from my_table where RSSI > -127 group by tagShortID,Timestamp,ListenerShortID,rootOrgID,subOrgID order by Timestamp ASC")
val outStream = filterop.writeStream.outputMode("complete").format("console").start()

我创建了 cron 作业，因此每 5 分钟我将获得一个输入 csv 文件。我正在尝试解析 Spark 流。

【问题讨论】：

df与其他数据集filterop和outStream之间有什么关系？您不要在粘贴的代码中使用df。这是故意的吗？我会说代码不能按原样执行。缺少一些重要的东西。

标签： scala apache-spark spark-structured-streaming

【解决方案1】：

（这不是一个解决方案，而是一个评论，但考虑到它的长度，它最终出现在此处。我将在收集到足够的信息进行调查后立即将其作为答案）。

我的猜测是您在df 上做的一些不正确的事情没有包含在您的问题中。

由于错误消息是关于 FileSource 的，路径如下，它是一个流数据集，必须是 df 。

文件源[file:///home/Teju/Desktop/SparkInputFiles/*.csv]

考虑到其他行，我猜您将流数据集注册为临时表（即my_table），然后在spark.sql 中使用它来执行SQL 和writeStream 到控制台.

df.createOrReplaceTempView("my_table")

如果正确，则您在问题中包含的代码不完整，并且没有显示错误原因。

【讨论】：

【解决方案2】：

将.writeStream.start 添加到您的df，正如异常告诉您的那样。

阅读docs了解更多详情。

【讨论】：

.writeStream.start应该在粘贴的代码中添加到哪里？我看不出它如何解决这个问题。需要详细说明吗？
@JacekLaskowski 数据框df 永远不会明确地得到writeStream.started。我怀疑查询必须是“完整的”，没有隐式/带外查询关闭循环。我怀疑 csv 的流式加载永远不会触发，因为输入和输出之间没有链接。我怀疑用df.select 替换spark.sql 可以解决这个问题。