flink源码分析_Flink基础：实时处理管道与ETL

往期推荐：

Flink基础：入门介绍

Flink基础：DataStream API

Flink深入浅出：资源管理

Flink深入浅出：部署模式

Flink深入浅出：内存模型

Flink深入浅出：JDBC Source从理论到实战

Flink深入浅出：Sql Gateway源码分析

Flink深入浅出：JDBC Connector源码分析

Flink的经典使用场景是ETL，即Extract抽取、Transform转换、Load加载，可以从一个或多个数据源读取数据，经过处理转换后，存储到另一个地方，本篇将会介绍如何使用DataStream API来实现这种应用。注意Flink Table和SQL
api 会很适合来做ETL，但是不妨碍从底层的DataStream API来了解其中的细节。

1 无状态的转换

无状态即不需要在操作中维护某个中间状态，典型的例子如map和flatmap。

map()

下面是一个转换操作的例子，需要根据输入数据创建一个出租车起始位置和目标位置的对象。首先定义出租车的位置对象：

public static class EnrichedRide extends TaxiRide {
    public int startCell;
    public int endCell;

    public EnrichedRide() {}

    public EnrichedRide(TaxiRide ride) {
        this.rideId = ride.rideId;
        this.isStart = ride.isStart;
        ...
        this.startCell = GeoUtils.mapToGridCell(ride.startLon, ride.startLat);
        this.endCell = GeoUtils.mapToGridCell(ride.endLon, ride.endLat);
    }

    public String toString() {
        return super.toString() + "," +
            Integer.toString(this.startCell) + "," +
            Integer.toString(this.endCell);
    }
}

使用的时候可以注册一个MapFunction，该函数接收TaxiRide对象，输出EnrichRide对象。

public static class Enrichment implements MapFunction<TaxiRide, EnrichedRide> {
    @Override
    public EnrichedRide map(TaxiRide taxiRide) throws Exception {
        return new EnrichedRide(taxiRide);
    }
}

使用时只需要创建map对象即可：

DataStream rides = env.addSource(new TaxiRideSource(...));
DataStream enrichedNYCRides = rides
    .filter(new RideCleansingSolution.NYCFilter())
    .map(new Enrichment());
enrichedNYCRides.print();

flatmap()

MapFunction适合一对一的转换，对于输入流的每个元素都有一个元素输出。如果需要一对多的场景，可以使用flatmap：

DataStream rides = env.addSource(new TaxiRideSource(...));
DataStream enrichedNYCRides = rides
    .flatMap(new NYCEnrichment());
enrichedNYCRides.print();

FlatMapFunction的定义：

public static class NYCEnrichment implements FlatMapFunction<TaxiRide, EnrichedRide> {
    @Override
    public void flatMap(TaxiRide taxiRide, Collector out) throws Exception {
        FilterFunction valid = new RideCleansing.NYCFilter();if (valid.filter(taxiRide)) {
            out.collect(new EnrichedRide(taxiRide));
        }
    }
}

通过collector，可以在flatmap中任意添加零个或多个元素。

2 Keyed Streams

keyBy()

有时需要对数据流按照某个字段进行分组，每个事件会根据该字段相同的值汇总到一起。比如，希望查找相同出发位置的路线。如果在SQL中可能会使用GROUP BY startCell，在Flink中可以直接使用keyBy函数：

rides
    .flatMap(new NYCEnrichment())
    .keyBy(value -> value.startCell)

keyBy会引起重分区而导致网络数据shuffle，通常这种代价都很昂贵，因为每次shuffle时需要进行数据的序列化和反序列化，既浪费CPU资源，又占用网络带宽。

通过对startCell进行分组，这种方式的分组可能会由于编译器而丢失字段的类型信息，因此Flink也支持把字段包装成Tuple，基于元素位置进行分组。当然也支持使用KeySelector函数，自定义分组规则。

rides
    .flatMap(new NYCEnrichment())
    .keyBy(
        new KeySelectorint>() {@Overridepublic int getKey(EnrichedRide enrichedRide) throws Exception {return enrichedRide.startCell;
            }
        })

可以直接使用lambda表达式：

rides
    .flatMap(new NYCEnrichment())
    .keyBy(enrichedRide -> enrichedRide.startCell)

key可以自定义计算规则

keyselector不限制从必须从事件中抽取key，也可以自定义任何计算key的方法。但需要保证输出的key是一致的，并且实现了对应的hashCode和equals方法。生成key的规则一定要稳定，因为生成key可能在应用运行的任何时间，因此一定要保证key生成规则的持续稳定。

key可以通过某个字段选择：

keyBy(enrichedRide -> enrichedRide.startCell)

也可以直接替换成某个方法：

keyBy(ride -> GeoUtils.mapToGridCell(ride.startLon, ride.startLat))

Keyed Stream的聚合

下面的例子中，创建了一个包含startCell和花费时间的二元组：

import org.joda.time.Interval;

DataStream> minutesByStartCell = enrichedNYCRides
    .flatMap(new FlatMapFunction>() {@Overridepublic void flatMap(EnrichedRide ride,
                            Collector> out) throws Exception {if (!ride.isStart) {
                Interval rideInterval = new Interval(ride.startTime, ride.endTime);
                Minutes duration = rideInterval.toDuration().toStandardMinutes();
                out.collect(new Tuple2<>(ride.startCell, duration));
            }
        }
    });

现在需要输出每个起始位置最长距离的路线，有很多种方式可以实现。以上面的数据为例，可以通过startcell进行聚合，然后选择时间最大的元素输出：

minutesByStartCell
  .keyBy(value -> value.f0) // .keyBy(value -> value.startCell)
  .maxBy(1) // duration
  .print();

可以得到输出结果：

