示例3：将DMS Kafka作为源表，Print作为结果表（适用于Kafka集群已开启SASL_SSL场景） 创建DMS的kafka集群，开启SASL_SSL，并下载SSL证书，将下载的证书client.jks上传到OBS桶中。 其中，properties.sasl.jaas.config字段包含账号密码，使用DEW进行加密。 CREATE TABLE ordersSource ( order_id string, order_channel string, order_time timestamp(3), pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = 'KafkaTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:9093,KafkaAddress2:9093', 'properties.group.id' = 'GroupId', 'scan.startup.mode' = 'latest-offset', 'properties.connector.auth.open' = 'true', 'properties.ssl.truststore.location' = 'obs://xx/client.jks', -- 用户上传证书的位置 'properties.sasl.mechanism' = 'PLAIN', 'properties.security.protocol' = 'SASL_SSL', 'properties.sasl.jaas.config' = 'xx', -- dew凭据管理中的key，其值如：org.apache.kafka.common.security.plain.PlainLoginModule required username=xx password=xx; 'format' = 'json', 'dew.endpoint' = 'kms.xx.com', --使用的DEW服务所在的endpoint信息 'dew.csms.secretName' = 'xx', --DEW服务通用凭据的凭据名称 'dew.csms.decrypt.fields' = 'properties.sasl.jaas.config', --其中properties.sasl.jaas.config字段值，需要利用DEW凭证管理,进行解密替换 'dew.csms.version' = 'v1' ); CREATE TABLE ordersSink ( order_id string, order_channel string, order_time timestamp(3), pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'print' ); insert into ordersSink select * from ordersSource;

常见问题 Flink作业运行失败，作业运行日志中如下报错信息，应该怎么解决？ org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata 跨源未绑定或未绑定成功，或是Kafka集群安全组未配置放通 DLI 队列的网段地址。重新配置跨源，或者Kafka集群安全组放通DLI队列的网段地址。 具体操作请参考增强型跨源连接。 Flink作业运行失败，作业运行日志中如下报错信息，应该怎么解决？ Caused by: java.lang.RuntimeException: RealLine:45;Table 'default_catalog.default_database.printSink' declares persistable metadata columns, but the underlying DynamicTableSink doesn't implement the SupportsWritingMetadata interface. If the column should not be persisted, it can be declared with the VIRTUAL keyword. sink表中定义了metadata类型，但是Print connector并不支持把sink表中的matadata去掉即可。

注意事项 更多具体使用可参考开源社区文档：Apache Kafka SQL 连接器。 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 with参数中字段只能使用单引号，不能使用双引号。 建表时数据类型的使用请参考Format章节。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据

语法格式 1 2 3 4 5 6 7 8 9 10 11 12 13 14 create table kafkaSource( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) (',' WATERMARK FOR rowtime_column_name AS watermark-strategy_expression) ) with ( 'connector' = 'kafka', 'topic' = '', 'properties.bootstrap.servers' = '', 'properties.group.id' = '', 'scan.startup.mode' = '', 'format' = '' );

Topic和Partition的探测 topic 和 topic-pattern 配置项决定了 source 消费的 topic 或 topic 的匹配规则。topic 配置项可接受使用分号间隔的 topic 列表，例如 topic-1;topic-2。 topic-pattern 配置项使用正则表达式来探测匹配的 topic。例如 topic-pattern 设置为 test-topic-[0-9]，则在作业启动时，所有匹配该正则表达式的 topic（以 test-topic- 开头，以一位数字结尾）都将被 consumer 订阅。 为允许 consumer 在作业启动之后探测到动态创建的 topic，请将 scan.topic-partition-discovery.interval 配置为一个非负值。这将使 consumer 能够探测匹配名称规则的 topic 中新的 partition。 topic列表和topic匹配规则只适用于 source。对于sink端，Flink目前只支持单一topic。

示例2：将json格式DMS Kafka作为源表，输出到Kafka sink中（适用于Kafka集群未开启SASL_SSL场景） 将Kafka作为源表，Kafka作为结果表，从Kafka中读取编码格式为json数据类型的数据，输出到日志文件中。 参考增强型跨源连接，根据Kafka所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根据Kafka的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，并提交运行。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE kafkaSource( order_id string, order_channel string, order_time timestamp(3), pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = 'KafkaSourceTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort', 'properties.group.id' = 'GroupId', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ); CREATE TABLE kafkaSink ( order_id string, order_channel string, order_time timestamp(3), pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = 'KafkaSinkTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort', 'format' = 'json' ); insert into kafkaSink select * from kafkaSource; 向Kafka的源表的topic中发送如下数据： {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"} 读取Kafka的结果表的topic，其数据结果参考如下： {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"}

