华为云用户手册

语法格式 create table hbaseSource ( attr_name attr_type (',' attr_name attr_type)* (',' watermark for rowtime_column_name as watermark-strategy_expression) ','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'hbase-2.2', 'table-name' = '', 'zookeeper.quorum' = '' );

数据类型映射 HBase以字节数组存储所有数据，在读和写过程中要序列化和反序列化数据。 Flink的HBase连接器利用HBase（Hadoop) 的工具类org.apache.hadoop.hbase.util.Bytes进行字节数组和Flink数据类型转换。 Flink的HBase连接器将所有数据类型（除字符串外）null值编码成空字节。对于字符串类型，null值的字面值由null-string-literal选项值决定。 表2 数据类型映射表 Flink数据类型 HBase转换 CHAR/VARCHAR/STRING byte[] toBytes(String s) String toString(byte[] b) BOOLEAN byte[] toBytes(boolean b) boolean toBoolean(byte[] b) BINARY/VARBINARY 返回 byte[]。 DECIMAL byte[] toBytes(BigDecimal v) BigDecimal toBigDecimal(byte[] b) TINYINT new byte[] { val } bytes[0] // returns first and only byte from bytes SMALLINT byte[] toBytes(short val) short toShort(byte[] bytes) INT byte[] toBytes(int val) int toInt(byte[] bytes) BIGINT byte[] toBytes(long val) long toLong(byte[] bytes) FLOAT byte[] toBytes(float val) float toFloat(byte[] bytes) DOUBLE byte[] toBytes(double val) double toDouble(byte[] bytes) DATE 从 1970-01-01 00:00:00 UTC 开始的天数，int 值。 TIME 从 1970-01-01 00:00:00 UTC 开始天的毫秒数，int 值。 TIMESTAMP 从 1970-01-01 00:00:00 UTC 开始的毫秒数，long 值。 ARRAY 不支持 MAP/MULTISET 不支持 ROW 不支持

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 创建HBase源表的列簇必须定义为ROW类型，字段名对应列簇名（column family），嵌套的字段名对应列限定符名（column qualifier）。 用户只需在表结构中声明查询中使用的的列簇和列限定符。除了ROW类型的列，剩下的原子数据类型字段（比如，STRING, BIGINT）将被识别为HBase的rowkey，一张表中只能声明一个rowkey。rowkey字段的名字可以是任意的，如果是保留关键字，需要用反引号进行转义。

语法格式 create table dorisSource ( attr_name attr_type (',' attr_name attr_type)* ) with ( 'connector' = 'doris', 'fenodes' = 'FE_IP:PORT,FE_IP:PORT,FE_IP:PORT', 'table.identifier' = 'database.table', 'username' = 'dorisUsername', 'password' = 'dorisPassword' );

前提条件 该场景作业需要运行在 DLI 的独享队列上，因此要与HBase建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《 数据湖探索 用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。 如果使用 MRS Doris，请在增强型跨源的主机信息中添加MRS集群所有节点的主机ip信息。 详细操作请参考《 数据湖 探索用户指南》中的“修改主机信息”章节描述。 集群未启用Kerberos认证（普通模式） 使用admin用户连接Doris后，创建具有管理员权限的角色并绑定给用户。

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 集群未启用Kerberos认证（普通模式） Doris的表名是区分大小写。 使用cloudTable的doris时，'fenodes'字段值的端口请用8030，如'xx:8030'。同时安全组请放开端口8030，8040，9030。 开启HTTPS后，需要在创建表的with子句中添加如下配置参数： 'doris.enable.https' = 'true' 'doris.ignore.https.ca' = 'true' 请在Flink“作业编辑”页面选择“运行参数配置”，选择“开启Checkpoint”，否则会导致Doris结果表无法写入数据。

