大数据之Flink流处理 API

向右看齐 2022-08-28 11:40 512阅读 0赞

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16][]

# 1、Environment #

## 1.1 getExecutionEnvironment ##

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

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

如果没有设置并行度，会以 flink-conf.yaml 中的配置为准，默认是 1。

## 1.2 createLocalEnvironment ##

返回本地执行环境，需要在调用时指定默认的并行度。

LocalStreamEnvironment localEnvironment = StreamExecutionEnvironment.createLocalEnvironment(1);

## 1.3 createRemoteEnvironment ##

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

StreamExecutionEnvironment remoteEnvironment = StreamExecutionEnvironment.createRemoteEnvironment("remote_host", 8888, 1);

# 2、Source #

## 2.1 从集合读取数据 ##

package application;
    
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.util.Arrays;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 9:33 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkFromCollection {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStream<Integer> integerDataStream = env.fromCollection(Arrays.asList(1, 4, 3, 7, 8, 9));
            
            integerDataStream.print();
            
            env.execute();
        }
    }

## 2.2 从文件读取数据 ##

package application.function;
    
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 11:25 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkFromFile {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink/src/main/resources/words.txt";
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
            
            inputDataStream.print();
            
            env.execute();
        }
    }

## 2.3 以 kafka 消息队列的数据作为来源 ##

需要引入 kafka 连接器的依赖：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>org.example</groupId>
        <artifactId>Flink1.13.2</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <maven.compiler.source>8</maven.compiler.source>
            <maven.compiler.target>8</maven.compiler.target>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-java</artifactId>
                <version>1.13.2</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.12</artifactId>
                <version>1.13.2</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.12</artifactId>
                <version>1.13.2</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka_2.12</artifactId>
                <version>1.13.2</version>
            </dependency>
    
    
        </dependencies>
    
    </project>

./kafka-console-producer.sh --broker-list aiops02:6667 --topic sensor

package application;
    
    import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
    
    import java.util.Properties;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 10:36 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkFromKafka {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers", "192.168.48.12:6667");
            properties.setProperty("group.id", "consumer-group");
            properties.setProperty("key.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("auto.offset.reset", "latest");
            DataStream<String> inputDataStream = env.addSource(new FlinkKafkaConsumer011<String>("sensor",new SimpleStringSchema(),properties));
    
            inputDataStream.print();
            env.execute();
        }
    }

## 2.4 自定义 Source ##

除了以上的 source 数据来源，我们还可以自定义 source。需要做的，只是传入一个 SourceFunction 就可以。具体调用如下：

DataStreamSource<SensorReading> inputDataStream = env.addSource(new MySensorSource());

我们希望可以随机生成传感器数据， MySensorSource 具体的代码实现如下：

package application.application.entity;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 11:39 上午
     * @Version: 1.0
     * @Description:
     */
    public class SensorReading {
        /**
         * sensor id
         */
        private String sensorId;
        /**
         * 当前时间的时间戳
         */
        private long currentMillis;
        /**
         * sensor监控的值
         */
        private double value;
    
        public SensorReading() {
        }
    
        public SensorReading(String sensorId, long currentMillis, double value) {
            this.sensorId = sensorId;
            this.currentMillis = currentMillis;
            this.value = value;
        }
    
        public String getSensorId() {
            return sensorId;
        }
    
        public void setSensorId(String sensorId) {
            this.sensorId = sensorId;
        }
    
        public long getCurrentMillis() {
            return currentMillis;
        }
    
        public void setCurrentMillis(long currentMillis) {
            this.currentMillis = currentMillis;
        }
    
        public double getValue() {
            return value;
        }
    
        public void setValue(double value) {
            this.value = value;
        }
    
        @Override
        public String toString() {
            return "SensorReading{" +
                    "sensorId='" + sensorId + '\'' +
                    ", currentMillis=" + currentMillis +
                    ", value=" + value +
                    '}';
        }
    }

