Flink

爱被打了一巴掌 2021-10-23 13:52 404阅读 0赞

# Flink #

## 一、简介 ##

Flink核心是一个流式的数据流执行引擎，其针对数据流的分布式计算提供了数据分布、数据通信以及容错机制等功能。基于流执行引擎，Flink提供了诸多更高抽象层的API以便用户编写分布式任务：

DataSet API， 对静态数据进行批处理操作，将静态数据抽象成分布式的数据集，用户可以方便地使用Flink提供的各种操作符对分布式数据集进行处理，支持Java、Scala和Python。

DataStream API，对数据流进行流处理操作，将流式的数据抽象成分布式的数据流，用户可以方便地对分布式数据流进行各种操作，支持Java和Scala。

Table API，对结构化数据进行查询操作，将结构化数据抽象成关系表，并通过类SQL的DSL对关系表进行各种查询操作，支持Java和Scala。

此外，Flink还针对特定的应用领域提供了领域库，例如：

Flink ML，Flink的机器学习库，提供了机器学习Pipelines API并实现了多种机器学习算法。

Gelly，Flink的图计算库，提供了图计算的相关API及多种图计算算法实现。

## 二、优点 ##

Flink 是一个开源的分布式流式处理框架：

①提供准确的结果，甚至在出现无序或者延迟加载的数据的情况下。

②它是状态化的容错的，同时在维护一次完整的的应用状态时，能无缝修复错误。

③大规模运行，在上千个节点运行时有很好的吞吐量和低延迟。

Flink处理不同数据分类：

①有界数据流–Flink批处理–DataSet–groupBy

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

②无界数据流–Flink流处理–DataStream–keyBy–流处理默认一条条数据处理

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

③有/无界数据流–Flink SQL

Flink & Storm & SparkStreaming 流处理框架的区别：

•Strom:纯实时处理数据，吞吐量小 --水龙头滴水

•SparkStreaming : 准实时处理数据，微批处理数据，吞吐量大 --河道中开闸关闸

•Flink:纯实时处理数据，吞吐量大 --河流远远不断

## 三、注意点 ##

Flink程序的执行过程：
    	获取flink的执行环境(execution environment)
    	加载数据				-- soure
    	对加载的数据进行转换 		-- transformation
    	对结果进行保存或者打印	    -- sink
    	触发flink程序的执行(execute(),count(),collect(),print())，例如：调用						ExecutionEnvironment或者StreamExecutionEnvironment的execute()方法。

1.创建环境
          批处理：ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
          流处理：StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          
    2.在批处理中Flink处理的数据对象是DataSet
        在流处理中Flink处理的数据对象时DataStream
    
    3.代码流程必须符合 source ->transformation -> sink
          transformation 都是懒执行，需要最后使用env.execute()触发执行或者使用 print() 					,count(),collect() 触发执行
    
    4.Flink编程不是基于K,V格式的编程，通过某些方式来指定虚拟key
    
    5.Flink中的tuple最多支持25个元素，每个元素是从0开始

## 四、WordCount ##

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.common.operators.Order;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.operators.*;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.core.fs.FileSystem;
    import org.apache.flink.util.Collector;
    
    public class WordCount {
        public static void main(String[] args) throws Exception{
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            DataSource<String> dataSource = env.readTextFile("./data/words");
            FlatMapOperator<String, String> words = dataSource.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public void flatMap(String line, Collector<String> collector) throws Exception {
                    String[] split = line.split(" ");
                    for(String word : split){
                        collector.collect(word);
                    }
                }
            });
    
            MapOperator<String, Tuple2<String, Integer>> pairWords = words.map(new MapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public Tuple2<String, Integer> map(String word) throws Exception {
                    return new Tuple2<>(word, 1);
                }
            });
    
            UnsortedGrouping<Tuple2<String, Integer>> groupBy = pairWords.groupBy(0);
            AggregateOperator<Tuple2<String, Integer>> result = groupBy.sum(1).setParallelism(1);
            SortPartitionOperator<Tuple2<String, Integer>> sortResult = result.sortPartition(1, Order.DESCENDING);
    
            sortResult.writeAsCsv("./data/result/r2","\n","&", FileSystem.WriteMode.OVERWRITE);
            env.execute("myflink");
            result.print();
        }
    }