示例 该示例是从Datagen数据源中生成数据，并将结果写入到Doris结果表中。 参考增强型跨源连接，在DLI上根据Doris所在的虚拟私有云和子网分别创建相应的增强型跨源连接，并绑定所要使用的Flink弹性资源池。参考“修改主机信息”章节描述，在增强型跨源中增加MRS的主机信息。 设置Doris的安全组，添加入向规则使其对Flink的队列网段放通。分别根据Doris的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 参考测试地址连通性。 参考MRS Doris使用指南，创建doris表，创建语句如下： CREATE TABLE IF NOT EXISTS dorisdemo ( `user_id` varchar(10) NOT NULL, `city` varchar(10), `age` int, `gender` int ) DISTRIBUTED BY HASH(`user_id`) BUCKETS 10 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业脚本，并提交运行。该作业脚本将Datagen作为数据源，将数据写入到Doris作为结果表中。 create table student_datagen_source( `user_id` String NOT NULL, `city` String, `age` int, `gender` int ) with ( 'connector' = 'datagen', 'rows-per-second' = '1', 'fields.user_id.kind' = 'random', 'fields.user_id.length' = '7', 'fields.city.kind' = 'random', 'fields.city.length' = '7' ); CREATE TABLE dorisDemo ( `user_id` String NOT NULL, `city` String, `age` int, `gender` int ) with ( 'connector' = 'doris', 'fenodes' = 'FE_IP:PORT', 'table.identifier' = 'demo.dorisdemo', 'username' = 'dorisUser', 'password' = 'dorisPassword', 'sink.label-prefix' = 'demo', 'sink.enable-2pc' = 'true', 'sink.buffer-count' = '10' ); insert into dorisDemo select * from student_datagen_source 查看doris结果表是否已成功写入数据。 user_id city age gender 50aff04 93406c5 12 1 681a230 1f27d06 16 1 006eff4 3521ded 18 0

数据类型映射 HBase以字节数组存储所有数据。在读和写过程中要序列化和反序列化数据。 Flink的HBase连接器利用HBase（Hadoop) 的工具类 org.apache.hadoop.hbase.util.Bytes 进行字节数组和 Flink 数据类型转换。 Flink的HBase连接器将所有数据类型（除字符串外）null 值编码成空字节。对于字符串类型，null 值的字面值由null-string-literal选项值决定。 表2 数据类型映射表 Flink 数据类型 HBase 转换 CHAR / VARCHAR / STRING byte[] toBytes(String s) String toString(byte[] b) BOOLEAN byte[] toBytes(boolean b) boolean toBoolean(byte[] b) BINARY / VARBINARY 返回 byte[]。 DECIMAL byte[] toBytes(BigDecimal v) BigDecimal toBigDecimal(byte[] b) TINYINT new byte[] { val } bytes[0] // returns first and only byte from bytes SMALLINT byte[] toBytes(short val) short toShort(byte[] bytes) INT byte[] toBytes(int val) int toInt(byte[] bytes) BIGINT byte[] toBytes(long val) long toLong(byte[] bytes) FLOAT byte[] toBytes(float val) float toFloat(byte[] bytes) DOUBLE byte[] toBytes(double val) double toDouble(byte[] bytes) DATE 从 1970-01-01 00:00:00 UTC 开始的天数，int 值。 TIME 从 1970-01-01 00:00:00 UTC 开始天的毫秒数，int 值。 TIMESTAMP 从 1970-01-01 00:00:00 UTC 开始的毫秒数，long 值。 ARRAY 不支持 MAP / MULTISET 不支持 ROW 不支持

常见问题 Q：Flink作业日志中有如下报错信息应该怎么解决？ org.apache.zookeeper.ClientCnxn$SessionTimeoutException: Client session timed out, have not heard from server in 90069ms for connection id 0x0 A：可能是跨源连接未绑定或跨源绑定失败。参考增强型跨源连接重新配置跨源，Kafka集群安全组放通DLI队列的网段地址。

前提条件 该场景作业需要运行在DLI的独享队列上，因此要与HBase建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。 若使用MRS HBase，请在增强型跨源的主机信息中添加MRS集群所有节点的主机IP信息。 详细操作请参考《数据湖探索用户指南》中的“修改主机信息”章节描述。 Flink跨源开发场景中直接配置跨源认证信息存在密码泄露的风险，优先推荐您使用DLI提供的跨源认证。 跨源认证简介及操作方法请参考跨源认证简介。

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 所有 HBase 表的列簇必须定义为ROW类型，字段名对应列簇名（column family），嵌套的字段名对应列限定符名（column qualifier）。用户只需在表结构中声明查询中使用的的列簇和列限定符。除了 ROW 类型的列，剩下的原子数据类型字段（比如，STRING, BIGINT）将被识别为 HBase 的 rowkey，一张表中只能声明一个 rowkey。rowkey 字段的名字可以是任意的，如果是保留关键字，需要用反引号。