package application;
    
    import application.application.entity.SensorReading;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 11:38 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkMySource {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStreamSource<SensorReading> inputDataStream = env.addSource(new MySensorSource());
    
            inputDataStream.print();
    
            env.execute();
        }
    
    
        public static class MySensorSource implements SourceFunction<SensorReading>{
    
            private volatile boolean running = true;
    
            @Override
            public void run(SourceContext<SensorReading> sourceContext) throws Exception {
                Random random = new Random();
                Map<String,Double> sensorMap = new HashMap<>();
                for (int i = 0; i < 10; i++) {
                    sensorMap.put("sensor_" + (i+1),random.nextGaussian());
                }
                while (running){
                    for (String sensorId : sensorMap.keySet()){
                        double value = sensorMap.get(sensorId) + random.nextGaussian();
                        sensorMap.put(sensorId,value);
                        sourceContext.collect(new SensorReading(sensorId,System.currentTimeMillis(),value));
                    }
    
                    TimeUnit.SECONDS.sleep(1);
                }
            }
    
            @Override
            public void cancel() {
                running = false;
            }
        }
    }

# 3、Transform #

转换算子

## 3.1 map ##

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

DataStream<Integer> inputDataStream = env.fromCollection(Arrays.asList(1, 4, 3, 8, 0, 9, 10, 6));
            DataStream<Boolean> mapDataStream = inputDataStream.map(new MapFunction<Integer, Boolean>() {
                @Override
                public Boolean map(Integer i) throws Exception {
                    return i > 5;
                }
            });

## 3.2 flatMap ##

## 3.3 Filter ##

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 2][]

DataStream<String> inputDataStream = env.fromCollection(Arrays.asList(
                    "hello world",
                    "how are you",
                    "hello kitty"
            ));
            DataStream<Tuple2<String,Integer>> flatMapDataStream = inputDataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                    String[] words = line.split(" ");
                    Arrays.asList(words).forEach(word->{
                        collector.collect(new Tuple2<>(word,1));
                    });
                }
            });
    
            DataStream<Tuple2<String, Integer>> filterDataStream = flatMapDataStream.filter(new FilterFunction<Tuple2<String, Integer>>() {
                @Override
                public boolean filter(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                    return stringIntegerTuple2.f0.length() > 3;
                }
            });

## 3.4 KeyBy ##

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 3][]

**DataStream → KeyedStream**：逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同 key 的元素，在内部以 hash 的形式实现的。

sensor.txt：

sensor_01,1632814080332,3.02
    sensor_03,1632814080345,3.12
    sensor_05,1632814081236,2.32
    sensor_06,1632814081234,3.57
    sensor_01,1632814087652,3.38
    sensor_01,1632814080389,3.99

package application;
    
    import application.application.entity.SensorReading;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 4:24 下午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkTransformAggregation {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink/src/main/resources/sensor.txt";
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<SensorReading> sensorReadingDataStream = inputDataStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0],Long.parseLong(fields[1]),Double.parseDouble(fields[2]));
            });
    
            KeyedStream<SensorReading, String> keyedStream = sensorReadingDataStream.keyBy(SensorReading::getSensorId);
    
    //        DataStream<SensorReading> maxBy = keyedStream.max("value");
            //这种方式会刷新对象
            DataStream<SensorReading> maxByDataStream = keyedStream.maxBy("value");
    
            maxByDataStream.print();
    
            env.execute();
        }
    }

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 4][]

##  3.5 滚动聚合算子（Rolling Aggregation） ##

这些算子可以针对 KeyedStream 的每一个支流做聚合。

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

## 3.6 Reduce ##

**KeyedStream → DataStream**：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。

package application;
    
    import application.application.entity.SensorReading;
    import org.apache.flink.api.common.functions.ReduceFunction;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 5:22 下午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkTransformAggregationReduce {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink/src/main/resources/sensor.txt";
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<SensorReading> sensorReadingDataStream = inputDataStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0],Long.parseLong(fields[1]),Double.parseDouble(fields[2]));
            });
    
            //
            KeyedStream<SensorReading, String> keyedStream = sensorReadingDataStream.keyBy(SensorReading::getSensorId);
    
    //        DataStream<SensorReading> maxBy = keyedStream.max("value");
            //这种方式会刷新对象
            DataStream<SensorReading> reduceDataStream = keyedStream.reduce(new ReduceFunction<SensorReading>() {
                @Override
                public SensorReading reduce(SensorReading s1, SensorReading s2) throws Exception {
                    return new SensorReading(s1.getSensorId(),s1.getValue() > s2.getValue() ? s1.getCurrentMillis(): s2.getCurrentMillis(),Math.max(s1.getValue(),s2.getValue()));
                }
            });
    
            reduceDataStream.print();
    
            env.execute();
        }
    }

