资讯专栏INFORMATION COLUMN

上云采购季！| 2核2G4M爆款云服务器低至59元/年，更有多台、长期优惠，快来选购！

立即前往
首页/文章专栏/《从0到1学习Flink》—— 如何自定义 Data Source ？

《从0到1学习Flink》—— 如何自定义 Data Source ？

songze / 1951人阅读

摘要：从上面自定义的可以看到我们继承的就是这个类，那么来了解一下一个抽象类，继承自。该类的子类有三个，两个是抽象类，在此基础上提供了更具体的实现，另一个是。

前言

在 《从0到1学习Flink》—— Data Source 介绍 文章中，我给大家介绍了 Flink Data Source 以及简短的介绍了一下自定义 Data Source，这篇文章更详细的介绍下，并写一个 demo 出来让大家理解。

Flink Kafka source 准备工作

我们先来看下 Flink 从 Kafka topic 中获取数据的 demo，首先你需要安装好了 FLink 和 Kafka 。

运行启动 Flink、Zookepeer、Kafka，

好了，都启动了！

maven 依赖 
</>复制代码 
  3.     org.apache.flink
  4.     flink-java
  5.     ${flink.version}
  6.     provided
  9.     org.apache.flink
  10.     flink-streaming-java_${scala.binary.version}
  11.     ${flink.version}
  12.     provided
  16.     org.slf4j
  17.     slf4j-log4j12
  18.     1.7.7
  19.     runtime
  22.     log4j
  23.     log4j
  24.     1.2.17
  25.     runtime
  29.     org.apache.flink
  30.     flink-connector-kafka-0.11_${scala.binary.version}
  31.     ${flink.version}
  35.     com.alibaba
  36.     fastjson
  37.     1.2.51
测试发送数据到 kafka topic

实体类，Metric.java

</>复制代码 
  1. package com.zhisheng.flink.model;
  2. import java.util.Map;
  3. /**
  4.  * Desc:
  5.  * weixi: zhisheng_tian
  6.  * blog: http://www.54tianzhisheng.cn/
  7.  */
  8. public class Metric {
  9.     public String name;
  10.     public long timestamp;
  11.     public Map fields;
  12.     public Map tags;
  13.     public Metric() {
  14.     }
  15.     public Metric(String name, long timestamp, Map fields, Map tags) {
  16.         this.name = name;
  17.         this.timestamp = timestamp;
  18.         this.fields = fields;
  19.         this.tags = tags;
  20.     }
  21.     @Override
  22.     public String toString() {
  23.         return "Metric{" +
  24.                 "name="" + name + """ +
  25.                 ", timestamp="" + timestamp + """ +
  26.                 ", fields=" + fields +
  27.                 ", tags=" + tags +
  28.                 "}";
  29.     }
  30.     public String getName() {
  31.         return name;
  32.     }
  33.     public void setName(String name) {
  34.         this.name = name;
  35.     }
  36.     public long getTimestamp() {
  37.         return timestamp;
  38.     }
  39.     public void setTimestamp(long timestamp) {
  40.         this.timestamp = timestamp;
  41.     }
  42.     public Map getFields() {
  43.         return fields;
  44.     }
  45.     public void setFields(Map fields) {
  46.         this.fields = fields;
  47.     }
  48.     public Map getTags() {
  49.         return tags;
  50.     }
  51.     public void setTags(Map tags) {
  52.         this.tags = tags;
  53.     }
  54. }

往 kafka 中写数据工具类：KafkaUtils.java

