val tuple = in.reduce((x, y) => (x._1, x._2 + y._2))
out.collect(tuple)
}
// 打印测试
result.print()
复制代码
minBy和maxBy
选择具有最小值或最大值的元素
// 使用minBy操作，求List中每个人的最小值
// List("张三,1", "李四,2", "王五,3", "张三,4")
case class User(name: String, id: String)
// 将List转换为一个scala的样例类
val text: DataSet[User] = textDataSet.mapPartition(line => {
line.map(index => User(index._1, index._2))
})
val result = text
.groupBy(0) // 按照姓名分组
.minBy(1) // 每个人的最小值
复制代码
Aggregate
在数据集上进行聚合求最值（最大值、最小值）
val data = new mutable.MutableList[(Int, String, Double)]
data.+=((1, "yuwen", 89.0))
data.+=((2, "shuxue", 92.2))
data.+=((3, "yuwen", 89.99))
// 使用 fromElements 构建数据源
val input: DataSet[(Int, String, Double)] = env.fromCollection(data)
// 使用group执行分组操作
val value = input.groupBy(1)
// 使用aggregate求最大值元素
.aggregate(Aggregations.MAX, 2)
// 打印测试
value.print()
复制代码
Aggregate只能作用于元组上
留意：
要使用aggregate，只能使用字段索引名或索引名称来进行分组 groupBy(0) ，否则会报一下错误:
Exception in thread “main” java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.
distinct
去除重复的数据
// 数据源使用上一题的
// 使用distinct操作，根据科目去除集合中重复的元组数据
val value: DataSet[(Int, String, Double)] = input.distinct(1)
//val myProducer = new FlinkKafkaProducer011<>(brokerList, topic, new SimpleStringSchema());
//使用支持仅一次语义的形式
val myProducer =
new FlinkKafkaProducer011[String](topic, new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema), prop, FlinkKafkaProducer011.Semantic.EXACTLY_ONCE);
SELECT：
SELECT 用于从 DataSet/DataStream 中选择数据，用于筛选出某些列。
示例：
SELECT * FROM Table; // 取出表中的所有列
SELECT name，age FROM Table; // 取出表中 name 和 age 两列
与此同时 SELECT 语句中可以使用函数和别名，例如我们上面提到的 WordCount 中：
SELECT word, COUNT(word) FROM table GROUP BY word;
WHERE：
WHERE 用于从数据集/流中过滤数据，与 SELECT 一起使用，用于根据某些条件对关系做水平分割，即选择符合条件的记载。
示例：
SELECT name，age FROM Table where name LIKE ‘% 小明 %’;
SELECT * FROM Table WHERE age = 20;
WHERE 是从原数据中进行过滤，那么在 WHERE 条件中，Flink SQL 同样支持 =、<、>、<>、>=、<=，以及 AND、OR 等表达式的组合，终极满足过滤条件的数据会被选择出来。而且 WHERE 可以联合 IN、NOT IN 联合使用。举个例子：
SELECT name, age
FROM Table
WHERE name IN (SELECT name FROM Table2)
复制代码
DISTINCT：
DISTINCT 用于从数据集/流中去重根据 SELECT 的结果进行去重。
示例：
SELECT DISTINCT name FROM Table;
对于流式查询，计算查询结果所需的 State 可能会无限增长，用户需要自己控制查询的状态范围，以防止状态过大。
GROUP BY：
GROUP BY 是对数据进行分组操作。例如我们需要计算成绩明细表中，每个门生的总分。
示例：
SELECT name, SUM(score) as TotalScore FROM Table GROUP BY name;
UNION 和 UNION ALL：
UNION 用于将两个结果聚集并起来，要求两个结果集字段完全同等，包括字段范例、字段序次。不同于 UNION ALL 的是，UNION 会对结果数据去重。
示例：
SELECT * FROM T1 UNION (ALL) SELECT * FROM T2;
JOIN：
JOIN 用于把来自两个表的数