## 3.7 Split 和 Select ##

### Split ###

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 5][]

**DataStream → SplitStream**：根据某些特征把一个 DataStream 拆分成两个或者多个 DataStream。

### Select   ![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 6][] ###

**SplitStream→ DataStream**：从一个 SplitStream 中获取一个或者多个DataStream。

需求： 传感器数据按照温度高低（以 30 度为界），拆分成两个流。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 7][]

package application;
    
    import application.application.entity.SensorReading;
    import org.apache.flink.streaming.api.collector.selector.OutputSelector;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SplitStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.util.Collections;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 5:46 下午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkTransformSplitSelect {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink/src/main/resources/sensor.txt";
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<SensorReading> sensorReadingDataStream = inputDataStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0],Long.parseLong(fields[1]),Double.parseDouble(fields[2]));
            });
    
            //分流
            SplitStream<SensorReading> splitStream = sensorReadingDataStream.split(new OutputSelector<SensorReading>() {
                @Override
                public Iterable<String> select(SensorReading sensorReading) {
                    return sensorReading.getValue() > 3.3 ? Collections.singletonList("high") : Collections.singletonList("low");
                }
            });
    
            DataStream<SensorReading> highDataStream = splitStream.select("high");
            DataStream<SensorReading> lowDataStream = splitStream.select("low");
            DataStream<SensorReading> allDataStream = splitStream.select("high", "low");
    
            highDataStream.print("high");
            lowDataStream.print("low");
            allDataStream.print("all");
    
            env.execute();
        }
    }

## 3.8 Connect 和 CoMap ##

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 8][]

**DataStream,DataStream → ConnectedStreams**：连接两个保持他们类型的数据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

### CoMap,CoFlatMap ###

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 9][]

**ConnectedStreams → DataStream**：作用于 ConnectedStreams 上，功能与 map和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap处理。

package application;
    
    import application.application.entity.SensorReading;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.api.java.tuple.Tuple3;
    import org.apache.flink.streaming.api.collector.selector.OutputSelector;
    import org.apache.flink.streaming.api.datastream.ConnectedStreams;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.datastream.SplitStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.co.CoMapFunction;
    
    import java.util.Collections;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/28 5:46 下午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkTransformSplitSelect {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink/src/main/resources/sensor.txt";
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<SensorReading> sensorReadingDataStream = inputDataStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0],Long.parseLong(fields[1]),Double.parseDouble(fields[2]));
            });
    
            //分流
            SplitStream<SensorReading> splitStream = sensorReadingDataStream.split(new OutputSelector<SensorReading>() {
                @Override
                public Iterable<String> select(SensorReading sensorReading) {
                    return sensorReading.getValue() > 3.3 ? Collections.singletonList("high") : Collections.singletonList("low");
                }
            });
    
            DataStream<SensorReading> highDataStream = splitStream.select("high");
            DataStream<SensorReading> lowDataStream = splitStream.select("low");
            DataStream<SensorReading> allDataStream = splitStream.select("high", "low");
    
    //        highDataStream.print("high");
    //        lowDataStream.print("low");
    //        allDataStream.print("all");
    
            DataStream<Tuple2<String, Double>> highResultDataStream = highDataStream.map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
                @Override
                public Tuple2<String, Double> map(SensorReading sensorReading) throws Exception {
                    return new Tuple2<String, Double>(sensorReading.getSensorId(), sensorReading.getValue());
                }
            });
    
            ConnectedStreams<Tuple2<String, Double>, SensorReading> connectedStreams = highResultDataStream.connect(lowDataStream);
    
            DataStream<Object> resultStream = connectedStreams.map(new CoMapFunction<Tuple2<String, Double>, SensorReading, Object>() {
                @Override
                public Object map1(Tuple2<String, Double> value) throws Exception {
                    return new Tuple3<>(value.f0,value.f1,"high warn");
                }
    
                @Override
                public Object map2(SensorReading value) throws Exception {
                    return new Tuple2<>(value.getSensorId(),value.getValue());
                }
            });
    
            resultStream.print("connect stream");
    
            env.execute();
        }
    }

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 10][]

##  3.9 Union ##

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 11][]

