Spark 结构化流式传输 -答案

【问题标题】：Spark Structured Streaming -Spark 结构化流式传输 -
【发布时间】：2017-08-02 17:13:14
【问题描述】：

我正在尝试从 IntelliJ idea 运行以下代码，以将消息从 Kafka 打印到控制台。但它会引发以下错误 -

Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;

Stacktrace 从Dataset.checkpoint 开始，一直向上。如果我删除.checkpoint()，那么我会收到一些其他错误 - 与权限相关

17/08/02 12:10:52 ERROR StreamMetadata: Error writing stream metadata StreamMetadata(4e612f22-efff-4c9a-a47a-a36eb533e9d6) to C:/Users/rp/AppData/Local/Temp/temporary-2f570b97-ad16-4f00-8356-d43ccb7660db/metadata
java.io.IOException: (null) entry in command string: null chmod 0644 C:\Users\rp\AppData\Local\Temp\temporary-2f570b97-ad16-4f00-8356-d43ccb7660db\metadata

来源：

def main(args : Array[String]) = {
 val spark = SparkSession.builder().appName("SparkStreaming").master("local[*]").getOrCreate()
  val canonicalSchema = new StructType()
                          .add("cid",StringType)
                          .add("uid",StringType)
                          .add("sourceSystem",
                              new StructType().add("id",StringType)
                                              .add("name",StringType))
                          .add("name", new StructType()
                                        .add("firstname",StringType)
                                        .add("lastname",StringType))


val messages = spark
                    .readStream
                    .format("kafka")
                    .option("kafka.bootstrap.servers","localhost:9092")
                    .option("subscribe","c_canonical")
                    .option("startingOffset","earliest")
                    .load()
                    .checkpoint()
.select(from_json(col("value").cast("string"),canonicalSchema))
.writeStream.outputMode("append").format("console").start.awaitTermination

 }

谁能帮我理解我做错了什么？

【问题讨论】：

尝试以管理员身份运行 IntelliJ。
感谢您的回复，但没有奏效。

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

【解决方案1】：

结构化流不支持Dataset.checkpoint()。有一张公开票可以提供更好的信息或忽略它：https://issues.apache.org/jira/browse/SPARK-20927
IOException 可能是因为你没有在 Windows 上安装 cygwin。

【讨论】：

虽然结构化流不支持checkpoint()，但它支持option("checkpointLocation", "/path/to/store")。你说这叫什么？
更新了答案。不幸的是，它们使用相同的词，但它们是完全不同的东西。
我在哪里可以找到checkpointLocation 的实际含义以及它有何不同？
"checkpointLocation" 是结构化流存储其元数据和运行状态的位置，以便它可以从故障中重新启动。 “Dataset.checkpoint”用于非流式Dataset，它只是将Dataset的当前内容存储到文件系统中（由SparkContext.setCheckpointDir设置），因此不需要存储整个大型逻辑计划。它不支持恢复。
这绝对应该记录在案。