前言

Flink 可以从各种来源获取数据，然后构建 DataStream 进行转换处理。一般将数据的输入来源称为数据源(data source)，而读取数据的算子就是源算子（source operator）。所以，source就是我们整个处理程序的输入端。

Flink 代码中通用的添加 source 的方式，是调用执行环境的 addSource()方法：

DataStream<String> stream = env.addSource(...);

方法传入一个对象参数，需要实现 SourceFunction 接口；返回 DataStreamSource。这里的 DataStreamSource 类继承自 SingleOutputStreamOperator 类，又进一步继承自 DataStream。所以很明显，读取数据的 source 操作是一个算子，得到的是一个数据流（DataStream）。

传入的参数是一个“源函数”（source function），需要实现SourceFunction 接口。

Flink 直接提供了很多预实现的接口，此外还有很多外部连接工具也帮我们实现了对应的 source function，通常情况下足以应对我们的实际需求。

Flink 已实现的Source：https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/overview/

准备工作

为了更好地理解，我们先构建一个实际应用场景。比如网站的访问操作，可以抽象成一个三元组（用户名，用户访问的 urrl，用户访问 url 的时间戳），所以在这里，我们可以创建一个类 Event，将用户行为包装成它的一个对象。Event 包含了以下一些字段。

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Event {

    private String user;

    private String url;

    private Long timestamp;
    
}

导入相关maven依赖

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.15.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>1.15.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>1.15.0</version>
    </dependency>

    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.36</version>
    </dependency>

    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.36</version>
    </dependency>

    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-to-slf4j</artifactId>
        <version>2.17.2</version>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.24</version>
    </dependency>
</dependencies>

一、从集合中读取数据

public class SourceTest {

    public static void main(String[] args) throws Exception {
        readCollection();
    }

    /**
     * 从集合中读取数据
     * @throws Exception
     */
    private static void readCollection() throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<Event> list = new ArrayList<>();
        list.add(new Event("Mary","./home",1000L));
        list.add(new Event("Bob","./cart",2000L));
        //从集合中读取数据
        DataStreamSource<Event> dataStream = env.fromCollection(list);

        dataStream.print();

        env.execute();
    }
}

二、从元素中读取数据

public class SourceTest {

    public static void main(String[] args) throws Exception {
        readElement();
    }

    /**
     * 从元素中读取数据
     * @throws Exception
     */
    private static void readElement() throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //从元素中读取数据
        DataStreamSource<Event> dataStream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );

        dataStream.print();

        env.execute();
    }
}

三、从文件中读取数据

真正的实际应用中，自然不会直接将数据写在代码中。通常情况下，我们会从存储介质中获取数据，一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。

说明:

• 参数可以是目录，也可以是文件；
• 路径可以是相对路径，也可以是绝对路径；
• 相对路径是从系统属性 user.dir 获取路径: idea 下是 project 的根目录, standalone 模式
  下是集群节点根目录；
• 也可以从 hdfs 目录下读取, 使用路径 hdfs://..., 由于 Flink 没有提供 hadoop 相关依赖,
  需要 pom 中添加相关依赖:

public class SourceTest {

    public static void main(String[] args) throws Exception {
        readFile();
    }

    /**
     * 从文件中读取数据
     * @throws Exception
     */
    private static void readFile() throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //从文件中读取数据
        DataStreamSource<String> dataStream = env.readTextFile("src/main/resources/clicks.txt");

        dataStream.print();

        env.execute();
    }
}

四、从hdfs中读取数据

引入依赖

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.3.3</version>
</dependency>

public class SourceTest {

    public static void main(String[] args) throws Exception {
        readHdfs();
    }
    
    /**
     * 从hdfs中读取数据
     * @throws Exception
     */
    private static void readHdfs() throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取hdfs文件路径
        DataStreamSource<String> hdfsSource = env.readTextFile("hdfs://192.168.111.188:8020/input/README.txt");

        //将hdfs文件路径打印输出
        hdfsSource.print();

        env.execute();
    }   
}

五、从socket中读取数据

public class SourceTest {

    public static void main(String[] args) throws Exception {
        readSocket();
    }