** DataStream → DataStream**：对两个或者两个以上的 DataStream 进行 union 操作，产生一个包含所有 DataStream 元素的新 DataStream。

DataStream<SensorReading> unionStream = highDataStream.union(lowDataStream);

Connect 与 Union 区别：

1． Union 之前两个流的类型必须是一样， Connect 可以不一样，在之后的 coMap中再去调整成为一样的。  
2. Connect 只能操作两个流， Union 可以操作多个。

# 4、支持的数据类型 #

Flink 流应用程序处理的是以数据对象表示的事件流。 所以在 Flink 内部， 我们需要能够处理这些对象。它们需要被序列化和反序列化，以便通过网络传送它们；或者从状态后端、检查点和保存点读取它们。为了有效地做到这一点， Flink 需要明确知道应用程序所处理的数据类型。 Flink 使用类型信息的概念来表示数据类型，并为每个数据类型生成特定的序列化器、反序列化器和比较器。  
Flink 还具有一个类型提取系统，该系统分析函数的输入和返回类型，以自动获取类型信息，从而获得序列化器和反序列化器。但是，在某些情况下，例如 lambda函数或泛型类型，需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。  
Flink 支持 Java 和 Scala 中所有常见数据类型。使用最广泛的类型有以下几种。

## 4.1 基础数据类型 ##

Flink 支持所有的 Java 和 Scala 基础数据类型， Integer, Double, Long, String, …

DataStream<Integer> numberStream = env.fromElements(1, 2, 3, 4);
    numberStream.map(data -> data * 2);

## 4.2 Java 和 Scala 元组（Tuples） ##

DataStream<Tuple2<String, Integer>> personStream = env.fromElements(
    new Tuple2("Adam", 17),
    new Tuple2("Sarah", 23) );
    personStream.filter(p -> p.f1 > 18);

## 4.3 Scala 样例类（case classes） ##

case class Person(name: String, age: Int)
    val persons: DataStream[Person] = env.fromElements(
    Person("Adam", 17),
    Person("Sarah", 23) )
    persons.filter(p => p.age > 18)

## 4.4 Java 简单对象（POJOs） ##

## 4.5 其它（Arrays, Lists, Maps, Enums, 等等） ##

Flink 对 Java 和 Scala 中的一些特殊目的的类型也都是支持的，比如 Java 的ArrayList， HashMap， Enum 等等。

# 5、实现 UDF 函数——更细粒度的控制流 #

## 5.1 函数类（Function Classes） ##

Flink 暴露了所有 udf 函数的接口(实现方式为接口或者抽象类)。例如MapFunction, FilterFunction, ProcessFunction 等等。  
下面例子实现了 FilterFunction 接口：

DataStream<String> flinkTweets = tweets.filter(new FlinkFilter());
            public static class FlinkFilter implements FilterFunction<String> {
                @Override
                public boolean filter(String value) throws Exception {
                    return value.contains("flink");
                }
            }

还可以将函数实现成匿名类

DataStream<String> flinkTweets = tweets.filter(new FilterFunction<String>() {
                @Override
                public boolean filter(String value) throws Exception {
                    return value.contains("flink");
                }
            });

我们 filter 的字符串"flink"还可以当作参数传进去。

DataStream<String> tweets = env.readTextFile("INPUT_FILE ");
            DataStream<String> flinkTweets = tweets.filter(new KeyWordFilter("flink"));
            public static class KeyWordFilter implements FilterFunction<String> {
                private String keyWord;
                KeyWordFilter(String keyWord) { this.keyWord = keyWord; }
                @Override
                public boolean filter(String value) throws Exception {
                    return value.contains(this.keyWord);
                }
            }

## 5.2 匿名函数（Lambda Functions） ##

DataStream<String> tweets = env.readTextFile("INPUT_FILE");
    DataStream<String> flinkTweets = tweets.filter( tweet -> tweet.contains("flink") );

## 5.3 富函数（Rich Functions） ##

“富函数”是 DataStream API 提供的一个函数类的接口， 所有 Flink 函数类都有其 Rich 版本。 它与常规函数的不同在于，可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

*  RichMapFunction
 *  RichFlatMapFunction
 *  RichFilterFunction
 *  …

Rich Function 有一个生命周期的概念。 典型的生命周期方法有：

