Flink学习（四）-- Flink常用流处理API

青旅半醒 2022-12-27 01:19 214阅读 0赞

### 文章目录 ###

*  Flink流处理API
 *   *  1. environment 创建执行环境
     *   *  1.1 createLocalEnvironment
         *  1.2 createRemoteEnvironment
         *  1.3 getExecutionEnvironment
     *  2. source 获取数据源
     *   *  2.1 fromCollection从本地集合获取
         *  2.2 readTextFile从文件获取
         *  2.3 socketTextStream 从网络套接字获取
         *  2.4 从kafka获取数据
         *  2.5 从其他自定义数据源获取
     *  3. transform 转换算子
     *   *  3.1 map
         *  3.2 flatMap
         *  3.3 filter
         *  3.4 keyBy
         *  3.5 滚动聚合算子
         *   *  3.5.1 sum
             *  3.5.2 min & minBy
             *  3.5.3 reduce
             *  3.5.4 split & select
             *  3.5.5 Connect & CoMap
             *  3.5.6 union
     *  4. sink 结果集下沉
     *   *  4.1 sink到文件
         *  4.2 sink到kafka

# Flink流处理API #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70]  
flink的流处理API主要分为四个模块：environment，source，transform和sink，各模块的作用分别如下：  
**environment**: 创建执行环境  
**source**: 获取数据源  
**transform**: 转换算子, 做主要的数据处理逻辑  
**sink**: 将处理好的数据输出到外部系统中

## 1. environment 创建执行环境 ##

创建执行环境共有三种方式：

1.  `createLocalEnvironment`创建本地环境
2.  `createRemoteEnvironment` 提交到远程集群环境执行
3.  `getExecutionEnvironment` 根据当前执行环境创建执行环境

### 1.1 createLocalEnvironment ###

创建本地执行环境，需要在调用时指定默认的并行度，创建方式如下：

//返回本地执行环境，需要在调用时指定默认的并行度
     LocalStreamEnvironment localEnvironment = StreamExecutionEnvironment.createLocalEnvironment();

有三种创建方式，可选指定并行度，可传自定义配置，不指定并行度默认为`Runtime.getRuntime().availableProcessors()`,即当前运行环境的计算资源可用线程数

/** * Creates a {@link LocalStreamEnvironment}. The local execution environment * will run the program in a multi-threaded fashion in the same JVM as the * environment was created in. The default parallelism of the local * environment is the number of hardware contexts (CPU cores / threads), * unless it was specified differently by {@link #setParallelism(int)}. * * @return A local execution environment. */
    	public static LocalStreamEnvironment createLocalEnvironment() { 
    		return createLocalEnvironment(defaultLocalParallelism);
    	}
    
    	/** * Creates a {@link LocalStreamEnvironment}. The local execution environment * will run the program in a multi-threaded fashion in the same JVM as the * environment was created in. It will use the parallelism specified in the * parameter. * * @param parallelism * The parallelism for the local environment. * @return A local execution environment with the specified parallelism. */
    	public static LocalStreamEnvironment createLocalEnvironment(int parallelism) { 
    		return createLocalEnvironment(parallelism, new Configuration());
    	}
    
    	/** * Creates a {@link LocalStreamEnvironment}. The local execution environment * will run the program in a multi-threaded fashion in the same JVM as the * environment was created in. It will use the parallelism specified in the * parameter. * * @param parallelism * The parallelism for the local environment. * @param configuration * Pass a custom configuration into the cluster * @return A local execution environment with the specified parallelism. */
    	public static LocalStreamEnvironment createLocalEnvironment(int parallelism, Configuration configuration) { 
    		final LocalStreamEnvironment currentEnvironment;
    
    		currentEnvironment = new LocalStreamEnvironment(configuration);
    		currentEnvironment.setParallelism(parallelism);
    
    		return currentEnvironment;
    	}

![在这里插入图片描述][20201206203704906.png]

### 1.2 createRemoteEnvironment ###