    /**
     * 从socket中读取数据
     * @throws Exception
     */
    private static void readSocket() throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取文本流, 监听linux主机端口, linux通过nc -lk 7777发送文本
        DataStreamSource<String> dataStreamSource = env.socketTextStream("192.168.111.188",7777);

        dataStreamSource.print();

        env.execute();
    }
}

六、从kafka中读取数据

Kafka 作为分布式消息传输队列，是一个高吞吐、易于扩展的消息系统。而消息队列的传输方式，恰恰和流处理是完全一致的。所以可以说 Kafka 和 Flink 天生一对，是当前处理流式数据的双子星。在如今的实时流处理应用中，由 Kafka 进行数据的收集和传输，Flink 进行分析计算，这样的架构已经成为众多企业的首选。

Flink官方提供了连接工具flink-connector-kafka，直接帮我们实现了一个消费者FlinkKafkaConsumer，它就是用来读取 Kafka 数据的SourceFunction。

所以想要以 Kafka 作为数据源获取数据，我们只需要引入 Kafka 连接器的依赖。Flink 官方提供的是一个通用的 Kafka 连接器，它会自动跟踪最新版本的 Kafka 客户端。目前最新版本只支持 0.10.0 版本以上的 Kafka，读者使用时可以根据自己安装的 Kafka 版本选定连接器的依赖版本。这里我们需要导入的依赖如下。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.15.0</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-hadoop-compatibility_2.12</artifactId>
    <version>1.15.0</version>
</dependency>

public class SourceTest {

    public static void main(String[] args) throws Exception {
        readKafka();
    }

    /**
     * 从kafka中读取数据
     * @throws Exception
     */
    private static void readKafka() throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.111.188:9092");
        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");
        DataStreamSource<String> kafkaDataStream = env.addSource(new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(), properties));

        kafkaDataStream.print();

        env.execute();
    }
}

创建 FlinkKafkaConsumer 时需要传入三个参数：

• 第一个参数 topic，定义了从哪些主题中读取数据。可以是一个 topic，也可以是 topic
  列表，还可以是匹配所有想要读取的 topic 的正则表达式。当从多个 topic 中读取数据
  时，Kafka 连接器将会处理所有 topic 的分区，将这些分区的数据放到一条流中去。
• 第二个参数是一个 DeserializationSchema 或者 KeyedDeserializationSchema。Kafka 消
  息被存储为原始的字节数据，所以需要反序列化成 Java 或者 Scala 对象。上面代码中
  使用的 SimpleStringSchema，是一个内置的 DeserializationSchema，它只是将字节数
  组简单地反序列化成字符串。DeserializationSchema 和 KeyedDeserializationSchema 是公共接口，所以我们也可以自定义反序列化逻辑。
• 第三个参数是一个 Properties 对象，设置了 Kafka 客户端的一些属性。

七、从Pulsar中读取数据

随着数据日益膨胀，采用事件流处理数据至关重要。Apache Flink 将批流处理统一到计算引擎中，提供了一致化的编程接口。Apache Pulsar（与 Apache BookKeeper 一起）以 "流 "的方式统一数据。在 Pulsar 中，数据存储成一个副本，以流（streaming）（通过 pub-sub 接口）和 segment（用于批处理）的方式进行访问。Pulsar 解决了企业在使用不同的存储和消息技术解决方案时遇到的数据孤岛问题。

Flink 可以直接与 Pulsar broker 进行实时的流式读写，同时 Flink 也可以批量读取 Pulsar 底层离线存储，与 BookKeeper 的内容进行批次读写。同时支持批流，使得 Pulsar 和 Flink 先天就是契合的伙伴。把 Flink 和 Pulsar 结合使用，这两种开源技术可以创建一个统一的数据架构，为实时数据驱动企业提供最佳解决方案。

为了将 Pulsar 与 Flink 的功能进行整合，为用户提供更强大的开发能力，StreamNative 开发并开源了 Pulsar Flink Connector。经过多次的打磨，Pulsar Flink Connector 已合并进 Flink 代码仓库，并在 Flink 1.14.0 及其之后版本中发布！

Pulsar Flink Connector 基于 Apache Pulsar 和 Apache Flink 提供弹性数据处理，允许 Apache Flink 读写 Apache Pulsar 中的数据。使用 Pulsar Flink Connector，企业能够更专注于业务逻辑，无需关注存储问题。