## 五、排序 ##

在Flink中，排序是默认在各个分区中进行的排序，所以最终结果在默认情况下是不能进行排序展示的

这时， ①可以给相关的算子进行设计–sum().setParallelism(1)

AggregateOperator<Tuple2<String, Integer>> result = groupBy.sum(1).setParallelism(1);

②全局设计，但是全局设计限制分区数，影响框架运行速度，通常不建议–env. setParallelism(1)

## 六、source数据源和Sink源 ##

### 1 数据源Source ###

Source 是Flink 获取数据的地方。以下source 中和批处理的source 类似，但是以下  
源作为dataStream 流处理时，是一条条处理，最终得到的不是一个总结果，而是每  
次处理后都会得到一个结果。  
 1).socketTextStream – 读取Socket 数据流  
 2). readTextFile() – 逐行读取文本文件获取数据流，每行都返回字符串。  
 3).fromCollection() – 从集合中创建数据流。  
 4).fromElements – 从给定的数据对象创建数据流，所有数据类型要一致。  
 5).addSource – 添加新的源函数,例如从kafka 中读取数据，参见读取kafka 数据案例。

### 2 数据写出Sink ###

1).writeAsText() – 以字符串的形式逐行写入文件，调用每个元素的toString()得到写  
入的字符串。  
 2).writeAsCsv() – 将元组写出以逗号分隔的csv 文件。注意：只能作用到元组数据上。  
 3).print() – 控制台直接输出结果，调用对象的toString()方法得到输出结果。  
 4).addSink() – 自定义接收函数。例如将结果保存到kafka 中，参见kafka 案例。

## 七、累加器&计数器 ##

Flink 累加器

Accumulator即累加器，可以在分布式统计数据，只有在任务结束之后才能获取累加器的最终结果。计数器是累加器的具体实现，有：IntCounter,LongCounter和DoubleCounter。

累加器注意事项

1.需要在算子内部创建累加器对象

2.通常在Rich函数中的open方法中注册累加器，指定累加器的名称

3.在当前算子内任意位置可以使用累加器

4.必须当任务执行结束后，通过env.execute(xxx)执行后的JobExecutionResult对象获取累加器的值。

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    DataSource<String> dataSource = env.fromElements("a", "b", "c", "d", "e", "f");
    MapOperator<String, String> map = dataSource.map(new RichMapFunction<String, String>(){
    
    //1.创建累加器,在算子中创建累加器对象
    	private IntCounter numLines = new IntCounter();
    
    //2.注册累加器对象，通常在Rich 函数的open 方法中使用
    // getRuntimeContext().addAccumulator("num-lines", this.numLines);注册累加器
        public void open(Configuration parameters) throws Exception {
            getRuntimeContext().addAccumulator("num-lines", this.numLines);
    	}
    
        @Override
        public String map(String s) throws Exception {
        //3.使用累加器，可以在任意操作中使用，包括在open 或者close 方法中
            this.numLines.add(1);
            return s;
        }
        }).setParallelism(8);
        map.writeAsText("./TempResult/result",FileSystem.WriteMode.OVERWRITE);
        JobExecutionResult myJobExecutionResult = env.execute("IntCounterTest");
        
        //4.当作业执行完成之后，在JobExecutionResult 对象中获取累加器的值。
            int accumulatorResult = myJobExecutionResult.getAccumulatorResult("num-lines");
            System.out.println("accumulator value = "+accumulatorResult);

## 八、指定虚拟Key ##

Apache Flink 指定虚拟key

1.使用Tuples来指定key

2.使用Field Expression来指定key

3.使用Key Selector Functions来指定key

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

1.  使用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，  
    就要使用第二种方式Field Expression 来指定key。
2.  使用Field Expression 来指定key  
    可以使用Field Expression 来指定key,一般作用的对象可以是类对象，或者嵌套的  
    Tuple 格式的数据。  
    使用注意点：  
    a) 对于类对象可以使用类中的字段来指定key。  
    类对象定义需要注意：  
    i. 类的访问级别必须是public  
    ii. 必须写出默认的空的构造函数  
    iii. 类中所有的字段必须是public 的或者必须有getter，setter 方法。例如类  
    中有个字段是foo,那么这个字段的getter，setter 方法为：getFoo() 和  
    setFoo().  
    iv. Flink 必须支持字段的类型。一般类型都支持  
    b) 对于嵌套的Tuple 类型的Tuple 数据可以使用”xx.f0”表示嵌套tuple 中第一个元  
    素， 也可以直接使用”xx.0” 来表示第一个元素， 参照案例  
    GroupByUseFieldExpressions。