参数说明 表1 参数说明 参数 是否必选 说明 connector.type 是 数据源类型，‘jdbc’表示使用JDBC connector，必须为jdbc connector.url 是 数据库的URL connector.table 是 读取数据库中的数据所在的表名 connector.driver 否 连接数据库所需要的驱动。若未配置，则会自动通过URL提取 connector.username 否 数据库认证用户名，需要和'connector.password'一起配置 connector.password 否 数据库认证密码，需要和'connector.username'一起配置 connector.read.partition.column 否 用于对输入进行分区的列名 与connector.read.partition.lower-bound、connector.read.partition.upper-bound、 connector.read.partition.num必须同时存在或者同时不存在 connector.read.partition.lower-bound 否 第一个分区的最小值 与connector.read.partition.column、connector.read.partition.upper-bound、 connector.read.partition.num必须同时存在或者同时不存在 connector.read.partition.upper-bound 否 最后一个分区的最大值 与connector.read.partition.column、connector.read.partition.lower-bound、 connector.read.partition.num必须同时存在或者同时不存在 connector.read.partition.num 否 分区的个数 与connector.read.partition.column、connector.read.partition.upper-bound、 connector.read.partition.upper-bound必须同时存在或者同时不存在 connector.read.fetch-size 否 每次从数据库拉取数据的行数。默认值为0，表示忽略该提示。 connector.lookup.cache.max-rows 否 维表配置，缓存的最大行数，超过该值时，最先添加的数据将被标记为过期。-1表示不使用缓存。 connector.lookup.cache.ttl 否 维表配置，缓存超时时间，超过该时间的数据会被剔除。格式为：{length value}{time unit label}，如123ms, 321s，支持的时间单位包括: d,h,min,s,ms等，默认为ms。 connector.lookup.max-retries 否 维表配置，数据拉取最大重试次数，默认为3。

示例 RDS表用于与输入流连接。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 CREATE TABLE car_infos ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT, proctime as PROCTIME() ) WITH ( 'connector.type' = 'dis', 'connector.region' = 'cn-north-1', 'connector.channel' = 'disInput', 'format.type' = 'csv' ); CREATE TABLE db_info ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( 'connector.type' = 'jdbc', 'connector.url' = 'jdbc:mysql://xx.xx.xx.xx:3306/xx', 'connector.table' = 'jdbc_table_name', 'connector.driver' = 'com.mysql.jdbc.Driver', 'connector.username' = 'xxx', 'connector.password' = 'xxxxx' ); CREATE TABLE audi_cheaper_than_30w ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( 'connector.type' = 'dis', 'connector.region' = 'cn-north-1', 'connector.channel' = 'disOutput', 'connector.partition-key' = 'car_id,car_owner', 'format.type' = 'csv' ); INSERT INTO audi_cheaper_than_30w SELECT a.car_id, b.car_owner, b.car_brand, b.car_price FROM car_infos as a join db_info FOR SYSTEM_TIME AS OF a.proctime AS b on a.car_id = b.car_id;

语法格式 1 2 3 4 5 6 7 8 9 10 11 CREATE TABLE table_id ( attr_name attr_type (',' attr_name attr_type)* ) WITH ( 'connector.type' = 'jdbc', 'connector.url' = '', 'connector.table' = '', 'connector.username' = '', 'connector.password' = '' );