*  open()方法是 rich function 的初始化方法，当一个算子例如 map 或者 filter被调用之前 open()会被调用。
 *  close()方法是生命周期中的最后一个调用的方法，做一些清理工作。
 *  getRuntimeContext()方法提供了函数的 RuntimeContext 的一些信息，例如函数执行的并行度，任务的名字，以及 state 状态

public static class MyRichFunction extends RichMapFunction<String,SensorReading>{
    
            @Override
            public SensorReading map(String s) throws Exception {
                return new SensorReading(getRuntimeContext().getTaskName(), getRuntimeContext().getIndexOfThisSubtask(), 0.0);
            }
    
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("open");
            }
    
            @Override
            public void close() throws Exception {
                System.out.println("close");
            }
        }

# 6、Sink #

Flink 没有类似于 spark 中 foreach 方法，让用户进行迭代的操作。虽有对外的输出操作都要利用 Sink 完成。最后通过类似如下方式完成整个任务最终输出操作。

stream.addSink(new MySink(xxxx));

官方提供了一部分的框架的 sink。除此以外，需要用户自定义实现 sink。

# DataStream Connectors [\#][Link 1] #

## Predefined Sources and Sinks [\#][Link 2] ##

A few basic data sources and sinks are built into Flink and are always available. The [predefined data sources][] include reading from files, directories, and sockets, and ingesting data from collections and iterators. The [predefined data sinks][] support writing to files, to stdout and stderr, and to sockets.

## Bundled Connectors [\#][Link 3] ##

Connectors provide code for interfacing with various third-party systems. Currently these systems are supported:

*  [Apache Kafka][] (source/sink)
 *  [Apache Cassandra][] (sink)
 *  [Amazon Kinesis Streams][] (source/sink)
 *  [Elasticsearch][] (sink)
 *  [FileSystem (Hadoop included) - Streaming only][FileSystem _Hadoop included_ - Streaming only] (sink)
 *  [FileSystem (Hadoop included) - Streaming and Batch][FileSystem _Hadoop included_ - Streaming and Batch] (sink)
 *  [RabbitMQ][] (source/sink)
 *  [Apache NiFi][] (source/sink)
 *  [Twitter Streaming API][] (source)
 *  [Google PubSub][] (source/sink)
 *  [JDBC][] (sink)

Keep in mind that to use one of these connectors in an application, additional third party components are usually required, e.g. servers for the data stores or message queues. Note also that while the streaming connectors listed in this section are part of the Flink project and are included in source releases, they are not included in the binary distributions. Further instructions can be found in the corresponding subsections.

## Connectors in Apache Bahir [\#][Link 4] ##

Additional streaming connectors for Flink are being released through [Apache Bahir][], including:

*  [Apache ActiveMQ][] (source/sink)
 *  [Apache Flume][] (sink)
 *  [Redis][] (sink)
 *  [Akka][] (sink)
 *  [Netty][] (source)

## 6.1 Kafka ##

<dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka_2.12</artifactId>
                <version>1.13.2</version>
            </dependency>

主函数中添加 sink：

package application;
    
    import application.entity.SensorReading;
    import org.apache.flink.api.common.functions.RichMapFunction;
    import org.apache.flink.api.common.serialization.SerializationSchema;
    import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.sink.SinkFunction;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
    import org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/29 10:04 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkTransformAggregation {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink1.13.2/src/main/resources/sensor.txt";
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<String> mapDataStream = inputDataStream.map(new RichMapFunction<String, String>() {
                @Override
                public String map(String line) throws Exception {
                    String[] fields = line.split(",");
    
                    return new SensorReading(fields[0], Long.parseLong(fields[1]), Double.parseDouble(fields[2])).toString();
                }
            });
    
            mapDataStream.addSink(new FlinkKafkaProducer<String>("192.168.48.12:6667", "sensorSink", new SimpleStringSchema()));
    
    
    
            env.execute();
        }
    }

[root@aiops03 kafka]# bin/kafka-console-consumer.sh --bootstrap-server  aiops01:6667,aiops02:6667,aiops03:6667 --topic sensorSink
    SensorReading{sensorId='sensor_05', currentMillis=1632814081236, value=2.32}
    SensorReading{sensorId='sensor_01', currentMillis=1632814080389, value=3.99}
    SensorReading{sensorId='sensor_01', currentMillis=1632814087652, value=3.38}
    SensorReading{sensorId='sensor_03', currentMillis=1632814080345, value=3.12}
    SensorReading{sensorId='sensor_06', currentMillis=1632814081234, value=3.57}
    SensorReading{sensorId='sensor_01', currentMillis=1632814080332, value=3.02}