引入依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-pulsar</artifactId>
    <version>1.15.0</version>
</dependency>

public class SourceTest {

    public static void main(String[] args) throws Exception {
        readPulsar();
    }

    /**
     * 从Pulsar中读取数据
     * @throws Exception
     */
    private static void readPulsar() throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 添加source源, 用于读取数据 pulsar
        String serviceUrl = "pulsar://192.168.23.111:6650,192.168.23.112:6650,192.168.23.113:6650";
        String adminUrl = "http://192.168.23.111:8080,192.168.23.112:8080,192.168.23.113:8080";
        String topic = "persistent://my-tenant/my-ns/my-partitioned-topic";

        PulsarSource<String> pulsarSource = PulsarSource.builder()
                .setServiceUrl(serviceUrl)
                .setAdminUrl(adminUrl)
                .setStartCursor(StartCursor.earliest())
                .setTopics(topic)
                .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
                .setSubscriptionName("my-subscription")
                .setSubscriptionType(SubscriptionType.Exclusive)
                .build();
        DataStreamSource<String> streamSource = env.fromSource(pulsarSource, WatermarkStrategy.noWatermarks(),"Pulsar Source");

        streamSource.print();

        env.execute();
    }
}

八、自定义 Source

大多数情况下，前面的数据源已经能够满足需要。但是凡事总有例外，如果遇到特殊情况，我们想要读取的数据源来自某个外部系统，而 flink 既没有预实现的方法、也没有提供连接器，又该怎么办呢？

那就只好自定义实现 SourceFunction 了。

Flink还提供了数据源接口，实现该接口就可以实现自定义数据源，不同的接口有不同的功能，分类如下：

• SourceFunction：非并行数据源（并行度只能=1）
• ParallelSourceFunction：并行数据源（并行度能够>=l)
• RichSourceFunction：多功能非并行数据源（并行度只能=1）
• RichParallelSourceFunction：多功能并行数据源（并行度能够>=1）

接下来我们创建一个自定义的数据源，实现 SourceFunction 接口。主要重写两个关键方法：run()和 cancel()。

• run()方法：使用运行时上下文对象（SourceContext）向下游发送数据；
• cancel()方法：通过标识位控制退出循环，来达到中断数据源的效果。

8.1 SourceFunction——单并行度Source

实现代码

public class ClickSource implements SourceFunction<Event> {

    //声明一个标志位
    private boolean running = true;

    @Override
    public void run(SourceContext<Event> sourceContext) throws Exception {
        //随机生成数据
        Random random = new Random();
        //定义字段选取的数据集
        String[] users = {"Mary", "Alice", "Bobo", "lucy"};
        String[] urls = {"./home", "./cart", "./prod", "./order"};

        //循环生成数据
        while (running){
            String user = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(urls.length)];
            long timestamp = System.currentTimeMillis();
            sourceContext.collect(new Event(user,url,timestamp));

            Thread.sleep(1000);
        }
    }

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

调用代码

public class SourceCustomTest {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> customDataStream = env.addSource(new ClickSource());

        parallelDataStream.print();

        env.execute();
    }
}

注意：SourceFunction 接口定义的数据源，并行度只能设置为 1，如果数据源设置为大于 1 的并行度，则会抛出异常。

8.2 ParallelSourceFunction——多并行度Source

实现代码

public class ParallelClickSource implements ParallelSourceFunction<Event> {

    //声明一个标志位
    private boolean running = true;

    @Override
    public void run(SourceContext<Event> sourceContext) throws Exception {
        //随机生成数据
        Random random = new Random();
        //定义字段选取的数据集
        String[] users = {"Mary", "Alice", "Bobo", "lucy"};
        String[] urls = {"./home", "./cart", "./prod", "./order"};

        //循环生成数据
        while (running){
            String user = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(urls.length)];
            long timestamp = System.currentTimeMillis();
            sourceContext.collect(new Event(user,url,timestamp));

            Thread.sleep(1000);
        }
    }

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

调用代码

public class SourceCustomTest {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> parallelDataStream = env.addSource(new ParallelClickSource()).setParallelism(2);

        parallelDataStream.print();

        env.execute();
    }
}

