博客

前言

在Flink中比如某些算子（join,coGroup,keyBy,groupBy）要求在数据元上定义key。另外有些算子操作，例如reduce，groupReduce，Aggregate，Windows需要数据在处理之前根据key进行分组。

在Flink中数据模型不是基于Key,Value格式处理的，因此不需将数据处理成键值对的格式，key是“虚拟的”，可以人为的来指定，实际数据处理过程中根据指定的key来对数据进行分组，DataSet中使用groupBy来指定key,DataStream中使用keyBy来指定key。那么如何指定keys呢?

一.使用Tuples来指定key

定义元组来指定key可以指定tuple中的第几个元素当做key，或者指定tuple中的联合元素当做key。需要使用

org.apache.flink.api.java.tuple.TupleXX

包下的tuple,最多支持25个元素且Tuple必须new创建。

如果Tuple是嵌套的格式，例如：DataStream<Tuple3<Tuple2<Integer, Float>,String,Long>> ds，如果指定keyBy(0)则会使用内部的整个Tuple2作为key。如果想要使用内部Tuple2中的Float格式当做key，可以使用

keyBy("f0.f1")

这样的形式指定。

这里需要注意，在Flink的Tuple中指定的key的下标从0开始算起，这里不像Scala中的Tuple从1开始算起，同时一般需要指定key的函数中都可以有两种写法，一种是直接写数字0，1，2等等，还有一种是写字符串的形式前面的0，1，2对应的字符串的表达形式为f1，f2，f3。

如果需要指定多个字段当做联合的Key，可以写成

keyBy(0,1)

，如果写成字符串形式在字符串中指定多个key，还可以写成

keyBy("f0","f1")

的形式。

二.使用Field Expression来指定key

可以使用Field Expression来指定key,一般作用的对象可以是类对象，或者嵌套的Tuple格式的数据。

对于这种形式的使用，注意点如下：

1.对于类对象可以使用类中的字段来指定key，类对象定义需要注意：

• 类的访问级别必须是public
• 必须写出默认的空的构造函数
• 类中所有的字段必须是public的或者必须有getter，setter方法。
• Flink必须支持字段的类型。

2.对于嵌套的Tuple类型的Tuple数据可以使用”xx.f0″表示嵌套tuple中第一个元素，也可以直接使用”xx.0”来表示第一个元素。

三.使用Key Selector Functions来指定key

使用key Selector这种方式选择key，非常方便，可以从数据类型中指定想要的key.

KeyedStream<String, String> keyBy = socketText.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String line) throws Exception {
                return line.split("\t")[2];
            }
        });