kafka作为source和sink的示例：

package application;
    
    import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
    
    import java.util.Properties;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/29 10:04 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkTransformAggregation {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers", "192.168.48.12:6667");
            properties.setProperty("group.id", "consumer-group");
            properties.setProperty("key.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("auto.offset.reset", "latest");
    
            DataStream<String> inputDataStream = env.addSource(new FlinkKafkaConsumer<String>("sensor", new SimpleStringSchema(), properties));
    
            inputDataStream.addSink(new FlinkKafkaProducer<String>("192.168.48.12:6667", "sensorSink", new SimpleStringSchema()));
    
            env.execute();
        }
    }

## 6.2 Redis  ##

pom.xml：

<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
    </dependency>

定义一个 redis 的 mapper 类，用于定义保存到 redis 时调用的命令：

package application;
    
    import application.entity.SensorReading;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.redis.RedisSink;
    import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
    import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
    import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
    import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/29 3:48 下午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkSinkRedis {
    
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink1.13.2/src/main/resources/sensor.txt";
            
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<SensorReading> mapDataStream = inputDataStream.map(new MapFunction<String, SensorReading>() {
                @Override
                public SensorReading map(String line) throws Exception {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0],Long.parseLong(fields[1]),Double.parseDouble(fields[2]));
                }
            });
            
            FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
                    .setHost("localhost")
                    .setPort(6379)
                    .build();
            
            mapDataStream.addSink(new RedisSink<SensorReading>(config,new MyRedisMapper()));
            
            env.execute();
        }
        
        public static class MyRedisMapper implements RedisMapper<SensorReading>{
    
            @Override
            public RedisCommandDescription getCommandDescription() {
                //创建一个描述，该描述指定了使用hset，key的名称为mySet
                return new RedisCommandDescription(RedisCommand.HSET,"mySet");
            }
    
            @Override
            public String getKeyFromData(SensorReading sensorReading) {
                return sensorReading.getSensorId();
            }
    
            @Override
            public String getValueFromData(SensorReading sensorReading) {
                return String.valueOf(sensorReading.getValue());
            }
        }
    }

## 6.3 Elasticsearch ##

pom.xml：

<dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-elasticsearch7_2.12</artifactId>
                <version>1.13.2</version>
            </dependency>

package application.sink;
    
    import application.entity.SensorReading;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.common.functions.RuntimeContext;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
    import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
    import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
    import org.apache.http.HttpHost;
    import org.elasticsearch.action.index.IndexRequest;
    import org.elasticsearch.client.Requests;
    
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/30 9:56 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkSinkElasticSearch {
    
        public static void main(String[] args) throws Exception{
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink1.13.2/src/main/resources/sensor.txt";
    
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<SensorReading> mapDataStream = inputDataStream.map(new MapFunction<String, SensorReading>() {
                @Override
                public SensorReading map(String line) throws Exception {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0],Long.parseLong(fields[1]),Double.parseDouble(fields[2]));
                }
            });
    
            List<HttpHost> httpHosts = new ArrayList<>();
            HttpHost httpHost = new HttpHost("192.168.48.12",9200);
    
            httpHosts.add(httpHost);
    
            mapDataStream.addSink(new ElasticsearchSink.Builder<SensorReading>(httpHosts, new MyElasticSearchSinkFunction()).build());
    
            env.execute();
        }
    
    
        public static class MyElasticSearchSinkFunction implements ElasticsearchSinkFunction<SensorReading>{
    
            @Override
            public void process(SensorReading sensorReading, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                Map<String,Object> dataSource = new HashMap<>();
                dataSource.put("id",sensorReading.getSensorId());
                dataSource.put("currentMillis",sensorReading.getCurrentMillis());
                dataSource.put("value",sensorReading.getValue());
    
                IndexRequest indexRequest = Requests.indexRequest();
    
                indexRequest.index("sensor")
                        .source(dataSource);
    
                //用requestIndex发送请求
                requestIndexer.add(indexRequest);
            }
        }
    }