创建集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号，并指定要在集群中运行的 Jar 包，创建方式如下：

//返回集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号，并指定要在集群中运行的 Jar 包。
            StreamExecutionEnvironment remoteEnvironment = StreamExecutionEnvironment.
                    createRemoteEnvironment("JobManagerHost",6201,"/user/flink-demo.jar");

同样有三种创建方式，可选指定并行度，可传自定义配置，不指定并行度默认为`Runtime.getRuntime().availableProcessors()`,即当前运行环境的计算资源可用线程数

public static StreamExecutionEnvironment createRemoteEnvironment(
    			String host, int port, String... jarFiles) { 
    		return new RemoteStreamEnvironment(host, port, jarFiles);
    	}
    
    	public static StreamExecutionEnvironment createRemoteEnvironment(
    			String host, int port, int parallelism, String... jarFiles) { 
    		RemoteStreamEnvironment env = new RemoteStreamEnvironment(host, port, jarFiles);
    		env.setParallelism(parallelism);
    		return env;
    	}
    
    	public static StreamExecutionEnvironment createRemoteEnvironment(
    			String host, int port, Configuration clientConfig, String... jarFiles) { 
    		return new RemoteStreamEnvironment(host, port, clientConfig, jarFiles);
    	}

### 1.3 getExecutionEnvironment ###

创建一个执行环境，表示当前执行程序的上下文。 如果程序是独立调用的，则此方法返回本地执行环境; 如果从命令行客户端调用程序以提交到集群，则此方法返回此集群的执行环境，也就是说，getExecutionEnvironment 会根据查询运行的方 式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式,创建方式如下：

StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

## 2. source 获取数据源 ##

获取数据源的的方式主要有以下五种方式：

1.  `fromCollection`从本地集合获取
2.  `readTextFile`从文件获取
3.  `socketTextStream`从网络套接字获取
4.  `addSource(new FlinkKafkaConsumer<>("kafka-topic", new SimpleStringSchema(), properties));`从kafka获取
5.  `addSource(SourceFunction<OUT> function)`从其他自定义数据源获取

### 2.1 fromCollection从本地集合获取 ###

从集合获取示例代码如下：

ArrayList<Integer> list = new ArrayList<>(Arrays.asList(1,2,3));
            DataStreamSource<Integer> fromCollection = executionEnvironment.fromCollection(list);
            fromCollection.print("collectionStream");

获取内容如下：  
![在这里插入图片描述][20201206210552813.png]

### 2.2 readTextFile从文件获取 ###

从文件获取示例代码如下：

//从文件读取数据
            String filePath = "C:\\DATA\\projectHere\\myGitHub\\flink-demo\\src\\main\\resources\\FlinkData.txt";
            DataStreamSource<String> textFile = executionEnvironment.readTextFile(filePath);
            textFile.print("fileStream");

文件内容如下：  
![在这里插入图片描述][20201206210828325.png]  
输出内容如下：  
![在这里插入图片描述][20201206210850476.png]

### 2.3 socketTextStream 从网络套接字获取 ###

监听socket实时监听数据示例代码如下：

// socket读取数据
            DataStreamSource<String> socketTextStream = executionEnvironment.socketTextStream("localhost", 9000);
            socketTextStream.print("socketStream");

1.  执行命令开放一个socket服务,端口为9000，命令如下：

nc -lk 9000

1.  运行flink程序，
2.  在socket窗口发送信息如下：  
    ![在这里插入图片描述][20201206211815451.png]
3.  flink接收信息如下：  
    ![在这里插入图片描述][20201206211843484.png]

### 2.4 从kafka获取数据 ###

从kafka获取数据源需先引入以下依赖（依赖版本请根据自己实际情况调整）

<dependency>
    			<groupId>org.apache.flink</groupId>
    			<artifactId>flink-connector-kafka_2.11</artifactId>
    			<version>1.11.2</version>
    		</dependency>

从kafka读取的主要代码如下：