3.  使用Key Selector Functions 来指定key  
    使用key Selector 这种方式选择key，非常方便，可以从数据类型中指定想要的key.

## 九、Flink & Kafka ##

IDEA代码—MyFlinkCode

### 1 Flink消费kafka数据起始offset配置 ###

\\1.flinkKafkaConsumer.setStartFromEarliest()

从topic的最早offset位置开始处理数据，如果kafka中保存有消费者组的消费位置将被忽略。

\\2. flinkKafkaConsumer.setStartFromLatest()

从topic的最新offset位置开始处理数据，如果kafka中保存有消费者组的消费位置将被忽略。

\\3. flinkKafkaConsumer.setStartFromTimestamp(…)

从指定的时间戳（毫秒）开始消费数据，Kafka中每个分区中数据大于等于设置的时间戳的数据位置将被当做开始消费的位置。如果kafka中保存有消费者组的消费位置将被忽略。

\\4. flinkKafkaConsumer.setStartFromGroupOffsets()

默认的设置。根据代码中设置的group.id设置的消费者组，去kafka中或者zookeeper中找到对应的消费者offset位置消费数据。如果没有找到对应的消费者组的位置，那么将按照auto.offset.reset设置的策略读取offset。

### 2 Flink消费kafka数据，消费者offset提交配置 ###

1.Flink提供了消费kafka数据的offset如何提交给Kafka或者zookeeper(kafka0.8之前)的配置。注意，Flink并不依赖提交给Kafka或者zookeeper中的offset来保证容错。提交的offset只是为了外部来查询监视kafka数据消费的情况

2.配置offset的提交方式取决于是否为job设置开启checkpoint。可以使用env.enableCheckpointing(5000)来设置开启checkpoint。

3.关闭checkpoint：

如何禁用了checkpoint，那么offset位置的提交取决于Flink读取kafka客户端的配置，enable.auto.commit ( auto.commit.enable【Kafka 0.8】)配置是否开启自动提交offset, auto.commit.interval.ms决定自动提交offset的周期。

4.开启checkpoint：

如果开启了checkpoint，那么当checkpoint保存状态完成后，将checkpoint中保存的offset位置提交到kafka。这样保证了Kafka中保存的offset和checkpoint中保存的offset一致，可以通过配置setCommitOffsetsOnCheckpoints(boolean)来配置是否将checkpoint中的offset提交到kafka中（默认是true）。如果使用这种方式，那么properties中配置的kafka offset自动提交参数enable.auto.commit和周期提交参数auto.commit.interval.ms参数将被忽略。

### 3 Flink中外部状态实现两阶段提交 ###

Flink外部状态实现两阶段提交将逻辑封装到TwoPhaseComitSinkFunction类中，需要扩展TwoPhaseCommitSinkFunction来实现仅一次消费数据。若要实现支持exactly-once语义的自定义sink,需要实现以下4个方法：

\\1. beginTransaction:开启一个事务，创建一个临时文件，将数据写入到临时文件中

\\2. preCommit:在pre-commit阶段，flush缓存数据到磁盘，然后关闭这个文件，确保不会 有新的数据写入到这个文件，同时开启一个新事务执行属于下一个checkpoint的写入 操作。

\\3. commit：在commit阶段，我们以原子性的方式将上一阶段的文件写入真正的文件目 录下。【注意：数据有延时，不是实时的】。

\\4. abort:一旦异常终止事务，程序如何处理。这里要清除临时文件。

exactly-once语义的自定义sink,需要实现以下4个方法：

\\1. beginTransaction:开启一个事务，创建一个临时文件，将数据写入到临时文件中

\\3. commit：在commit阶段，我们以原子性的方式将上一阶段的文件写入真正的文件目 录下。【注意：数据有延时，不是实时的】。

\\4. abort:一旦异常终止事务，程序如何处理。这里要清除临时文件。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]: /images/20211023/7f0126c3da6945e486ade9cd62a3c2d7.png