Press "Enter" to skip to content

Flink基础:实时处理管道与ETL

本站内容均来自兴趣收集,如不慎侵害的您的相关权益,请留言告知,我们将尽快删除.谢谢.

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<TaxiRide> rides = env.addSource(new TaxiRideSource(...));
DataStream<EnrichedRide> enrichedNYCRides = rides
    .filter(new RideCleansingSolution.NYCFilter())
    .map(new Enrichment());
enrichedNYCRides.print();

 

flatmap()

 

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

 

DataStream<TaxiRide> rides = env.addSource(new TaxiRideSource(...));
DataStream<EnrichedRide> enrichedNYCRides = rides
    .flatMap(new NYCEnrichment());
enrichedNYCRides.print();

 

FlatMapFunction的定义:

 

public static class NYCEnrichment implements FlatMapFunction<TaxiRide, EnrichedRide> {
    @Override
    public void flatMap(TaxiRide taxiRide, Collector<EnrichedRide> out) throws Exception {
        FilterFunction<TaxiRide> 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 KeySelector<EnrichedRide, int>() {
            @Override
            public 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<Tuple2<Integer, Minutes>> minutesByStartCell = enrichedNYCRides
    .flatMap(new FlatMapFunction<EnrichedRide, Tuple2<Integer, Minutes>>() {
        @Override
        public void flatMap(EnrichedRide ride,
                            Collector<Tuple2<Integer, Minutes>> 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<Boolean> keyHasBeenSeen;
    @Override
    public void open(Configuration conf) {
        ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor<>("keyHasBeenSeen", Types.BOOLEAN);
        keyHasBeenSeen = getRuntimeContext().getState(desc);
    }
    @Override
    public void flatMap(Event event, Collector<Event> 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<String> control = env.fromElements("DROP", "IGNORE").keyBy(x -> x);
    DataStream<String> 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<Boolean> blocked;
    @Override
    public void open(Configuration config) {
        blocked = getRuntimeContext().getState(new ValueStateDescriptor<>("blocked", Boolean.class));
    }
    @Override
    public void flatMap1(String control_value, Collector<String> out) throws Exception {
        blocked.update(Boolean.TRUE);
    }
    @Override
    public void flatMap2(String data_value, Collector<String> out) throws Exception {
        if (blocked.value() == null) {
            out.collect(data_value);
        }
    }
}

 

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

 

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

Be First to Comment

发表评论

电子邮件地址不会被公开。 必填项已用*标注