参数说明 请参考使用Hive语法创建OBS表，和Hive 文档了解每个DDL语句的语义。 表1 TBLPROPERTIES 参数说明 参数 是否必选 默认参数 数据类型 说明 streaming-source.enable 否 false Boolean 是否启用流源。 注意： 请确保每个分区/文件都应该以原子方式写入，否则读取器可能会得到不完整的数据。 streaming-source.partition.include 否 all String 设置分区读取的选项，支持的选项是 'all' 和 'latest'。默认情况下，该选项为 'all' 。 'all' 表示读取所有分区; 'latest'仅在流式处理 Hive 源表用作temporal table时才有效。'latest' 表示按'streaming-source.partition.order'的顺序读取最新的分区。 Flink 支持对最新的 hive 分区进行临时连接，通过启用 'streaming-source.enable'，并将 'streaming-source.partition.include' 设置为 'latest'。同时，用户可以通过配置以下分区相关选项来分配分区比较顺序和数据更新间隔。 streaming-source.monitor-interval 否 None Duration 连续监视分区/文件的时间间隔。注意：Hive 流式处理读取的默认间隔为'1 min'，Hive 流式处理temporal join的默认间隔为 '60 min'，这是因为在当前 Hive 流式处理临时联接实现中，每个 TM 都会访问 Hive metaStore，这可能会对 metaStore 产生压力，这将在未来得到改善。 streaming-source.partition-order 否 partition-name String 流源的分区顺序，支持 create-time、partition-time 和 partition-name。 create-time 比较分区/文件创建时间，这不是 Hive metaStore 中的分区创建时间，而是文件系统中的文件夹/文件修改时间，如果分区文件夹以某种方式更新，例如将新文件添加到文件夹中，可能会影响数据的使用方式。 partition-time 比较从分区名称中提取的时间。 partition-name 比较分区名称的字母顺序。 对于非分区表，此值应始终为“create-time”。 默认情况下，该值为 partition-name。该选项与已弃用的选项“streaming-source.consume-order”相等。 streaming-source.consume-start-offset 否 None String 流式处理消费的起始偏移量。如何解析和比较偏移量取决于您的订单。对于 create-time 和 partition-time，应为时间戳字符串 （yyyy-[m]m-[d]d [hh：mm：ss]）。 对于partition-time，将使用分区时间提取器从分区中提取时间。对于 partition-name，是分区名称字符串（例如 pt_year=2020/pt_mon=10/pt_day=01）。 is_lakehouse 否 无 Boolean 如果使用hive语法的DLI Lakehouse表，则需要设置is_lakehouse为true。 Source Parallelism Inference 默认情况下，Flink 会根据文件数量和每个文件中的块数来推断其 Hive 读取器的最佳并行度。 Flink 支持灵活配置并行推理策略。您可以在 TableConfig 中配置以下参数（请注意，这些参数会影响作业的所有源）： Key Default Type Description table.exec.hive.infer-source-parallelism true Boolean 如果为 true，则根据拆分数推断源并行度。如果为 false，则源的并行度由 config 设置。 table.exec.hive.infer-source-parallelism.max 1000 Integer 设置源运算符的最大推断并行度。 Load Partition Splits 多线程用于拆分 hive 的分区。您可以使用 table.exec.hive.load-partition-splits.thread-num 来配置线程号。默认值为 3，配置的值应大于 0。 Key Default Type Description table.exec.hive.load-partition-splits.thread-num 3 Integer 配置的值应大于0。 SQL 提示可用于将配置应用于 Hive 表，而无需更改其在 Hive 元存储中的定义。Hints | Apache Flink Vectorized Optimization upon Read 当满足以下条件时，Flink 会自动对 Hive 表进行矢量化读取： 格式：ORC 或 Parquet。 没有复杂数据类型的列，如配置单元类型：List、Map、Struct、Union。 默认情况下，此功能处于启用状态。可以使用以下配置禁用它。 table.exec.hive.fallback-mapred-reader=true

