java – 使用带水印的附加输出模式时的结构化流异常

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了java – 使用带水印的附加输出模式时的结构化流异常，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含1494字，纯文字阅读大概需要3分钟。

内容图文

尽管我正在使用Watermark(),但是当我运行我的spark工作时,我收到以下错误消息：

Exception in thread “main” org.apache.spark.sql.AnalysisException: Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark;;

从我在programming guide中看到的内容,这与预期用法(和示例代码)完全匹配.有谁知道什么可能是错的？

提前致谢！

相关代码(Java 8,Spark 2.2.0)：

StructType logSchema = new StructType()
        .add("timestamp", TimestampType)
        .add("key", IntegerType)
        .add("val", IntegerType);

Dataset<Row> kafka = spark
        .readStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", brokers)
        .option("subscribe", topics)
        .load();

Dataset<Row> parsed = kafka
        .select(from_json(col("value").cast("string"), logSchema).alias("parsed_value"))
        .select("parsed_value.*");

Dataset<Row> tenSecondCounts = parsed
        .withWatermark("timestamp", "10 minutes")
        .groupBy(
            parsed.col("key"),
            window(parsed.col("timestamp"), "1 day"))
        .count();

StreamingQuery query = tenSecondCounts
        .writeStream()
        .trigger(Trigger.ProcessingTime("10 seconds"))
        .outputMode("append")
        .format("console")
        .option("truncate", false)
        .start();

解决方法:

问题出在parsed.col中.用col替换它将解决问题.我建议总是使用col函数而不是Dataset.col.

当col返回未解析的列时,Dataset.col返回已解析的列.

parsed.withWatermark(“timestamp”,“10 minutes”)将创建一个新数据集,其中包含具有相同名称的新列.水印信息附加在新数据集中的timestamp列,而不是parsed.col(“timestamp”),因此groupBy中的列没有水印.

当您使用未解析的列时,Spark会为您找出正确的列.

内容总结

以上是互联网集市为您收集整理的java – 使用带水印的附加输出模式时的结构化流异常全部内容，希望文章能够帮你解决java – 使用带水印的附加输出模式时的结构化流异常所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端