将 3 列的 CSV 文件读入 Datastream。 JAVA Apache Flink答案

【问题标题】：Read CSV file with 3 columns into Datastream. JAVA Apache Flink将 3 列的 CSV 文件读入 Datastream。 JAVA Apache Flink
【发布时间】：2020-02-11 05:21:57
【问题描述】：

我一直在努力设置一个从 csv 文件创建 Datastream<Tuple3<Integer, java.sql.Time, Double>> 的 flink 应用程序。此文件中的列（列ID, dateTime and Result）都是字符串，但它们应转换为整数、java.sql.Time 和 Double。我想要的另一件事是使用每天的数据创建翻滚窗口并平均该窗口中result 列的值。问题是我不知道它的确切语法。请参阅我尝试过的代码。最后一部分我有 sum(2)，但我想计算窗口的平均值。我没有在文档中看到此功能。我需要自己为此编写一个方法吗？


DataStream<Tuple3<String, java.sql.Time>> dataStream = env
                .readfile(path)
                .map()
                .keyBy(0)
                .timeWindow(Time.days(1));

【问题讨论】：

标签： java apache-flink data-stream streaming-analytics

【解决方案1】：

您可以使用自己的逻辑来读取 csv 或使用诸如 univocity_parsers 之类的库。除了使用 env.readfile 之外，您还可以使用 env.fromCollection(list)。

这是图书馆的链接如果你想要： https://www.univocity.com/pages/univocity_parsers_tutorial#using-annotations-to-map-your-java-beans

您可以使用 anotaion @Convert(conversionClass = YourDataTimeCoverter.class)

给自己的转换器

一般情况下，请参考以下 flink 文档示例：

https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/windows.html#aggregatefunction

【讨论】：