示例 使用Spark SQL创建Hive语法OBS表，并插入10条数据。模拟数据源。 CREATE TABLE IF NOT EXISTS demo.student( name STRING, score DOUBLE) PARTITIONED BY (classNo INT) STORED AS PARQUET LOCATION 'obs://demo/spark.db/student'; INSERT INTO demo.student PARTITION(classNo=1) VALUES ('Alice', 90.0), ('Bob', 80.0), ('Charlie', 70.0), ('David', 60.0), ('Eve', 50.0), ('Frank', 40.0), ('Grace', 30.0), ('Hank', 20.0), ('Ivy', 10.0), ('Jack', 0.0); 使用Flink SQL展示使用批的方式，从Hive语法OBS表demo.student中读取数据，并打印。需要开启checkpoint。 CREATE CATA LOG myhive WITH ( 'type' = 'hive', 'default-database' = 'demo', 'hive-conf-dir' = '/opt/flink/conf' ); USE CATALOG myhive; create table if not exists print ( name STRING, score DOUBLE, classNo INT) with ('connector' = 'print'); insert into print select * from student; 结果（taskmanager的out日志）： +I[Alice, 90.0, 1] +I[Bob, 80.0, 1] +I[Charlie, 70.0, 1] +I[David, 60.0, 1] +I[Eve, 50.0, 1] +I[Frank, 40.0, 1] +I[Grace, 30.0, 1] +I[Hank, 20.0, 1] +I[Ivy, 10.0, 1] +I[Jack, 0.0, 1] 使用Flink SQL展示使用流的方式，从Hive语法OBS表demo.student中读取数据，并打印。 CREATE CATALOG myhive WITH ( 'type' = 'hive' , 'default-database' = 'demo', 'hive-conf-dir' = '/opt/flink/conf' ); USE CATALOG myhive; create table if not exists print ( name STRING, score DOUBLE, classNo INT) with ('connector' = 'print'); insert into print select * from student /*+ OPTIONS('streaming-source.enable' = 'true', 'streaming-source.monitor-interval' = '3 m') */;

简介 Apache Hive 已经成为了 数据仓库 生态系统中的核心。 它不仅仅是一个用于大数据分析和ETL场景的SQL引擎，同样它也是一个数据管理平台，可用于发现，定义，和演化数据。 Flink与Hive的集成包含两个层面，一是利用了Hive的MetaStore作为持久化的Catalog，二是利用Flink来读写Hive的表。Overview | Apache Flink 从Flink 1.11.0开始，在使用 Hive方言时，Flink允许用户用Hive语法来编写SQL语句。通过提供与Hive语法的兼容性，改善与Hive的互操作性，并减少用户需要在Flink和Hive之间切换来执行不同语句的情况。详情可参考：Apache Flink Hive 方言 使用HiveCatalog，Apache Flink可以用于统一处理Apache Hive表的BATCH和STREAM。Flink可以作为Hive批处理引擎的更高效的替代方案，或者用于连续读写Hive表，以支持实时数据仓库应用程序。Apache Flink Hive Read & Write

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 数据类型的使用，请参考Format章节。 Hive 方言支持的 DDL 语句，Flink 1.15 当前仅支持使用Hive语法创建OBS表和使用hive语法的DLI Lakehouse表。 使用Hive语法创建OBS表 defalut方言： with 属性中需要设置hive.is-external为true。 使用hive 方言：建表语句需要使用EXTERNAL关键字。 使用hive语法的DLI Lakehouse表 使用hive 方言：表属性中需要添加'is_lakehouse'='true'。 开启checkpoint功能。 建议切换到Hive方言来创建Hive兼容表。如果你想用默认的方言创建Hive兼容表，确保在你的表属性中设置'connector'='hive'，否则在HiveCatalog中一个表默认被认为是通用的。注意，如果你使用Hive方言，就不需要connector属性了。 监视策略是扫描当前位置路径中的所有目录/文件。许多分区可能会导致性能下降。 对未分区表进行流式读取时，要求将每个文件以原子方式写入目标目录。 分区表的流式读取要求在 hive 元存储的视图中以原子方式添加每个分区。否则，将使用添加到现有分区的新数据。 流式读取不支持 Flink DDL 中的水印语法。这些表不能用于窗口运算符。