</>复制代码 
  1. import com.alibaba.fastjson.JSON;
  2. import com.zhisheng.flink.model.Metric;
  3. import org.apache.kafka.clients.producer.KafkaProducer;
  4. import org.apache.kafka.clients.producer.ProducerRecord;
  5. import java.util.HashMap;
  6. import java.util.Map;
  7. import java.util.Properties;
  8. /**
  9.  * 往kafka中写数据
  10.  * 可以使用这个main函数进行测试一下
  11.  * weixin: zhisheng_tian 
  12.  * blog: http://www.54tianzhisheng.cn/
  13.  */
  14. public class KafkaUtils {
  15.     public static final String broker_list = "localhost:9092";
  16.     public static final String topic = "metric";  // kafka topic，Flink 程序中需要和这个统一 
  17.     public static void writeToKafka() throws InterruptedException {
  18.         Properties props = new Properties();
  19.         props.put("bootstrap.servers", broker_list);
  20.         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //key 序列化
  21.         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //value 序列化
  22.         KafkaProducer producer = new KafkaProducer(props);
  23.         Metric metric = new Metric();
  24.         metric.setTimestamp(System.currentTimeMillis());
  25.         metric.setName("mem");
  26.         Map tags = new HashMap<>();
  27.         Map fields = new HashMap<>();
  28.         tags.put("cluster", "zhisheng");
  29.         tags.put("host_ip", "101.147.022.106");
  30.         fields.put("used_percent", 90d);
  31.         fields.put("max", 27244873d);
  32.         fields.put("used", 17244873d);
  33.         fields.put("init", 27244873d);
  34.         metric.setTags(tags);
  35.         metric.setFields(fields);
  36.         ProducerRecord record = new ProducerRecord(topic, null, null, JSON.toJSONString(metric));
  37.         producer.send(record);
  38.         System.out.println("发送数据: " + JSON.toJSONString(metric));
  39.         producer.flush();
  40.     }
  41.     public static void main(String[] args) throws InterruptedException {
  42.         while (true) {
  43.             Thread.sleep(300);
  44.             writeToKafka();
  45.         }
  46.     }
  47. }

运行：

如果出现如上图标记的，即代表能够不断的往 kafka 发送数据的。

Flink 程序

Main.java

</>复制代码 
  1. package com.zhisheng.flink;
  2. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  3. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
  6. import java.util.Properties;
  7. /**
  8.  * Desc:
  9.  * weixi: zhisheng_tian
  10.  * blog: http://www.54tianzhisheng.cn/
  11.  */
  12. public class Main {
  13.     public static void main(String[] args) throws Exception {
  14.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         Properties props = new Properties();
  16.         props.put("bootstrap.servers", "localhost:9092");
  17.         props.put("zookeeper.connect", "localhost:2181");
  18.         props.put("group.id", "metric-group");
  19.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //key 反序列化
  20.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  21.         props.put("auto.offset.reset", "latest"); //value 反序列化
  22.         DataStreamSource dataStreamSource = env.addSource(new FlinkKafkaConsumer011<>(
  23.                 "metric",  //kafka topic
  24.                 new SimpleStringSchema(),  // String 序列化
  25.                 props)).setParallelism(1);
  26.         dataStreamSource.print(); //把从 kafka 读取到的数据打印在控制台
  27.         env.execute("Flink add data source");
  28.     }
  29. }

运行起来：

看到没程序，Flink 程序控制台能够源源不断的打印数据呢。

自定义 Source

上面就是 Flink 自带的 Kafka source，那么接下来就模仿着写一个从 MySQL 中读取数据的 Source。

首先 pom.xml 中添加 MySQL 依赖：

</>复制代码 
  2.     mysql
  3.     mysql-connector-java
  4.     5.1.34

数据库建表如下：