4> (64549,5M)
4> (46298,18M)
1> (51549,14M)
1> (53043,13M)
1> (56031,22M)
1> (50797,6M)
...
1> (50797,8M)
...
1> (50797,11M)
...
1> (50797,12M)

状态

上面是一个有状态的例子，Flink需要记录每个key的最大值。无论何时在应用中涉及到状态，都需要考虑这个状态有多大。如果key的空间是无限大的，那么flink可能需要维护大量的状态信息。当使用流时，一定要对无限窗口的聚合十分敏感，因为它是对整个流进行操作，很有可能因为维护的状态信息不断膨胀，而导致内存溢出。在上面使用的maxBy就是经典的的聚合操作，也可以使用更通用的reduce来自定义聚合方法。

3 有状态的操作

Flink针对状态的管理有很多易用的特性，比如：

支持本地保存：基于进程内存来保存状态
状态的持久化：定期保存到检查点，保证容错
垂直扩展：Flink状态可以把状态保存到RocksDB中，也支持扩展到本地磁盘
水平扩展：状态支持在集群中扩缩容，通过调整并行度，自动拆分状态
可查询：Flink的状态可以在外部直接查询

Rich函数

Flink有几种函数接口，包括FilterFunction, MapFunction，FlatMapFunction等。对于每个接口，Flink都提供了对应的Rich方法。比如RichFlatMapFunction，提供了额外的一些方法：

open(Configuration c) 在初始化的时候调用一次，用于加载静态数据，开启外部服务的连接等
close() 流关闭时调用
getRuntimeContext() 提供进入全局状态的方法，需要了解如何创建和查询状态

使用Keyed State的例子

下面是一个针对事件的key进行去重的例子：

private static class Event {
    public final String key;
    public final long timestamp;
    ...
}

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.addSource(new EventSource())
        .keyBy(e -> e.key)
        .flatMap(new Deduplicator())
        .print();

    env.execute();
}

为了实现这个功能，deduplicator需要记住一些信息，对于每个key，都需要记录是否已经存在。Flink支持几种不同类型的状态，最简单的一种是valueState。对于每个key，flink都为它保存一个对象，在上面的例子中对象是Boolean。Deduplicator有两个方法：open()和flatMap()。open方法通过descriptor为状态起了一个标识名称，并声明类型为Boolean。

public static class Deduplicator extends RichFlatMapFunction<Event, Event> {
    ValueState keyHasBeenSeen;@Overridepublic void open(Configuration conf) {
        ValueStateDescriptor desc = new ValueStateDescriptor<>("keyHasBeenSeen", Types.BOOLEAN);
        keyHasBeenSeen = getRuntimeContext().getState(desc);
    }@Overridepublic void flatMap(Event event, Collector out) throws Exception {if (keyHasBeenSeen.value() == null) {
            out.collect(event);
            keyHasBeenSeen.update(true);
        }
    }
}

flatMap中调用state.value()获取状态。flink在上下文中为每个key保存了一个状态值，只有当值为null时，说明这个key之前没有出现过，然后将其更新为true。当flink调用open时，状态是空的。但是当调用flatMap时，key可以通过context进行访问。当在集群模式中运行时，会有很多个Deduplicator实例，每个负责维护一部分key的事件。因此，当使用单个事件的valuestate时，要理解它背后其实不是一个值，而是每个key都对应一个状态值，并且分布式的存储在集群中的各个节点进程上。

清除状态

有时候key的空间可能是无限制的，flink会为每个key存储一个boolean对象。如果key的数量是有限的还好，但是应用往往是持续不间断的运行，那么key可能会无限增长，因此需要清理不再使用的key。可以通过state.clear()进行清理。比如针对某个key按照某一时间频率进行清理，在processFunction中可以了解到如何在事件驱动的应用中执行定时器操作。也可以在状态描述符中为状态设置TTL生存时间，这样状态可以自动进行清理。

非keyed状态

状态也支持在非key类型的上下文中使用，这种叫做操作符状态，operator state。典型的场景是Flink读取Kafka时记录的offset信息。

4 连接流

大部分场景中Flink都是接收一个数据流输出一个数据流，类似管道式的处理数据：

也有的场景需要动态的修改函数中的信息，比如阈值、规则或者其他的参数，这种设计叫做connected streams，流会拥有两个输入，类似：

在下面的例子中，通过控制流用来指定必须过滤的单词：

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    DataStream control = env.fromElements("DROP", "IGNORE").keyBy(x -> x);
    DataStream streamOfWords = env.fromElements("Apache", "DROP", "Flink", "IGNORE").keyBy(x -> x);
    control
        .connect(datastreamOfWords)
        .flatMap(new ControlFunction())
        .print();
    env.execute();
}

两个流可以通过key的方式连接，keyby用来分组数据，这样保证相同类型的数据可以进入到相同的实例中。上面的例子两个流都是字符串，

public static class ControlFunction extends RichCoFlatMapFunction<String, String, String> {
    private ValueState blocked;@Overridepublic void open(Configuration config) {
        blocked = getRuntimeContext().getState(new ValueStateDescriptor<>("blocked", Boolean.class));
    }@Overridepublic void flatMap1(String control_value, Collector out) throws Exception {
        blocked.update(Boolean.TRUE);
    }@Overridepublic void flatMap2(String data_value, Collector out) throws Exception {if (blocked.value() == null) {
            out.collect(data_value);
        }
    }
}

blocked用于记录key的控制逻辑，key的state会在两个流间共享。flatMap1和flatMap2会被两个流调用，分别用来更新和获取状态，从而实现通过一个流控制另一个流的目的。

总结：本片从状态上讲述了有状态的操作和无状态的操作，还介绍了状态的使用以及连接流的适用场景。后面会介绍DataStream的操作和状态的管理。