语法格式 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 CREATE EXTERNAL TABLE [IF NOT EXISTS] table_name [(col_name data_type [column_constraint] [COMMENT col_comment], ... [table_constraint])] [COMMENT table_comment] [PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)] [ [ROW FORMAT row_format] [STORED AS file_format] ] [LOCATION obs_path] [TBLPROPERTIES (property_name=property_value, ...)] row_format: : DELIMITED [FIELDS TERMINATED BY char [ESCAPED BY char]] [COLLECTION ITEMS TERMINATED BY char] [MAP KEYS TERMINATED BY char] [LINES TERMINATED BY char] [NULL DEFINED AS char] | SERDE serde_name [WITH SERDEPROPERTIES (property_name=property_value, ...)] file_format: : SEQUENCEFILE | TEXTFILE | RCFILE | ORC | PARQUET | AVRO | INPUTFORMAT input_format_classname OUTPUTFORMAT output_format_classname column_constraint: : NOT NULL [[ENABLE|DISABLE] [VALIDATE|NOVALIDATE] [RELY|NORELY]] table_constraint: : [CONSTRAINT constraint_name] PRIMARY KEY (col_name, ...) [[ENABLE|DISABLE] [VALIDATE|NOVALIDATE] [RELY|NORELY]]

功能描述 本节介绍利用Flink来读写Hive的表。Hive源表的定义，以及创建源表时使用的参数和示例代码。详情可参考：Apache Flink Hive Read & Write Flink支持在BATCH 和 STREAMING模式下从Hive读取数据。当作为BATCH应用程序运行时，Flink将在执行查询的时间点对表的状态执行查询。STREAMING读取将持续监控表，并在新数据可用时以增量方式获取新数据。默认情况下，Flink会读取有界的表。 STREAMING读取支持同时使用分区表和非分区表。对于分区表，Flink将监控新分区的生成，并在可用时增量读取它们。对于未分区的表，Flink 会监控文件夹中新文件的生成情况，并增量读取新文件。

功能描述 Debezium是一个 CDC（Changelog Data Capture，变更数据捕获）的工具，可以把来自 MySQL、PostgreSQL、Oracle、Microsoft SQL Server 和许多其他数据库的更改实时流式传输到 Kafka 中。 Debezium 为变更日志提供了统一的格式结构，并支持使用 JSON 和 Apache Avro 序列化消息。 Flink 支持将 Debezium JSON 和 Avro 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中。在很多情况下，利用这个特性非常的有用，例如： 将增量数据从数据库同步到其他系统 日志审计 数据库的实时物化视图 关联维度数据库的变更历史 Flink 还支持将 Flink SQL 中的 INSERT / UPDATE / DELETE 消息编码为 Debezium 格式的 JSON 或 Avro 消息，输出到 Kafka 等存储中。 但需要注意的是，目前 Flink 还不支持将 UPDATE_BEFORE 和 UPDATE_AFTER 合并为一条 UPDATE 消息。因此，Flink 将 UPDATE_BEFORE 和 UPDATE_AFTER 分别编码为 DELETE 和 INSERT 类型的 Debezium 消息。 更多具体使用可参考开源社区文档：Debezium Format。

示例 示例1：该示例是从DMS Kafka数据源中读取数据，并写入到Print结果表中。 参考增强型跨源连接，根据Kafka所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根据Kafka的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业脚本，提交运行作业。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE upsertKafkaSource ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'KafkaTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkAddress2:KafkaPort', 'key.format' = 'csv', 'value.format' = 'json' ); CREATE TABLE printSink ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'print' ); INSERT INTO printSink SELECT * FROM upsertKafkaSource; 向Kafka中的指定topic中插入如下数据（注意：kafka插入数据时请指定key）。 {"order_id":"202303251202020001", "order_channel":"miniAppShop", "order_time":"2023-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2023-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"} {"order_id":"202303251505050001", "order_channel":"appshop", "order_time":"2023-03-25 15:05:05", "pay_amount":"500.00", "real_pay":"400.00", "pay_time":"2023-03-25 15:10:00", "user_id":"0003", "user_name":"Cindy", "area_id":"330108"} {"order_id":"202303251202020001", "order_channel":"miniAppShop", "order_time":"2023-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2023-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330111"} 查看taskmanager的out文件，数据结果参考如下： +I(202303251202020001,miniAppShop,2023-03-2512:02:02,60.0,60.0,2023-03-2512:03:00,0002,Bob,330110) +I(202303251505050001,appshop,2023-03-25 15:05:05,500.0,400.0,2023-03-2515:10:00,0003,Cindy,330108) -U(202303251202020001,miniAppShop,2023-03-2512:02:02,60.0,60.0,2023-03-2512:03:00,0002,Bob,330110) +U(202303251202020001,miniAppShop,2023-03-2512:02:02,60.0,60.0,2023-03-2512:03:00,0002,Bob,330111)