</>复制代码 
  1. DROP TABLE IF EXISTS `student`;
  2. CREATE TABLE `student` (
  3.   `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  4.   `name` varchar(25) COLLATE utf8_bin DEFAULT NULL,
  5.   `password` varchar(25) COLLATE utf8_bin DEFAULT NULL,
  6.   `age` int(10) DEFAULT NULL,
  7.   PRIMARY KEY (`id`)
  8. ) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8 COLLATE=utf8_bin;

插入数据：

</>复制代码 
  1. INSERT INTO `student` VALUES ("1", "zhisheng01", "123456", "18"), ("2", "zhisheng02", "123", "17"), ("3", "zhisheng03", "1234", "18"), ("4", "zhisheng04", "12345", "16");
  2. COMMIT;

新建实体类：Student.java

</>复制代码 
  1. package com.zhisheng.flink.model;
  2. /**
  3.  * Desc:
  4.  * weixi: zhisheng_tian
  5.  * blog: http://www.54tianzhisheng.cn/
  6.  */
  7. public class Student {
  8.     public int id;
  9.     public String name;
  10.     public String password;
  11.     public int age;
  12.     public Student() {
  13.     }
  14.     public Student(int id, String name, String password, int age) {
  15.         this.id = id;
  16.         this.name = name;
  17.         this.password = password;
  18.         this.age = age;
  19.     }
  20.     @Override
  21.     public String toString() {
  22.         return "Student{" +
  23.                 "id=" + id +
  24.                 ", name="" + name + """ +
  25.                 ", password="" + password + """ +
  26.                 ", age=" + age +
  27.                 "}";
  28.     }
  29.     public int getId() {
  30.         return id;
  31.     }
  32.     public void setId(int id) {
  33.         this.id = id;
  34.     }
  35.     public String getName() {
  36.         return name;
  37.     }
  38.     public void setName(String name) {
  39.         this.name = name;
  40.     }
  41.     public String getPassword() {
  42.         return password;
  43.     }
  44.     public void setPassword(String password) {
  45.         this.password = password;
  46.     }
  47.     public int getAge() {
  48.         return age;
  49.     }
  50.     public void setAge(int age) {
  51.         this.age = age;
  52.     }
  53. }

新建 Source 类 SourceFromMySQL.java，该类继承 RichSourceFunction ，实现里面的 open、close、run、cancel 方法：

</>复制代码 
  1. package com.zhisheng.flink.source;
  2. import com.zhisheng.flink.model.Student;
  3. import org.apache.flink.configuration.Configuration;
  4. import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
  5. import java.sql.Connection;
  6. import java.sql.DriverManager;
  7. import java.sql.PreparedStatement;
  8. import java.sql.ResultSet;
  9. /**
  10. * Desc:
  11. * weixi: zhisheng_tian
  12. * blog: http://www.54tianzhisheng.cn/
  13. */
  14. public class SourceFromMySQL extends RichSourceFunction {
  15.    PreparedStatement ps;
  16.    private Connection connection;
  17.    /**
  18.     * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接。
  19.     *
  20.     * @param parameters
  21.     * @throws Exception
  22.     */
  23.    @Override
  24.    public void open(Configuration parameters) throws Exception {
  25.        super.open(parameters);
  26.        connection = getConnection();
  27.        String sql = "select * from Student;";
  28.        ps = this.connection.prepareStatement(sql);
  29.    }
  30.    /**
  31.     * 程序执行完毕就可以进行，关闭连接和释放资源的动作了
  32.     *
  33.     * @throws Exception
  34.     */
  35.    @Override
  36.    public void close() throws Exception {
  37.        super.close();
  38.        if (connection != null) { //关闭连接和释放资源
  39.            connection.close();
  40.        }
  41.        if (ps != null) {
  42.            ps.close();
  43.        }
  44.    }
  45.    /**
  46.     * DataStream 调用一次 run() 方法用来获取数据
  47.     *
  48.     * @param ctx
  49.     * @throws Exception
  50.     */
  51.    @Override
  52.    public void run(SourceContext ctx) throws Exception {
  53.        ResultSet resultSet = ps.executeQuery();
  54.        while (resultSet.next()) {
  55.            Student student = new Student(
  56.                    resultSet.getInt("id"),
  57.                    resultSet.getString("name").trim(),
  58.                    resultSet.getString("password").trim(),
  59.                    resultSet.getInt("age"));
  60.            ctx.collect(student);
  61.        }
  62.    }
  63.    @Override
  64.    public void cancel() {
  65.    }
  66.    private static Connection getConnection() {
  67.        Connection con = null;
  68.            try {
  69.                Class.forName("com.mysql.jdbc.Driver");
  70.                con = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "root123456");
  71.            } catch (Exception e) {
  72.                System.out.println("-----------mysql get connection has exception , msg = "+ e.getMessage());
  73.            }
  74.        return con;
  75.    }
  76. }

Flink 程序：

</>复制代码 
  1. package com.zhisheng.flink;
  2. import com.zhisheng.flink.source.SourceFromMySQL;
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4. /**
  5.  * Desc:
  6.  * weixi: zhisheng_tian
  7.  * blog: http://www.54tianzhisheng.cn/
  8.  */
  9. public class Main2 {
  10.     public static void main(String[] args) throws Exception {
  11.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  12.         env.addSource(new SourceFromMySQL()).print();
  13.         env.execute("Flink add data sourc");
  14.     }
  15. }

运行 Flink 程序，控制台日志中可以看见打印的 student 信息。

RichSourceFunction

从上面自定义的 Source 可以看到我们继承的就是这个 RichSourceFunction 类，那么来了解一下：

一个抽象类，继承自 AbstractRichFunction。为实现一个 Rich SourceFunction 提供基础能力。该类的子类有三个，两个是抽象类，在此基础上提供了更具体的实现，另一个是 ContinuousFileMonitoringFunction。

MessageAcknowledgingSourceBase ：它针对的是数据源是消息队列的场景并且提供了基于 ID 的应答机制。

MultipleIdsMessageAcknowledgingSourceBase ： 在 MessageAcknowledgingSourceBase 的基础上针对 ID 应答机制进行了更为细分的处理，支持两种 ID 应答模型：session id 和 unique message id。

ContinuousFileMonitoringFunction：这是单个（非并行）监视任务，它接受 FileInputFormat，并且根据 FileProcessingMode 和 FilePathFilter，它负责监视用户提供的路径；决定应该进一步读取和处理哪些文件；创建与这些文件对应的 FileInputSplit 拆分，将它们分配给下游任务以进行进一步处理。

最后

本文主要讲了下 Flink 使用 Kafka Source 的使用，并提供了一个 demo 教大家如何自定义 Source，从 MySQL 中读取数据，当然你也可以从其他地方读取，实现自己的数据源 source。可能平时工作会比这个更复杂，需要大家灵活应对！

关注我

转载请务必注明原创地址为：http://www.54tianzhisheng.cn/2018/10/30/flink-create-source/

另外我自己整理了些 Flink 的学习资料，目前已经全部放到微信公众号了。你可以加我的微信：zhisheng_tian，然后回复关键字：Flink 即可无条件获取到。

相关文章

1、《从0到1学习Flink》—— Apache Flink 介绍

2、《从0到1学习Flink》—— Mac 上搭建 Flink 1.6.0 环境并构建运行简单程序入门

3、《从0到1学习Flink》—— Flink 配置文件详解

4、《从0到1学习Flink》—— Data Source 介绍

5、《从0到1学习Flink》—— 如何自定义 Data Source ？

6、《从0到1学习Flink》—— Data Sink 介绍

7、《从0到1学习Flink》—— 如何自定义 Data Sink ？

GPU云服务器 云服务器 从0到1搭建视频通话系统 从0开始到1千万用户 从0开始学习php 自定义如何实现

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/72116.html

相关文章

  • 《从0到1学习Flink》—— Flink Data transformation(转换)

    摘要：这些切片称为窗口。函数允许对常规数据流进行分组。通常，这是非并行数据转换，因为它在非分区数据流上运行。 showImg(https://segmentfault.com/img/remote/1460000017874226?w=1920&h=1271); 前言 在第一篇介绍 Flink 的文章 《《从0到1学习Flink》—— Apache Flink 介绍》 中就说过 Flink ...

    oujie 评论0 收藏0

  • 《从0到1学习Flink》—— Data Source 介绍

    摘要：指定了该迭代器返回元素的类型。这可能导致节点故障后的恢复速度较慢，因为该作业将从最后一个检查点恢复读取。监听的端口过来的数据这个在从到学习上搭建环境并构建运行简单程序入门文章里用的就是基于的程序。取消一个，也即将中的循环元素的行为终止。 showImg(https://segmentfault.com/img/remote/1460000016944116); 前言 Data Sou...

    XFLY 评论0 收藏0

发表评论

登陆后可评论

0条评论

songze

男|高级讲师
我要私信

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<