## 6.4 JDBC 自定义 sink ##

pom.xml：

<dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>8.0.11</version>
            </dependency>

添加 MyJdbcSink：

package application.sink;
    
    import application.entity.SensorReading;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.PreparedStatement;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/9/30 10:26 上午
     * @Version: 1.0
     * @Description:
     */
    public class FlinkSinkJDBC {
    
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            String inputPath = "/Users/tlzs/Desktop/working/IdeaProjects/Flink1.13.2/src/main/resources/sensor.txt";
    
            DataStream<String> inputDataStream = env.readTextFile(inputPath);
    
            DataStream<SensorReading> mapDataStream = inputDataStream.map(new MapFunction<String, SensorReading>() {
                @Override
                public SensorReading map(String line) throws Exception {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0],Long.parseLong(fields[1]),Double.parseDouble(fields[2]));
                }
            });
    
            mapDataStream.addSink(new MyJdbcSink());
    
            env.execute();
        }
    
        public static class MyJdbcSink extends RichSinkFunction<SensorReading> {
            private Connection connection = null;
            private PreparedStatement insertStatement = null;
    
            @Override
            public void open(Configuration parameters) throws Exception {
                connection = DriverManager.getConnection("jdbc:mysql://47.104.9.61:3306/bigdata?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=UTC&useSSL=false&connectTimeout=60000000&socketTimeout=60000000","root","wtl199201180271");
                insertStatement = connection.prepareStatement("insert into sensor values(?,?,?)");
            }
    
            @Override
            public void invoke(SensorReading sensorReading, Context context) throws Exception {
                insertStatement.setString(1,sensorReading.getSensorId());
                insertStatement.setLong(2,sensorReading.getCurrentMillis());
                insertStatement.setDouble(3,sensorReading.getValue());
                insertStatement.execute();
            }
    
            @Override
            public void close() throws Exception {
                insertStatement.close();
                connection.close();
            }
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/5df875ca6c444b9c9270b62920aa6ff4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/6fee7ce136c14f188126b98f9cd24026.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/6b07a050071e4ac6b6a255f56bcd8a24.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220828/b740e75b9dce4d27bd924c88fb0c68e3.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220828/227224ed81ed4621b7e7cc56d4d0f8b8.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220828/31baca3b89d5453f85fdededf4c32269.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 6]: /images/20220828/c4bf457e0eb748eb9b1b5aa5035db8f5.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 7]: /images/20220828/a555302bfbb4468aa63e46e25ac6bdae.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 8]: /images/20220828/5e4546377105418dbdda0cc67b59db0b.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 9]: /images/20220828/23f939ba64014f5ea66246be72d3fc7d.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 10]: /images/20220828/d9ddb140037a4bd0a1a5aa9681a3b94b.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16 11]: /images/20220828/841fffcb5e084b599785d21b6125e356.png
[Link 1]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/overview/#datastream-connectors
[Link 2]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/overview/#predefined-sources-and-sinks
[predefined data sources]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/datastream/overview/#data-sources
[predefined data sinks]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/datastream/overview/#data-sinks
[Link 3]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/overview/#bundled-connectors
[Apache Kafka]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/
[Apache Cassandra]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/cassandra/
[Amazon Kinesis Streams]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/kinesis/
[Elasticsearch]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/elasticsearch/
[FileSystem _Hadoop included_ - Streaming only]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/streamfile_sink/
[FileSystem _Hadoop included_ - Streaming and Batch]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/file_sink/
[RabbitMQ]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/rabbitmq/
[Apache NiFi]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/nifi/
[Twitter Streaming API]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/twitter/
[Google PubSub]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/pubsub/
[JDBC]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/jdbc/
[Link 4]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/overview/#connectors-in-apache-bahir
[Apache Bahir]: https://bahir.apache.org/
[Apache ActiveMQ]: https://bahir.apache.org/docs/flink/current/flink-streaming-activemq/
[Apache Flume]: https://bahir.apache.org/docs/flink/current/flink-streaming-flume/
[Redis]: https://bahir.apache.org/docs/flink/current/flink-streaming-redis/
[Akka]: https://bahir.apache.org/docs/flink/current/flink-streaming-akka/
[Netty]: https://bahir.apache.org/docs/flink/current/flink-streaming-netty/