程序员社区

Flink中指定Key的几种方式

前言

在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];
            }
        });
赞(0) 打赏
未经允许不得转载:IDEA激活码 » Flink中指定Key的几种方式

相关推荐

  • 暂无文章

一个分享Java & Python知识的社区