//kafka读取数据
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "kafka所在host:端口");
            props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("auto.offset.reset", "latest");
            DataStreamSource<String> kafkaStream = executionEnvironment.addSource(new FlinkKafkaConsumer<>("flink-topic", new SimpleStringSchema(), props));
            kafkaStream.print("kafkaStream");

从kafka获取数据步骤如下：

1.  创建topic,topic名即上文代码中指定的`flink-topic`，创建命令如下：

./bin/kafka-topics.sh --create  --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic flink-topic

1.  启动生产者，生产数据，命令如下：

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic flink-topic

3 . 在生产者控制台窗口发送信息如下：  
![在这里插入图片描述][2020120622405317.png]

1.  flink接收信息如下：  
    ![在这里插入图片描述][20201206224127109.png]

### 2.5 从其他自定义数据源获取 ###

source支持自定义数据源输入，可通过实现`SourceFunction<T>`接口中的`run()`和`cancel()`方法来自定义数据源，这里以生成随机数数据流为例，简单实现一个自定义数据源.

1.  自定义数据源, 实现如下：

package spendreport;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    import java.util.Random;
    
    /** * Created by HE31 on 2020 /12/6 17:12 */
    public class CustomSource implements SourceFunction<Integer> { 
        private  boolean isRun = true;
        @Override
        public void run(SourceContext<Integer> sourceContext) throws Exception { 
            while (isRun){ 
                sourceContext.collect(new Random().nextInt(10));
                Thread.sleep(100);
            }
        }
    
        // 取消方法
        @Override
        public void cancel() { 
            isRun = false;
        }
    }

1.  `addSource()`传入自定义数据源:

CustomSource source = new CustomSource();
            //source.cancel();
            DataStreamSource<Integer> customSource = executionEnvironment.addSource(source);
            customSource.print("customSourceStream");

1.  输出如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 1]

## 3. transform 转换算子 ##

转换算子主要对输入的数据流做处理，是四大模块中最关键的模块，其算子主要有：

1.  map
2.  flatMap
3.  filter
4.  keyBy
5.  滚动聚合算子
    
     *  sum()
     *  min()
     *  max()
     *  minBy()
     *  maxBy()
6.  reduce
7.  split & select
8.  connect & coMap
9.  union  
    下面分别简单介绍各算子的用法，很多例子中会用到一个`WordCount`自定义实体类，这里先行声明，其结构如下：

package spendreport;
    
    
    /** * Created by HE31 on 2020/11/29 18:35 */
    public  class WordCount { 
        private String word;
        private Integer count;
    
        public String getWord() { 
            return word;
        }
    
        public void setWord(String word) { 
            this.word = word;
        }
    
        public Integer getCount() { 
            return count;
        }
    
        public void setCount(Integer count) { 
            this.count = count;
        }
    
        public WordCount(String word, Integer count) { 
            this.word = word;
            this.count = count;
        }
    
        public WordCount() { 
        }
    
        @Override
        public String toString() { 
            return "WordCount{" +
                    "word='" + word + '\'' +
                    ", count=" + count +
                    '}';
        }
    
    }

### 3.1 map ###

map 与java8中的map很类似，都是用来做转换处理  
如将集合1,2,3 转换为`wordCount`对象 ,`word`为 “id\_”+值，`count`为值\*2

ArrayList<Integer> mapList = new ArrayList<>(Arrays.asList(1,2,3));
            SingleOutputStreamOperator<WordCount> map = executionEnvironment.fromCollection(mapList).map(s -> new WordCount("id_" + s, s * 2));
            map.print("tranform-map");

输出如下：  
![在这里插入图片描述][20201206231958147.png]

### 3.2 flatMap ###

flatMap主要是对流中的元素做打平等操作  
如：将`[[1,2],[3,4],[5,6]]`转换为`[1,2,3,4,5,6]`