功能描述 Kafka 连接器提供从 Kafka topic 中消费和写入数据的能力。 Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。 表1 支持类别 类别 详情 支持表类型 源表、结果表 支持数据格式 CS V JSON Apache Avro Confluent Avro Debezium CDC Canal CDC Maxwell CDC OGG CDC Raw

表类型 源表：源表是Flink作业的数据输入表，例如Kafka等实时流数据输入。 维表：数据源表的辅助表，用于丰富和扩展源表的数据。在Flink作业中，因为数据采集端采集到的数据往往比较有限，在做数据分析之前，就要先将所需的维度信息补全，而维表就是代表存储数据维度信息的数据源。常见的用户维表有 MySQL，Redis等。 结果表：Flink作业输出的结果数据表，将每条实时处理完的数据写入的目标存储，如 MySQL，HBase 等数据库。

Connector支持列表 表1 Connector支持列表 Connector 源表 维表 结果表 BlackHole 不支持 不支持 支持 ClickHouse 不支持 不支持 支持 DataGen 支持 不支持 不支持 Doris 支持 支持 支持 DWS 支持 支持 支持 Elasticsearch 不支持 不支持 支持 FileSystem 支持 不支持 支持 Hbase 支持 支持 支持 Hive 支持 支持 支持 JDBC 支持 支持 支持 Kafka 支持 不支持 支持 Print 不支持 不支持 支持 Redis 支持 支持 支持 Upsert Kafka 支持 不支持 支持

