Flink四种选择Key的方法.PDF

Flink四种选择Key的方法 Spark大数据博客 - Flink四种选择Key的方法 在Flink中有许多函数需要我们为其指定key，比如groupBy，Join中的where等。如果我们指定的K ey不对，可能会出现一些问题，正如下面的程序： package com.iteblog.flink import org.apache.flink.api.scala.{ExecutionEnvironment, _} import org.apache.flink.util.Collector ///////////////////////////////////////////////////////////////////// User: 过往记忆 Date: 2017-03-13 Time: 22:59 bolg: 本文地址：/archives/2069 过往记忆博客，专注于hadoop、hive、spark、shark、flume的技术博客，大量的干货 过往记忆博客微信公共帐号：iteblog_hadoop ///////////////////////////////////////////////////////////////////// object GroupCombine { def main(args: Array[String]): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment val input: DataSet[String] = env.fromElements(a, b, c, a) val combinedWords: DataSet[(String, Int)] = input .groupBy(0) .combineGroup { (words, out: Collector[(String, Int)]) = var key: String = null var count = 0 for (word - words) { key = word count += 1 } out.collect((key, count)) } combinedWords.print() } } 1 / 5 Flink四种选择Key的方法 Spark大数据博客 - 上面的代码没有任何语法错误，但是我们编译运行这个程序，就会出现以下的异常信息： Exception in thread main mon.InvalidProgramException: Specifying keys via field positions is only valid for tuple data types. Type: String at mon.operators.Keys$ExpressionKeys.init(Keys.java:232) at mon.operators.Keys$ExpressionKeys.init(Keys.java:223) at org.apache.flink.api.scala.DataSet.groupBy(DataSet.scala:875) at com.iteblog.flink.GroupCombine$.main(GroupCombine.scala:14) at com.iteblog.flink.GroupCombine.main(GroupCombine.scala) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java: