文档章节

《从0到1学习Flink》—— Flink Data transformation(转换)

群星纪元
 群星纪元
发布于 03/18 14:06
字数 1873
阅读 40
收藏 0

前言

在第一篇介绍 Flink 的文章 《《从0到1学习Flink》—— Apache Flink 介绍》 中就说过 Flink 程序的结构

Flink 应用程序结构就是如上图所示:

1、Source: 数据源,Flink 在流处理和批处理上的 source 大概有 4 类:基于本地集合的 source、基于文件的 source、基于网络套接字的 source、自定义的 source。自定义的 source 常见的有 Apache kafka、Amazon Kinesis Streams、RabbitMQ、Twitter Streaming API、Apache NiFi 等,当然你也可以定义自己的 source。

2、Transformation:数据转换的各种操作,有 Map / FlatMap / Filter / KeyBy / Reduce / Fold / Aggregations / Window / WindowAll / Union / Window join / Split / Select / Project 等,操作很多,可以将数据转换计算成你想要的数据。

3、Sink:接收器,Flink 将转换计算后的数据发送的地点 ,你可能需要存储下来,Flink 常见的 Sink 大概有如下几类:写入文件、打印出来、写入 socket 、自定义的 sink 。自定义的 sink 常见的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem 等,同理你也可以定义自己的 Sink。

在上四篇文章介绍了 Source 和 Sink:

1、《从0到1学习Flink》—— Data Source 介绍

2、《从0到1学习Flink》—— 如何自定义 Data Source ?

3、《从0到1学习Flink》—— Data Sink 介绍

4、《从0到1学习Flink》—— 如何自定义 Data Sink ?

那么这篇文章我们就来看下 Flink Data Transformation 吧,数据转换操作还是蛮多的,需要好好讲讲!

Transformation

Map

这是最简单的转换之一,其中输入是一个数据流,输出的也是一个数据流:

还是拿上一篇文章的案例来将数据进行 map 转换操作:

1
2
3
4
5
6
7
8
9
10
11
12
SingleOutputStreamOperator<Student> map = student.map(new MapFunction<Student, Student>() {
    @Override
    public Student map(Student value) throws Exception {
        Student s1 = new Student();
        s1.id = value.id;
        s1.name = value.name;
        s1.password = value.password;
        s1.age = value.age + 5;
        return s1;
    }
});
map.print();

将每个人的年龄都增加 5 岁,其他不变。

FlatMap

FlatMap 采用一条记录并输出零个,一个或多个记录。

1
2
3
4
5
6
7
8
9
SingleOutputStreamOperator<Student> flatMap = student.flatMap(new FlatMapFunction<Student, Student>() {
    @Override
    public void flatMap(Student value, Collector<Student> out) throws Exception {
        if (value.id % 2 == 0) {
            out.collect(value);
        }
    }
});
flatMap.print();

这里将 id 为偶数的聚集出来。

Filter

Filter 函数根据条件判断出结果。

1
2
3
4
5
6
7
8
9
10
SingleOutputStreamOperator<Student> filter = student.filter(new FilterFunction<Student>() {
    @Override
    public boolean filter(Student value) throws Exception {
        if (value.id > 95) {
            return true;
        }
        return false;
    }
});
filter.print();

这里将 id 大于 95 的过滤出来,然后打印出来。

KeyBy

KeyBy 在逻辑上是基于 key 对流进行分区。在内部,它使用 hash 函数对流进行分区。它返回 KeyedDataStream 数据流。

1
2
3
4
5
6
7
KeyedStream<Student, Integer> keyBy = student.keyBy(new KeySelector<Student, Integer>() {
    @Override
    public Integer getKey(Student value) throws Exception {
        return value.age;
    }
});
keyBy.print();

上面对 student 的 age 做 KeyBy 操作分区

Reduce

Reduce 返回单个的结果值,并且 reduce 操作每处理一个元素总是创建一个新值。常用的方法有 average, sum, min, max, count,使用 reduce 方法都可实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
SingleOutputStreamOperator<Student> reduce = student.keyBy(new KeySelector<Student, Integer>() {
    @Override
    public Integer getKey(Student value) throws Exception {
        return value.age;
    }
}).reduce(new ReduceFunction<Student>() {
    @Override
    public Student reduce(Student value1, Student value2) throws Exception {
        Student student1 = new Student();
        student1.name = value1.name + value2.name;
        student1.id = (value1.id + value2.id) / 2;
        student1.password = value1.password + value2.password;
        student1.age = (value1.age + value2.age) / 2;
        return student1;
    }
});
reduce.print();

上面先将数据流进行 keyby 操作,因为执行 reduce 操作只能是 KeyedStream,然后将 student 对象的 age 做了一个求平均值的操作。

Fold

Fold 通过将最后一个文件夹流与当前记录组合来推出 KeyedStream。 它会发回数据流。

1
2
3
4
5
6
KeyedStream.fold("1", new FoldFunction<Integer, String>() {
    @Override
    public String fold(String accumulator, Integer value) throws Exception {
        return accumulator + "=" + value;
    }
})

Aggregations

DataStream API 支持各种聚合,例如 min,max,sum 等。 这些函数可以应用于 KeyedStream 以获得 Aggregations 聚合。

1
2
3
4
5
6
7
8
9
10
KeyedStream.sum(0) 
KeyedStream.sum("key") 
KeyedStream.min(0) 
KeyedStream.min("key") 
KeyedStream.max(0) 
KeyedStream.max("key") 
KeyedStream.minBy(0) 
KeyedStream.minBy("key") 
KeyedStream.maxBy(0) 
KeyedStream.maxBy("key")

max 和 maxBy 之间的区别在于 max 返回流中的最大值,但 maxBy 返回具有最大值的键, min 和 minBy 同理。

Window

Window 函数允许按时间或其他条件对现有 KeyedStream 进行分组。 以下是以 10 秒的时间窗口聚合:

1
inputStream.keyBy(0).window(Time.seconds(10));

Flink 定义数据片段以便(可能)处理无限数据流。 这些切片称为窗口。 此切片有助于通过应用转换处理数据块。 要对流进行窗口化,我们需要分配一个可以进行分发的键和一个描述要对窗口化流执行哪些转换的函数

要将流切片到窗口,我们可以使用 Flink 自带的窗口分配器。 我们有选项,如 tumbling windows, sliding windows, global 和 session windows。 Flink 还允许您通过扩展 WindowAssginer 类来编写自定义窗口分配器。 这里先预留下篇文章来讲解这些不同的 windows 是如何工作的。

WindowAll

windowAll 函数允许对常规数据流进行分组。 通常,这是非并行数据转换,因为它在非分区数据流上运行。

与常规数据流功能类似,我们也有窗口数据流功能。 唯一的区别是它们处理窗口数据流。 所以窗口缩小就像 Reduce 函数一样,Window fold 就像 Fold 函数一样,并且还有聚合。

1
inputStream.keyBy(0).windowAll(Time.seconds(10));

Union

Union 函数将两个或多个数据流结合在一起。 这样就可以并行地组合数据流。 如果我们将一个流与自身组合,那么它会输出每个记录两次。

1
inputStream.union(inputStream1, inputStream2, ...);

Window join

我们可以通过一些 key 将同一个 window 的两个数据流 join 起来。

1
2
3
4
inputStream.join(inputStream1)
           .where(0).equalTo(1)
           .window(Time.seconds(5))     
           .apply (new JoinFunction () {...});

以上示例是在 5 秒的窗口中连接两个流,其中第一个流的第一个属性的连接条件等于另一个流的第二个属性。

Split

此功能根据条件将流拆分为两个或多个流。 当您获得混合流并且您可能希望单独处理每个数据流时,可以使用此方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
SplitStream<Integer> split = inputStream.split(new OutputSelector<Integer>() {
    @Override
    public Iterable<String> select(Integer value) {
        List<String> output = new ArrayList<String>(); 
        if (value % 2 == 0) {
            output.add("even");
        }
        else {
            output.add("odd");
        }
        return output;
    }
});

Select

此功能允许您从拆分流中选择特定流。

1
2
3
4
SplitStream<Integer> split;
DataStream<Integer> even = split.select("even"); 
DataStream<Integer> odd = split.select("odd"); 
DataStream<Integer> all = split.select("even","odd");

Project

Project 函数允许您从事件流中选择属性子集,并仅将所选元素发送到下一个处理流。

1
2
DataStream<Tuple4<Integer, Double, String, String>> in = // [...] 
DataStream<Tuple2<String, String>> out = in.project(3,2);

上述函数从给定记录中选择属性号 2 和 3。 以下是示例输入和输出记录:

1
2
(1,10.0,A,B)=> (B,A)
(2,20.0,C,D)=> (D,C)

最后

本文主要介绍了 Flink Data 的常用转换方式:Map、FlatMap、Filter、KeyBy、Reduce、Fold、Aggregations、Window、WindowAll、Union、Window Join、Split、Select、Project 等。并用了点简单的 demo 介绍了如何使用,具体在项目中该如何将数据流转换成我们想要的格式,还需要根据实际情况对待。

本文转载自:http://www.54tianzhisheng.cn/2018/11/04/Flink-Data-transformation/

群星纪元
粉丝 45
博文 456
码字总数 44497
作品 0
朝阳
高级程序员
私信 提问
《从0到1学习Flink》目录

1、《从0到1学习Flink》—— Apache Flink 介绍 2、《从0到1学习Flink》—— Mac 上搭建 Flink 1.6.0 环境并构建运行简单程序入门 3、《从0到1学习Flink》—— Flink 配置文件详解 4、《从0...

火力全開
03/18
279
0
Flink 源码解析 —— TaskManager 处理 SubmitJob 的过程

TaskManager 处理 SubmitJob 的过程 https://t.zsxq.com/eu7mQZj 博客 1、Flink 从0到1学习 —— Apache Flink 介绍 2、Flink 从0到1学习 —— Mac 上搭建 Flink 1.6.0 环境并构建运行简单程...

zhisheng_t
09/03
34
0
Flink 源码解析 —— 分析 Streaming WordCount 程序的执行过程

流处理 WordCount 程序 https://t.zsxq.com/qnMFEUJ 博客 1、Flink 从0到1学习 —— Apache Flink 介绍 2、Flink 从0到1学习 —— Mac 上搭建 Flink 1.6.0 环境并构建运行简单程序入门 3、F...

zhisheng_t
09/01
24
0
Flink 源码解析 —— Flink TaskManager 有什么作用?

TaskManager 有什么作用 https://t.zsxq.com/RZbu7yN 博客 1、Flink 从0到1学习 —— Apache Flink 介绍 2、Flink 从0到1学习 —— Mac 上搭建 Flink 1.6.0 环境并构建运行简单程序入门 3、F...

zhisheng_t
09/01
58
0
Flink 源码解析 —— 项目结构一览

Flink 源码项目结构一览 https://t.zsxq.com/MNfAYne 博客 1、Flink 从0到1学习 —— Apache Flink 介绍 2、Flink 从0到1学习 —— Mac 上搭建 Flink 1.6.0 环境并构建运行简单程序入门 3、F...

zhisheng_t
09/01
17
0

没有更多内容

加载失败,请刷新页面

加载更多

Error和Exception

1.Error类和Exception类都是继承Throwable类 2.Error(错误)是系统中的错误,程序员是不能改变的和处理的,是在程序编译时出现的错误,只能通过修改程序才能修正。一般是指与虚拟机相关的问...

大瑞清_liurq
24分钟前
4
0
8086汇编基础 start 程序入口标签的示例

    IDE : Masm for Windows 集成实验环境 2015     OS : Windows 10 x64 typesetting : Markdown    blog : my.oschina.net/zhichengjiu    gitee : gitee.com/zhichengjiu   ......

志成就
30分钟前
4
0
uni app 零基础小白到项目实战2

<template> <scroll-view v-for="(card, index) in list" :key="index"> <view v-for =(item, itemIndex) in card"> {{item.value}}</view> </scroll-view></template> GraceUi va......

达达前端小酒馆
31分钟前
4
0
http keep-alive 解释

本文转载于:专业的前端网站➜http keep-alive 解释 1、概念 keep-alive示例: keep-alive模式(又称持久连接、连接重用)时,keep-alive功能使客户端到服务器端的连接持续有效,当出现对服务...

前端老手
36分钟前
4
0
groovy爬虫实例——历史上的今天

最近做了一个历史上今天的爬虫程序,跟历史天气数据源一致,数据量比较小,几十秒就爬完了。中间遇到一些问题,一起分享出来供大家参考。本项目源码和相关数据已经上传到了github,有兴趣的朋...

八音弦
45分钟前
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部