List<List<Integer>> lists = new ArrayList<>();
            lists.add(Arrays.asList(1,2));
            lists.add(Arrays.asList(3,4));
            lists.add(Arrays.asList(5,6));
            DataStreamSource<List<Integer>> listDataStreamSource = executionEnvironment.fromCollection(lists);
            listDataStreamSource.print("transform-flatMap-before");
            SingleOutputStreamOperator<Integer> flatMapStream = listDataStreamSource.flatMap((List<Integer> l, Collector<Integer> collector) -> { 
                l.forEach(collector::collect);
            }).returns(Types.INT);
            flatMapStream.print("transform-flatMap-after");

输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 2]  
这里需要注意的是lambda表达式需要指定返回类型，不然会报参数类型丢失的异常，具体可看另一篇文章:[Flink学习（三） – Java版流处理工程运行问题记录][Flink_ _ Java]

### 3.3 filter ###

即对流中元素做过滤操作  
如筛选集合`[1,2,3]`中元素`[1]`

ArrayList<Integer> filterList = new ArrayList<>(Arrays.asList(1,2,3));
            SingleOutputStreamOperator<Integer> filter = executionEnvironment.fromCollection(filterList).filter(s -> s == 1);
            filter.print("transform-filter");

输出如下：  
![在这里插入图片描述][20201206233616743.png]

### 3.4 keyBy ###

逻辑地将一个流拆分成不相交的分区，每个分 区包含具有相同 key 的元素，在内部以 hash 的形式实现的,类似sql中的group by  
如：将 `wordCount`根据`word`做分区

ArrayList<WordCount> wordCounts = new ArrayList<>(Arrays.asList(new WordCount("flink", 1)
                    , new WordCount("flink", 1)
                    , new WordCount("java", 1)
                    , new WordCount("sql", 1)
                    , new WordCount("java", 1)));
            KeyedStream<WordCount, String> wordCountStringKeyedStream = executionEnvironment.fromCollection(wordCounts).keyBy(WordCount::getWord);
            wordCountStringKeyedStream.print("transform-keyBy");

输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 3]

### 3.5 滚动聚合算子 ###

通过上一步的`keyBy()`之后，滚动聚合算子可对分组后的组内元素做聚合操作，如：

*  sum()
 *  min()
 *  max()
 *  minBy()
 *  maxBy()

#### 3.5.1 sum ####

对分区元素做求和操作  
如：对wordCount中的count做求和

ArrayList<WordCount> aggWordCounts = new ArrayList<>(Arrays.asList(new WordCount("flink", 1)
                    , new WordCount("flink", 2)
                    , new WordCount("java", 1)
                    , new WordCount("sql", 3)
                    , new WordCount("java", 1)));
            DataStreamSource<WordCount> streamSource = executionEnvironment.fromCollection(aggWordCounts);
            SingleOutputStreamOperator<WordCount> sum = streamSource.keyBy(WordCount::getWord).sum("count");
            sum.print("sum");

输出如下：  
![在这里插入图片描述][20201213155437866.png]

#### 3.5.2 min & minBy ####

min(): 只求对应属性的最小值  
minBy(): 找到对应属性最小值的那条记录  
为了便于观察和区分min（）与minBy（），再定义一个`Person`实体类，如下：

package spendreport;
    
    /** * Created by HE31 on 2020/12/13 15:16 */
    public class Person { 
        /** * 姓名 */
        private String name;
        /** * 性别 */
        private String gender;
        /** * 年龄 */
        private Integer age;
    
        public Person(String name, String gender, Integer age) { 
            this.name = name;
            this.gender = gender;
            this.age = age;
        }
    
        public Person() { 
        }
    
        public String getName() { 
            return name;
        }
    
        public void setName(String name) { 
            this.name = name;
        }
    
        public String getGender() { 
            return gender;
        }
    
        public void setGender(String gender) { 
            this.gender = gender;
        }
    
        public Integer getAge() { 
            return age;
        }
    
        public void setAge(Integer age) { 
            this.age = age;
        }
    
        @Override
        public String toString() { 
            return "Person{" +
                    "name='" + name + '\'' +
                    ", gender='" + gender + '\'' +
                    ", age=" + age +
                    '}';
        }
    }

用min()求不同性别的最小年龄：

List<Person> students = new ArrayList<>(Arrays.asList(new Person("Dave", "male", 18)
                    , new Person("Tom", "male", 20)
                    , new Person("John", "male", 17)
                    , new Person("Sarah", "female", 19)
                    , new Person("Helen", "female", 25)
            ));
            DataStreamSource<Person> personDataStreamSource = executionEnvironment.fromCollection(students);
            SingleOutputStreamOperator<Person> min = personDataStreamSource.keyBy(Person::getGender).min("age");
           min.print("min");

输出如下：  
![在这里插入图片描述][20201213160154371.png]

用minBy()找出不同性别的年龄最小的人

List<Person> students = new ArrayList<>(Arrays.asList(new Person("Dave", "male", 18)
                    , new Person("Tom", "male", 20)
                    , new Person("John", "male", 17)
                    , new Person("Sarah", "female", 19)
                    , new Person("Helen", "female", 25)
            ));
            DataStreamSource<Person> personDataStreamSource = executionEnvironment.fromCollection(students);
            //SingleOutputStreamOperator<Person> min = personDataStreamSource.keyBy(Person::getGender).min("age");
          // min.print("min");
            SingleOutputStreamOperator<Person> minBy = personDataStreamSource.keyBy(Person::getGender).minBy("age");
            minBy.print("minBy");

输出如下：  
![在这里插入图片描述][20201213160517292.png]  
比较min（）与minBy（）的结果：  
min（）  
![在这里插入图片描述][20201213160927446.png]

minBy（）  
![在这里插入图片描述][20201213160816484.png]  
可以看出min（）只是改变了结果集对象中所要求的age属性，并不会改变name和gender属性  
而minBy()的结果集输出的就是即为age最小的整个对象  
max()和maxBy()也同理

#### 3.5.3 reduce ####

对流做规约操作  
如：对WordCount中count做规约求和

ArrayList<WordCount> reduceCounts = new ArrayList<>(Arrays.asList(new WordCount("flink", 1)
                    , new WordCount("flink", 2)
                    , new WordCount("java", 1)
                    , new WordCount("sql", 1)
                    , new WordCount("java", 1)));
            DataStreamSource<WordCount> personDataStreamSource = executionEnvironment.fromCollection(reduceCounts);
    
            SingleOutputStreamOperator<WordCount> reduce = personDataStreamSource.keyBy(WordCount::getWord)
                    .reduce((curWord, nextWord) -> new WordCount(curWord.getWord(),curWord.getCount()+nextWord.getCount()));
            reduce.print("reduce");

输出如下：  
![在这里插入图片描述][20201213170809609.png]

#### 3.5.4 split & select ####

对流做分流筛选  
如：对人群按年龄做筛选，以18岁为分界线，18岁以下为青少年，18岁及以上为成年人

//split & select
            List<Person> students = new ArrayList<>(Arrays.asList(new Person("Dave", "male", 18)
                    , new Person("Tom", "male", 20)
                    , new Person("John", "male", 17)
                    , new Person("Sarah", "female", 19)
                    , new Person("Helen", "female", 25)
            ));
    
            DataStreamSource<Person> personDataStreamSource1 = executionEnvironment.fromCollection(students);
            //打分流标签
            SplitStream<Person> splitStream = personDataStreamSource1.split(person -> { 
                ArrayList<String> tags = new ArrayList<>();
                if (person.getAge() >= 18) { 
    
                    tags.add("adult");
                } else { 
                    tags.add("teen");
                }
                return tags;
            });
            //根据标签筛选
            DataStream<Person> adult = splitStream.select("adult");
            DataStream<Person> teen = splitStream.select("teen");
            DataStream<Person> all = splitStream.select("adult", "teen");
            adult.print("adult");
            teen.print("teen");
            all.print("all");