语法格式 1 2 3 4 5 6 7 8 9 10 11 12 create table kafkaTable( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'upsert-kafka', 'topic' = '', 'properties.bootstrap.servers' = '', 'key.format' = '', 'value.format' = '' );

参数说明 表2 参数说明 参数 是否必选 默认参数 数据类型 说明 connector 是 无 String connector类型，对于upsert kafka连接器，需配置为'upsert-kafka'。 topic 是 无 String Kafka topic名。 properties.bootstrap.servers 是 无 String Kafka brokers地址，以逗号分隔。 key.format 是 无 String 用于对Kafka消息中key部分序列化和反序列化的格式。key字段由PRIMARY KEY语法指定。支持的格式如下： csv json avro 请参考Format页面以获取更多详细信息和格式参数。 key.fields-prefix 否 无 String 为键格式的所有字段定义自定义前缀，以避免与值格式的字段发生名称冲突。 默认情况下，前缀为空。如果定义了自定义前缀，则表架构和'key.fields'都将使用前缀名称。在构造密钥格式的数据类型时，将删除前缀，并在密钥格式中使用无前缀的名称。请注意，此选项要求'value.fields-include' 必须设置为'EXCEPT_KEY'。 value.format 是 无 String 用于对 Kafka消息中 value 部分序列化和反序列化的格式。支持的格式： csv json avro 请参考Format页面以获取更多详细信息和格式参数。 value.fields-include 是 ALL String 控制哪些字段应该出现在值中。取值范围如下： ALL：消息的value部分将包含schema的所有字段，包括定义中键的字段。 EXCEPT_KEY：记录的value部分包含schema的所有内容，定义为主键的字段除外。 properties.* 否 无 String 该选项可以传递任意的Kafka参数。 “properties.”后的后缀名必须匹配定义在 kafka参数文档中的参数名。 Flink会自动移除选项名中的 "properties." 前缀，并将转换后的键名以及值传入KafkaClient。 例如：你可以通过 'properties.allow.auto.create.topics' = 'false' 来禁止自动创建 topic。 但是'key.deserializer' 和 'value.deserializer' 是不允许通过该方式传递参数，因为Flink会重写这些参数的值。 sink.parallelism 否 无 Integer 定义upsert-kafka sink 算子的并行度。默认情况下，由框架确定并行度，与上游链接算子的并行度保持一致。 sink.buffer-flush.max-rows 否 0 Integer 缓存刷新前，最多能缓存的记录条数。 当sink收到很多同key上的更新时，缓存将保留同 key 的最后一条记录，因此sink缓存能帮助减少发往Kafka topic的数据量，以及避免发送潜在的tombstone消息。可以通过设置为'0'来禁用它。 默认情况下，该选项是未开启的。如果要开启 sink 缓存，需要同时设置'sink.buffer-flush.max-rows'和'sink.buffer-flush.interval'两个选项为大于零的值。 sink.buffer-flush.interval 否 0 Duration 缓存刷新的间隔时间，超过该时间后异步线程将刷新缓存数据。单位可以为毫秒（ms）、秒（s）、分钟（min）或小时（h）。例如'sink.buffer-flush.interval'='10 ms'。 默认情况下，该选项是未开启的。如果要开启 sink 缓存，需要同时设置'sink.buffer-flush.max-rows'和'sink.buffer-flush.interval'两个选项为大于零的值。

