一个流中几个不同字段的平均值答案

【问题标题】：Average of several different fields from one stream一个流中几个不同字段的平均值
【发布时间】：2019-04-16 00:19:43
【问题描述】：

我还没有选择流媒体框架，但我现在正在搞乱 Flink。但是，我愿意使用 Beam、Spark Streaming，只要我发现适合我的用例。您将如何执行以下 SQL 的等效操作：

SELECT a,b,c, avg(d), avg(e), ..., avg(z)
FROM whatever
GROUP BY a,b,c,d,e, ..., z

似乎对于 Flink 来说，平均是通过 AggregateFunction https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/functions/AggregateFunction.java#L61 完成的

但我不明白你是如何制作这个“规模”的。对于单个字段的平均值，这似乎是很多样板。如果我需要对多个不同字段的不同流进行平均怎么办？

Flink、Beam、Structured Streaming 等是否让这变得更容易？

顺便说一句，有没有一种简单的方法可以从 Postgres 中模拟这种漂亮的小计数过滤器语法，

SELECT
  COUNT(*) AS unfiltered,
  COUNT(*) FILTER (WHERE some_condition) AS filtered
FROM whatever

【问题讨论】：

标签： apache-spark-sql spark-streaming apache-flink apache-beam flink-streaming

【解决方案1】：

通常在 flink 作业中，我会将已定义的用户函数创建为单独的类，然后我可以将其应用于我喜欢的任何领域。 Flink 也有一个 SQL API，我不太熟悉，但这里有一个基于我在此处找到的代码的示例 (https://gist.github.com/mustafaakin/457859b8bf703c64029071c1139b593d)：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamTableEnvironment table = TableEnvironment.getTableEnvironment(env);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<String> text = env.socketTextStream("localhost", 9999, "\n");
DataStream<Tuple3<String, Double, Time>> dataset = text.map(...);

table.registerDataStream("dataset", dataset, "p1, p2, p3");
String query = "SELECT p1, AVG(p2) AS avgp2 FROM dataset GROUP p1";
Table tableResult = table.sql(query);

// print to System.out
table.toAppendStream(tableResult, Row.class).print();

env.execute();

我还会使用 Apache Ignite 来查看带有 SQL 查询的流式数据。我自己从来没有用过，但我听说过好东西。

【讨论】：