输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 4]

#### 3.5.5 Connect & CoMap ####

connect可将两种不同数据类型的流合并成一个流，CoMap可对该合并流做处理  
如：把以上`WordCount`和`Person`两个风马牛不相及的流合成一个，再在一个合并流中分别提取`WordCount`流的`word`，`Person`流的`name`

List<Person> coStudent = new ArrayList<>(Arrays.asList(new Person("Dave", "male", 18)
                    , new Person("Tom", "male", 20)
                    , new Person("John", "male", 17)
                    , new Person("Sarah", "female", 19)
                    , new Person("Helen", "female", 25)
            ));
    
            ArrayList<WordCount> coWordCounts = new ArrayList<>(Arrays.asList(new WordCount("flink", 1)
                    , new WordCount("flink", 2)
                    , new WordCount("java", 1)
                    , new WordCount("sql", 1)
                    , new WordCount("java", 1)));
    
            DataStreamSource<Person> personDataStreamSource2 = executionEnvironment.fromCollection(coStudent);
            DataStreamSource<WordCount> wordCountDataStreamSource = executionEnvironment.fromCollection(coWordCounts);
            //连接两个流
            ConnectedStreams<Person, WordCount> connectedStreams = personDataStreamSource2.connect(wordCountDataStreamSource);
            //使用coMap分别提取两条流
            SingleOutputStreamOperator<Object> coMap = connectedStreams.map(new CoMapFunction<Person, WordCount, Object>() { 
                @Override
                public Object map1(Person value) throws Exception { 
                    return value.getName();
                }
    
                @Override
                public Object map2(WordCount value) throws Exception { 
                    return value.getWord();
                }
            });
            coMap.print("coMap");

输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 5]

#### 3.5.6 union ####

union同样也是合流操作，但是与connect相比，主要有两点不同：

1.  union用于合并相同数据类型的流，connect用于合并不同数据类型的流
2.  union可以合并多条流，connect只能合并两条流

如：合并三个班级的同学名单，然后列出所有人的名字：

List<Person> class1 = new ArrayList<>(Arrays.asList(new Person("Dave", "male", 18)
                    , new Person("Tom", "male", 20)
                    , new Person("John", "male", 17)
                    , new Person("Sarah", "female", 19)
                    , new Person("Helen", "female", 25)
            ));
    
            List<Person> class2 = new ArrayList<>(Arrays.asList(new Person("Bob", "male", 19)
                    , new Person("Ted", "male", 23)
                    , new Person("Joe", "male", 27)
                    , new Person("Amy", "female", 24)
                    , new Person("Anne", "female", 25)
            ));
    
            List<Person> class3 = new ArrayList<>(Arrays.asList(new Person("Bill", "male", 23)
                    , new Person("Vivian", "female", 23)
                    , new Person("Ted", "male", 25)
                    , new Person("Emma", "female", 24)
                    , new Person("Lily", "female", 15)
            ));
    
            DataStreamSource<Person> class1Stream = executionEnvironment.fromCollection(class1);
            DataStreamSource<Person> class2Stream = executionEnvironment.fromCollection(class2);
            DataStreamSource<Person> class3Stream = executionEnvironment.fromCollection(class3);
            //合并三个班级
            DataStream<Person> personDataStream = class1Stream.union(class2Stream).union(class3Stream);
            SingleOutputStreamOperator<String> streamOperator = personDataStream.map(person -> person.getName());
            streamOperator.print("union");

输出如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 6]

## 4. sink 结果集下沉 ##

sink即将处理好的数据集输出到外部系统中，flink官方已提供多种外部系统连接器，常用的如：

*  Kafka
 *  ES
 *  JDBC
 *  Redis（由Apache Bahir提供）
 *  HDFS  
    等等，具体如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 7]  
    这里例举其中几种

### 4.1 sink到文件 ###