功能描述 Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。Upsert Kafka 连接器支持以upsert方式从Kafka topic中读取数据并将数据写入Kafka topic。表类型支持源表和结果表。 作为source，upsert-kafka 连接器生产changelog流，其中每条数据记录代表一个更新或删除事件。 数据记录中的value被解释为同一key的最后一个value的UPDATE，如果有这个key（如果不存在相应的key，则该更新被视为INSERT）。用表来类比，changelog 流中的数据记录被解释为UPSERT，也称为INSERT/UPDATE，因为任何具有相同key的现有行都被覆盖。另外，value为空的消息将会被视作为DELETE消息。 作为sink，upsert-kafka连接器可以消费changelog流。它会将INSERT/UPDATE_AFTER数据作为正常的Kafka消息写入，并将DELETE数据以value为空的Kafka消息写入（表示对应 key 的消息被删除）。Flink将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。 表1 支持类别 类别 详情 支持表类型 源表、结果表

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 Upsert Kafka 始终以upsert方式工作，并且需要在DDL中定义主键。在具有相同主键值的消息按序存储在同一个分区的前提下，在 changlog source 定义主键意味着 在物化后的 changelog 上主键具有唯一性。定义的主键将决定哪些字段出现在Kafka消息的key中。 由于该连接器以 upsert 的模式工作，该连接器作为 source 读入时，可以确保具有相同主键值下仅最后一条消息会生效。 数据类型的使用，请参考Format章节。

示例 RDS表用于与输入流连接。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 CREATE TABLE car_infos ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT, proctime as PROCTIME() ) WITH ( 'connector.type' = 'dis', 'connector.region' = 'cn-north-1', 'connector.channel' = 'disInput', 'format.type' = 'csv' ); CREATE TABLE db_info ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( 'connector.type' = 'gaussdb', 'connector.driver' = 'org.postgresql.Driver', 'connector.url' = 'jdbc:gaussdb://xx.xx.xx.xx:8000/xx', 'connector.table' = 'car_info', 'connector.username' = 'xx', 'connector.password' = 'xx', 'connector.lookup.cache.max-rows' = '10000', 'connector.lookup.cache.ttl' = '24h' ); CREATE TABLE audi_cheaper_than_30w ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( 'connector.type' = 'dis', 'connector.region' = 'cn-north-1', 'connector.channel' = 'disOutput', 'connector.partition-key' = 'car_id,car_owner', 'format.type' = 'csv' ); INSERT INTO audi_cheaper_than_30w SELECT a.car_id, b.car_owner, b.car_brand, b.car_price FROM car_infos as a join db_info FOR SYSTEM_TIME AS OF a.proctime AS b on a.car_id = b.car_id;

参数说明 表1 参数说明 参数 是否必选 说明 connector.type 是 connector类型，需配置为'gaussdb' connector.url 是 jdbc连接地址，格式为：jdbc:postgresql://${ip}:${port}/${dbName} 。 connector.table 是 读取数据库中的数据所在的表名 connector.driver 否 jdbc连接驱动，默认为: org.postgresql.Driver。 connector.username 否 数据库认证用户名，需要和'connector.password'一起配置 connector.password 否 数据库认证密码，需要和'connector.username'一起配置 connector.read.partition.column 否 用于对输入进行分区的列名 与connector.read.partition.lower-bound、connector.read.partition.upper-bound、 connector.read.partition.num必须同时存在或者同时不存在 connector.read.partition.lower-bound 否 第一个分区的最小值 与connector.read.partition.column、connector.read.partition.upper-bound、 connector.read.partition.num必须同时存在或者同时不存在 connector.read.partition.upper-bound 否 最后一个分区的最大值 与connector.read.partition.column、connector.read.partition.lower-bound、 connector.read.partition.num必须同时存在或者同时不存在 connector.read.partition.num 否 分区的个数 与connector.read.partition.column、connector.read.partition.upper-bound、 connector.read.partition.upper-bound必须同时存在或者同时不存在 connector.read.fetch-size 否 每次从数据库拉取数据的行数。默认值为0，表示忽略该提示 connector.lookup.cache.max-rows 否 维表配置，缓存的最大行数，超过该值时，最先添加的数据将被标记为过期。-1表示不使用缓存。 connector.lookup.cache.ttl 否 维表配置，缓存超时时间，超过该时间的数据会被剔除。格式为：{length value}{time unit label}，如123ms, 321s，支持的时间单位包括: d,h,min,s,ms等，默认为ms。 connector.lookup.max-retries 否 维表配置，数据拉取最大重试次数，默认为3。

语法格式 1 2 3 4 5 6 7 8 9 10 11 create table dwsSource ( attr_name attr_type (',' attr_name attr_type)* ) with ( 'connector.type' = 'gaussdb', 'connector.url' = '', 'connector.table' = '', 'connector.username' = '', 'connector.password' = '' );

示例 示例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)