RichSourceFunction、RichParallelSourceFunction提供了外部连接的open和close方法以及运行时的getRuntimeContext等方法

8.3 RichSourceFunction——多功能非并行Source

实现代码

public class ClickRichSource extends RichSourceFunction<Event> {

    //声明一个标志位
    private boolean running = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        RuntimeContext runtimeContext = getRuntimeContext();
        String taskName = runtimeContext.getTaskName();
        int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
        System.out.println(taskName + "-" + indexOfThisSubtask);
    }

    @Override
    public void run(SourceContext<Event> sourceContext) throws Exception {
        //随机生成数据
        Random random = new Random();
        //定义字段选取的数据集
        String[] users = {"Mary", "Alice", "Bobo", "lucy"};
        String[] urls = {"./home", "./cart", "./prod", "./order"};

        //循环生成数据
        while (running){
            String user = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(urls.length)];
            long timestamp = System.currentTimeMillis();
            Event event = new Event(user, url, timestamp);
            System.out.println(event.toString());
            sourceContext.collect(event);

            Thread.sleep(1000);
        }
    }

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

    @Override
    public void close() throws Exception {
        System.out.println("关闭资源.....");
    }
}

调用代码

public class SourceCustomTest {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> RichDataStream = env.addSource(new ClickRichSource());

        parallelDataStream.print();

        env.execute();
    }
}

8.4 RichParallelSourceFunction——多功能并行Source

实现代码

public class RichParallelClickSource extends RichParallelSourceFunction<Event> {

    //声明一个标志位
    private boolean running = true;


    @Override
    public void open(Configuration parameters) throws Exception {
        RuntimeContext runtimeContext = getRuntimeContext();
        String taskName = runtimeContext.getTaskName();
        int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
        System.out.println(taskName + "-" + indexOfThisSubtask);
    }

    @Override
    public void run(SourceContext<Event> sourceContext) throws Exception {
        //随机生成数据
        Random random = new Random();
        //定义字段选取的数据集
        String[] users = {"Mary", "Alice", "Bobo", "lucy"};
        String[] urls = {"./home", "./cart", "./prod", "./order"};

        //循环生成数据
        while (running){
            String user = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(urls.length)];
            long timestamp = System.currentTimeMillis();
            sourceContext.collect(new Event(user,url,timestamp));

            Thread.sleep(1000);
        }
    }

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

    @Override
    public void close() throws Exception {
        System.out.println("关闭资源.....");
    }
}

调用代码

public class SourceCustomTest {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> RichDataStream = env.addSource(new RichParallelClickSource()).setParallelism(2);

        parallelDataStream.print();

        env.execute();
    }
}

8.5 从MySQL实时加载数据

实现代码

public class MySQLSource extends RichParallelSourceFunction<Student> {

    private boolean flag = true;
    private Connection conn;
    private PreparedStatement statement;
    private ResultSet resultSet;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata", "root", "123456");
        String sql = "select id, name, age from t_student";
        statement = conn.prepareStatement(sql);
        super.open(parameters);
    }

    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        while (flag) {
            resultSet = statement.executeQuery();
            while (resultSet.next()){
                String id = resultSet.getString("id");
                String name = resultSet.getString("name");
                Integer age = resultSet.getInt("age");
                ctx.collect(new Student(id, name, age));
            }
            Thread.sleep(3000);
        }
    }

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

    @Override
    public void close() throws Exception {
        if(conn != null) conn.close();
        if(statement != null) statement.close();
        if(resultSet != null) resultSet.close();
    }
}

调用代码

public class SourceCustomTest {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStreamSource<Student> parallelDataStream = env.addSource(new MySQLSource()).setParallelism(4);

        parallelDataStream.print();

        env.execute();
    }
}

参考：
https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/connectors/datastream/pulsar/

https://blog.csdn.net/weixin_47491957/article/details/124317150

https://blog.csdn.net/weixin_45417821/article/details/124143407

https://blog.csdn.net/weixin_45417821/article/details/124145083

https://blog.csdn.net/weixin_45417821/article/details/124146085

https://blog.csdn.net/weixin_45417821/article/details/124147285

https://segmentfault.com/a/1190000041048040

https://blog.csdn.net/qq_41924766/article/details/130681921