示例 示例1：使用JDBC作为数据源，Print作为结果表，从RDS MySQL数据库中读取数据，并写入到Print结果表中。 参考增强型跨源连接，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考创建RDS数据库。 在flink数据库库下创建orders表： CREATE TABLE `flink`.`orders` ( `order_id` VARCHAR(32) NOT NULL, `order_channel` VARCHAR(32) NULL, PRIMARY KEY (`order_id`) ) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci; 插入表数据： insert into orders( order_id, order_channel ) values ('1', 'webShop'), ('2', 'miniAppShop'); 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 认证用的username和password硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 CREATE TABLE jdbcSource ( order_id string, order_channel string ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://MySQLAddress:MySQLPort/flink',--flink为RDS MySQL创建的数据库名 'table-name' = 'orders', 'username' = 'MySQLUsername', 'password' = 'MySQLPassword', 'scan.fetch-size' = '10', 'scan.auto-commit' = 'true' ); CREATE TABLE printSink ( order_id string, order_channel string ) WITH ( 'connector' = 'print' ); insert into printSink select * from jdbcSource; 查看taskmanager.out文件中的数据结果，数据结果参考如下： +I(1,webShop) +I(2,miniAppShop) 示例2：使用DataGen源表发送数据，通过JDBC结果表将数据输出到MySQL数据库中。 参考增强型跨源连接，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考创建RDS数据库。 在flink数据库库下创建orders表： CREATE TABLE `flink`.`orders` ( `order_id` VARCHAR(32) NOT NULL, `order_channel` VARCHAR(32) NULL, PRIMARY KEY (`order_id`) ) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci; 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE dataGenSource ( order_id string, order_channel string ) WITH ( 'connector' = 'datagen', 'fields.order_id.kind' = 'sequence', 'fields.order_id.start' = '1', 'fields.order_id.end' = '1000', 'fields.order_channel.kind' = 'random', 'fields.order_channel.length' = '5' ); CREATE TABLE jdbcSink ( order_id string, order_channel string, PRIMARY KEY(order_id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://MySQLAddress:MySQLPort/flink',--其中url中的flink表示MySQL中orders表所在的数据库名 'table-name' = 'orders', 'username' = 'MySQLUsername', 'password' = 'MySQLPassword', 'sink.buffer-flush.max-rows' = '1' ); insert into jdbcSink select * from dataGenSource; 查看表中数据，在MySQL中执行sql查询语句 select * from orders； 示例3：从DataGen源表中读取数据，将JDBC表作为维表，并将二者生成的表信息写入Print结果表中。 参考增强型跨源连接，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考创建RDS数据库。 在flink数据库库下创建orders表： CREATE TABLE `flink`.`orders` ( `order_id` VARCHAR(32) NOT NULL, `order_channel` VARCHAR(32) NULL, PRIMARY KEY (`order_id`) ) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci; 插入表数据： insert into orders( order_id, order_channel ) values ('1', 'webShop'), ('2', 'miniAppShop'); 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。该作业脚本将DataGen为数据源，JDBC作为维表，数据写入到Print结果表。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE dataGenSource ( order_id string, order_time timestamp, proctime as Proctime() ) WITH ( 'connector' = 'datagen', 'fields.order_id.kind' = 'sequence', 'fields.order_id.start' = '1', 'fields.order_id.end' = '2' ); --创建维表 CREATE TABLE jdbcTable ( order_id string, order_channel string ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://JDBCAddress:JDBCPort/flink',--flink为RDS MySQL中orders表所在的数据库名 'table-name' = 'orders', 'username' = 'JDBCUserName', 'password' = 'JDBCPassWord', 'lookup.cache.max-rows' = '100', 'lookup.cache.ttl' = '1000', 'lookup.cache.caching-missing-key' = 'false', 'lookup.max-retries' = '5' ); CREATE TABLE printSink ( order_id string, order_time timestamp, order_channel string ) WITH ( 'connector' = 'print' ); insert into printSink SELECT dataGenSource.order_id, dataGenSource.order_time, jdbcTable.order_channel from dataGenSource left join jdbcTable for system_time as of dataGenSource.proctime on dataGenSource.order_id = jdbcTable.order_id; 查看taskmanager.out文件中的数据结果，数据结果参考如下： +I(1, xxx, webShop) +I(2, xxx, miniAppShop)

数据类型映射 表3 数据类型映射 MySQL类型 PostgreSQL类型 Flink SQL类型 TINYINT - TINYINT SMALLINT TINYINT UNSIGNED SMALLINT INT2 SMALLSERIAL SERIAL2 SMALLINT INT MEDIUMINT SMALLINT UNSIGNED INTEGER SERIAL INT BIGINT INT UNSIGNED BIGINT BIGSERIAL BIGINT BIGINT UNSIGNED - DECIMAL(20, 0) BIGINT BIGINT BIGINT FLOAT REAL FLOAT4 FLOAT DOUBLE DOUBLE PRECISION FLOAT8 DOUBLE PRECISION DOUBLE NUMERIC(p, s) DECIMAL(p, s) NUMERIC(p, s) DECIMAL(p, s) DECIMAL(p, s) BOOLEAN TINYINT(1) BOOLEAN BOOLEAN DATE DATE DATE TIME [(p)] TIME [(p)] [WITHOUT TIMEZONE] TIME [(p)] [WITHOUT TIMEZONE] DATETIME [(p)] TIMESTAMP [(p)] [WITHOUT TIMEZONE] TIMESTAMP [(p)] [WITHOUT TIMEZONE] CHAR(n) VARCHAR(n) TEXT CHAR(n) CHARACTER(n) VARCHAR(n) CHARACTER VARYING(n) TEXT STRING BINARY VARBINARY BLOB BYTEA BYTES - ARRAY ARRAY

分区扫描功能介绍 为了加速Source任务实例中的数据读取，Flink为JDBC表提供了分区扫描功能。以下参数定义了从多个任务并行读取时如何对表进行分区。 scan.partition.column：用于对输入进行分区的列名，该列的数据类型必须是数字，日期或时间戳。 scan.partition.num: 分区数。 scan.partition.lower-bound：第一个分区的最小值。 scan.partition.upper-bound：最后一个分区的最大值。 建表时以上扫描分区参数必须同时存在或者同时不存在。 scan.partition.lower-bound和scan.partition.upper-bound参数仅用于决定分区步长，而不是用于过滤表中的行，表中的所有行都会被分区并返回。

Lookup Cache功能介绍 JDBC连接器可以用在时态表关联中作为一个可lookup的维表，当前只支持同步的查找模式。 默认情况下，Lookup cache是未启用的，所有请求都会发送到外部数据库。你可以设置Lookup.cache.max-rows和Lookup.cache.ttl参数来启用。Lookup cache的主要目的是用于提高时态表关联JDBC连接器的性能。 当Lookup cache被启用时，每个进程（即TaskManager）将维护一个缓存。Flink将优先查找缓存，只有当缓存未查找到时才向外部数据库发送请求，并使用返回的数据更新缓存。当缓存命中最大缓存行Lookup.cache.max-rows或当行超过最大存活时间Lookup.cache.ttl时，缓存中最先添加的条目将被标记为过期。缓存中的记录可能不是最新的，用户可以将Lookup.cache.ttl设置为一个更小的值以获得更好的刷新数据，但这可能会增加发送到数据库的请求数。所以要做好吞吐量和正确性之间的平衡。 默认情况下，Flink会缓存主键的空查询结果，你可以通过将Lookup.cache.caching-missing-key设置为false来切换行为。

参数说明 表2 参数说明 参数 是否必选 默认值 类型 说明 connector 是 无 String 指定要使用的连接器，当前固定为'jdbc'。 url 是 无 String 数据库的URL。 连接MySQL数据库时，格式为：jdbc:mysql://MySQLAddress:MySQLPort/dbName 。 连接PostgreSQL数据库时，格式为：jdbc:postgresql://PostgreSQLAddress:PostgreSQLPort/dbName。 table-name 是 无 String 读取数据库中的数据所在的表名。 driver 否 无 String 连接数据库所需要的驱动。如果未配置，则会自动通过URL提取。 MySQL数据库默认驱动为com.mysql.jdbc.Driver。 PostgreSQL数据库默认驱动为org.postgresql.Driver。 username 否 无 String 数据库认证用户名，需要和'password'一起配置。 password 否 无 String 数据库认证密码，需要和'username'一起配置。 connection.max-retry-timeout 否 60s Duration 尝试连接数据库服务器最大重试超时时间，不应小于1s。 scan.partition.column 否 无 String 用于对输入进行分区的列名。分区扫描参数，具体请参考分区扫描功能介绍。 scan.partition.num 否 无 Integer 分区的个数。分区扫描参数，具体请参考分区扫描功能介绍。 scan.partition.lower-bound 否 无 Integer 第一个分区的最小值。分区扫描参数，具体请参考分区扫描功能介绍。 scan.partition.upper-bound 否 无 Integer 最后一个分区的最大值。分区扫描参数，具体请参考分区扫描功能介绍。 scan.fetch-size 否 0 Integer 每次从数据库拉取数据的行数。如果指定为0，则会忽略sql hint。 scan.auto-commit 否 true Boolean 是否设置自动提交，以确定事务中的每个statement是否自动提交 lookup.cache.max-rows 否 无 Integer lookup cache的最大行数，如果超过该值，缓存中最先添加的条目将被标记为过期。 默认情况下，lookup cache是未开启的。具体请参考Lookup Cache功能介绍。 lookup.cache.ttl 否 无 Duration lookup cache中每一行记录的最大存活时间，如果超过该时间，缓存中最先添加的条目将被标记为过期。 默认情况下，lookup cache是未开启的。具体请参考Lookup Cache功能介绍。 lookup.cache.caching-missing-key 否 true Boolean 是否缓存空查询结果，默认为true。具体请参考Lookup Cache功能介绍。 lookup.max-retries 否 3 Integer 查询数据库失败的最大重试次数。 sink.buffer-flush.max-rows 否 100 Integer flush前缓存记录的最大值，可以设置为 '0' 来禁用它。 sink.buffer-flush.interval 否 1s Duration flush间隔时间，超过该时间后异步线程将flush数据。可以设置为 '0' 来禁用它。如果想完全异步地处理缓存的flush事件，可以将 'sink.buffer-flush.max-rows' 设置为 '0' ，并配置适当的flush时间间隔。 sink.max-retries 否 3 Integer 写入到数据库失败后的最大重试次数。 sink.parallelism 否 无 Integer 用于定义JDBC sink算子的并行度。默认情况下，并行度是由框架决定，即与上游并行度一致。

注意事项 JDBC结果表如果定义了主键，将以upsert模式与外部系统交换UPDATE/DELETE消息；否则，它将以append模式与外部系统交换消息，不支持消费UPDATE/DELETE消息。 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据