title: Flink系列

二、Flink Source 整理和实战

Flink Source 是程序的数据源输入，可以通过 StreamExecutionEnvironment.addSource(sourceFunction) 来为你的程序添加一个 Source。

Flink 提供了大量的已经实现好的 source 方法，也可以自定义 source：

1、通过实现 sourceFunction 接口来自定义无并行度的 source

2、通过实现 ParallelSourceFunction 接口 or 继承 RichParallelSourceFunction 来自定义有并行度的 source

大多数情况下，我们使用自带的 source 即可。

2.1 Flink 内置 Source

关于 Flink 的内置 Source 大致可以分为这四类：

基于 File：readTextFile(path)，读取文本文件，文件遵循 TextInputFormat 读取规则，逐行读取并返回。
基于数据集合：fromCollection(Collection)，通过 java 的 collection 集合创建一个数据流，集合中的所有元素必须是相同类型的。
基于 Socket：socketTextStream(hostname,port)，从 socker 中读取数据，元素可以通过一个分隔符切开。
扩展 Source：addSource() 方法可以实现读取第三方数据源的数据，系统内置提供了一批connectors，连接器会提供对应的 source，比如 Kafka，Pulsar 等

关于这四种类型的 Source 的使用，具体见程序

官网：

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/datastream/overview/#bundled-connectors

官网截图：

在这里插入图片描述

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 (sink)
RabbitMQ (source/sink)
Google PubSub (source/sink)
Hybrid Source (source)
Apache NiFi (source/sink)
Apache Pulsar (source)
Twitter Streaming API (source)
JDBC (sink)

2.1.0 flink程序pom文件添加

依赖如下：

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.14.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.14.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.14.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>1.14.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>1.14.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.14.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web_2.12</artifactId>
            <version>1.14.3</version>
        </dependency>
    </dependencies>

2.1.1 基于 File案例

FlinkSourceReadTextFile.java完整代码如下：

package com.aa.flinkjava.source.builtin;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;

/**
 * @Author AA
 * @Date 2022/2/24 16:35
 * @Project bigdatapre
 * @Package com.aa.flinkjava.source.builtin
 */
public class FlinkSourceReadTextFile {
    public static void main(String[] args) throws Exception {
        //1、初始化环境变量
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();

        //2、读取数据
        DataSource<String> dataSource = executionEnvironment.readTextFile("D://input//test1.txt");

        //3、打印输出
        dataSource.print();

        //4、执行
        //异常：No new data sinks have been defined since the last execution.
        // The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
        //上面异常解决方案：Flink批处理的时候注释掉下面的代码即可。
        //executionEnvironment.execute("FlinkSourceReadTextFile");
    }
}

2.1.2 基于数据集合案例

FlinkSourceFromCollection.java完整代码如下：

package com.aa.flinkjava.source.builtin;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;

import java.util.ArrayList;

/**
 * @Author AA
 * @Date 2022/2/24 16:46
 * @Project bigdatapre
 * @Package com.aa.flinkjava.source.builtin
 */
public class FlinkSourceFromCollection {
    public static void main(String[] args) throws Exception {
        //1、初始化环境变量
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();

        //2、造点数据
        ArrayList<String> list = new ArrayList<>();
        list.add("zhangsan");
        list.add("lisi");
        list.add("wangwu");
        list.add("zhaoliu");

        //3、从数据集合读取数据
        DataSource<String> dataSource = executionEnvironment.fromCollection(list);

        //4、做一个小的业务逻辑
        MapOperator<String, String> result = dataSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                return s + " , 欢迎你！";
            }
        });

        //5、打印输出
        result.print();

        //6、执行 注意，批处理场景下，给下面的依据注释掉。
        //executionEnvironment.execute("FlinkSourceFromCollection");
    }
}

2.1.3 基于 Socket 案例

FlinkSourceSocketTextStream.java 完整代码如下：

package com.aa.flinkjava.source.builtin;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author AA
 * @Date 2022/2/24 16:53
 * @Project bigdatapre
 * @Package com.aa.flinkjava.source.builtin
 */
public class FlinkSourceSocketTextStream {
    public static void main(String[] args) throws Exception {
        //1、初始化环境变量
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        //2、从socket读取数据
        DataStreamSource<String> dataStreamSource = executionEnvironment.socketTextStream("hadoop12", 9999);

        //3、业务逻辑处理
        //3-1 转换给键值对
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = dataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                if (s.isEmpty()){
                    return; //给输入的是空的过滤掉。
                }else {
                    String[] splits = s.split(" ");
                    for (String split : splits) {
                        collector.collect(new Tuple2<>(split, 1));
                    }
                }
            }
        });

        //3-2 累加
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(0).sum(1);

        //4、数据
        result.print();

        //5、提交执行
        executionEnvironment.execute();
    }
}

2.2 Flink 自定义 Source

样例UserDefineSourceDemo.java如下：

package com.aa.flinkjava.source.userdefine;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @Author AA
 * @Date 2022/2/24 17:15
 * @Project bigdatapre
 * @Package com.aa.flinkjava.source.userdefine
 */
public class UserDefineSourceDemo {
    public static void main(String[] args) throws Exception {
        //1、初始化环境变量
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        //2、添加自定义的数据源
        DataStreamSource<Long> dataStreamSource = executionEnvironment.addSource(new UserDefineSource());

        //3、业务逻辑
        SingleOutputStreamOperator<Long> result = dataStreamSource.map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long aLong) throws Exception {
                return aLong + 1000;
            }
        });

        //4、打印结果
        result.print();

        //5、执行
        executionEnvironment.execute();
    }
}

class UserDefineSource implements SourceFunction<Long>{
    private boolean flag = true;

    private long num = 100;

    @Override
    public void run(SourceContext<Long> sourceContext) throws Exception {
        while (flag){
            //每间隔两秒给num数据递增输出。
            sourceContext.collect(num++);
            Thread.sleep(2000);
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }
}

三、Flink Transform 整理和实战

DataStream Transformations 官网链接：

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/overview/

在这里插入图片描述

DataSet Transformations 官网链接：

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/dataset/transformations/

在这里插入图片描述

map 和 filter
flatMap，keyBy、sum
union
connect，coMap

四、Flink Sink 整理和实战

4.1 Flink 内置 Sink

关于 Flink 的内置 Sink 大致可以分为这三类：

1、标准输出/异常输出：print() / printToErr()，打印每个元素的 toString() 方法的值到标准输出或者标准错误输出流中
2、基于文件系统：writeAsText() / writeAsCsv(…) / write() / output()
3、扩展 Sink：常见的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem等

官网链接：

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/datastream/overview/#bundled-connectors

官网截图：

在这里插入图片描述