欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

如何自定义 Flink Connectors(Source 和 Sink)?

程序员文章站 2022-07-14 14:01:34
...

在前面文章 3.6 节中讲解了 Flink 中的 Data Source 和 Data Sink,然后介绍了 Flink 中自带的一些 Source 和 Sink 的 Connector,接着我们还有几篇实战会讲解了如何从 Kafka 处理数据写入到 Kafka、ElasticSearch 等,当然 Flink 还有一些其他的 Connector,我们这里就不一一介绍了,大家如果感兴趣的话可以去官网查看一下,如果对其代码实现比较感兴趣的话,也可以去看看其源码的实现。我们这篇文章来讲解一下如何自定义 Source 和 Sink Connector?这样我们后面再遇到什么样的需求都难不倒我们了。

如何自定义 Source Connector?

这里就演示一下如何自定义 Source 从 MySQL 中读取数据。

添加依赖

在 pom.xml 中添加 MySQL 依赖:

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

数据库建表

数据库建表如下:

DROP TABLE IF EXISTS `student`;
CREATE TABLE `student` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(25) COLLATE utf8_bin DEFAULT NULL,
  `password` varchar(25) COLLATE utf8_b