ArrayList<WordCount> sinkCounts = new ArrayList<>(Arrays.asList(new WordCount("flink", 1)
                    , new WordCount("flink", 2)
                    , new WordCount("java", 1)
                    , new WordCount("sql", 1)
                    , new WordCount("java", 1)));
    
            DataStreamSource<WordCount> collection = executionEnvironment.fromCollection(sinkCounts);
            SingleOutputStreamOperator<WordCount> operator = collection.map(wordCount -> new WordCount("sink_" + wordCount.getWord(), wordCount.getCount()));
            operator.addSink(
                    StreamingFileSink.forRowFormat(new Path("C:\\DATA\\projectHere\\myGitHub\\flink-demo\\src\\main\\resources\\sink")
                            , new SimpleStringEncoder<WordCount>() { 
            }).build()).setParallelism(1);

输出文件如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 8]

### 4.2 sink到kafka ###

1.  这里需要跟从kafka读取数据源一样，先引入相同的依赖：

<dependency>
    			<groupId>org.apache.flink</groupId>
    			<artifactId>flink-connector-kafka_2.11</artifactId>
    			<version>1.11.2</version>
    		</dependency>

1.  flink处理代码如下：

ArrayList<WordCount> sinkCounts = new ArrayList<>(Arrays.asList(new WordCount("flink", 1)
                    , new WordCount("flink", 2)
                    , new WordCount("java", 1)
                    , new WordCount("sql", 1)
                    , new WordCount("java", 1)));
    
            SingleOutputStreamOperator<String> streamOperator = executionEnvironment.fromCollection(sinkCounts)
                    .map(wordCount -> new WordCount("sink_" + wordCount.getWord(), wordCount.getCount()).toString());
            //FlinkKafkaProducer()三个参数分别为：brokerList,topicId,SerializationSchema,这里为方便起见，直接将WordCount对象转为了String
            streamOperator.addSink(new FlinkKafkaProducer<>("localhost:9092", "test-topic", new SimpleStringSchema()));

1.  启动Kafka消费者客户端接收数据：

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092  --topic test-topic

1.  kafka消费者接受数据如下：  
    ![在这里插入图片描述][20201213200935471.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70]: /images/20221120/78170c37073a4b099064227e1fc60966.png
[20201206203704906.png]: https://img-blog.csdnimg.cn/20201206203704906.png
[20201206210552813.png]: https://img-blog.csdnimg.cn/20201206210552813.png
[20201206210828325.png]: https://img-blog.csdnimg.cn/20201206210828325.png
[20201206210850476.png]: https://img-blog.csdnimg.cn/20201206210850476.png
[20201206211815451.png]: https://img-blog.csdnimg.cn/20201206211815451.png
[20201206211843484.png]: https://img-blog.csdnimg.cn/20201206211843484.png
[2020120622405317.png]: https://img-blog.csdnimg.cn/2020120622405317.png
[20201206224127109.png]: https://img-blog.csdnimg.cn/20201206224127109.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20201206225752183.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[20201206231958147.png]: https://img-blog.csdnimg.cn/20201206231958147.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201206232831550.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[Flink_ _ Java]: https://blog.csdn.net/he3more/article/details/110338357
[20201206233616743.png]: https://img-blog.csdnimg.cn/20201206233616743.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201206234001676.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[20201213155437866.png]: https://img-blog.csdnimg.cn/20201213155437866.png
[20201213160154371.png]: https://img-blog.csdnimg.cn/20201213160154371.png
[20201213160517292.png]: https://img-blog.csdnimg.cn/20201213160517292.png
[20201213160927446.png]: https://img-blog.csdnimg.cn/20201213160927446.png
[20201213160816484.png]: https://img-blog.csdnimg.cn/20201213160816484.png
[20201213170809609.png]: https://img-blog.csdnimg.cn/20201213170809609.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20201213173831399.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20201213181830198.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20201213183200532.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20201213193814949.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20201213194725278.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlM21vcmU=,size_16,color_FFFFFF,t_70
[20201213200935471.png]: https://img-blog.csdnimg.cn/20201213200935471.png