华为云用户手册

数据湖探索 DLI-Upsert Kafka源表:语法格式

语法格式 1 2 3 4 5 6 7 8 9 10 11 12 create table kafkaSource( 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' = '' );

数据湖探索 DLI
数据湖探索 DLI-Upsert Kafka源表:注意事项

注意事项创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 Upsert Kafka 始终以upsert方式工作，并且需要在DDL中定义主键。在具有相同主键值的消息按序存储在同一个分区的前提下，在 changlog source 定义主键意味着在物化后的 changelog 上主键具有唯一性。定义的主键将决定哪些字段出现在Kafka消息的key中。由于该连接器以 upsert 的模式工作，该连接器作为 source 读入时，可以确保具有相同主键值下仅最后一条消息会生效。数据类型的使用，请参考Format章节。

数据湖探索 DLI
数据湖探索 DLI-Upsert Kafka源表:功能描述

功能描述 Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。作为 source，upsert-kafka 连接器生产changelog流，其中每条数据记录代表一个更新或删除事件。更准确地说，数据记录中的 value 被解释为同一 key 的最后一个 value 的 UPDATE，如果有这个 key（如果不存在相应的 key，则该更新被视为 INSERT）。用表来类比，changelog 流中的数据记录被解释为 UPSERT，也称为 INSERT/UPDATE，因为任何具有相同 key 的现有行都被覆盖。另外，value 为空的消息将会被视作为 DELETE 消息。

数据湖探索 DLI
数据湖探索 DLI-Upsert Kafka源表:前提条件

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

数据湖探索 DLI
数据湖探索 DLI-SMN输出流:示例

示例将流over_speed_warning的数据输出到消息通知服务SMN中。 1 2 3 4 5 6 7 8 9 10 11 //静态主题配置 CREATE SINK STREAM over_speed_warning ( over_speed_message STRING /* over speed message */ ) WITH ( type = "smn", region = "xxx", topic_Urn = "xxx", message_subject = "message title", message_column = "over_speed_message" ); 1 2 3 4 5 6 7 8 9 10 11 12 //动态主题配置 CREATE SINK STREAM over_speed_warning2 ( over_speed_message STRING, /* over speed message */ over_speed_urn STRING ) WITH ( type = "smn", region = "xxx", urn_column = "over_speed_urn", message_subject = "message title", message_column = "over_speed_message" );

数据湖探索 DLI
数据湖探索 DLI-SMN输出流:功能描述

功能描述 DLI将Flink作业的输出数据输出到消息通知服务（SMN）中。消息通知服务（Simple Message Notification，简称SMN）为DLI提供可靠的、可扩展的、海量的消息处理服务，它大大简化系统耦合，能够根据用户的需求，向订阅终端主动推送消息。可用于连接云服务、向多个协议推送消息以及集成在产生或使用通知的任何其他应用程序等场景。 SMN的更多信息，请参见《消息通知服务用户指南》。

数据湖探索 DLI
数据湖探索 DLI-SMN输出流:语法格式

语法格式 1 2 3 4 5 6 7 8 9 CREATE SINK STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* ) WITH( type = "smn", region = "", topic_urn = "", urn_column = "", message_subject = "", message_column = "" )

数据湖探索 DLI
数据湖探索 DLI-SMN输出流:关键字

关键字表1 关键字说明参数是否必选说明 type 是输出通道类型，smn表示输出到消息通知服务中。 region 是 SMN所在区域。 topic_urn 否 SMN服务的主题URN，用于静态主题URN配置。作为消息通知的目标主题，需要提前在SMN服务中创建。与“urn_column”配置两者至少存在一个，同时配置时，“topic_urn”优先级更高。 urn_column 否主题URN内容的字段名，用于动态主题URN配置。与“topic_urn”配置两者至少存在一个，同时配置时，“topic_urn”优先级更高。 message_subject 是发往SMN服务的消息标题，用户自定义。 message_column 是输出流的字段名，其内容作为消息的内容，用户自定义。目前只支持默认的文本消息。

数据湖探索 DLI
数据湖探索 DLI-Upsert Kafka结果表:语法格式

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

数据湖探索 DLI
数据湖探索 DLI-Upsert Kafka结果表:参数说明

参数说明表1 参数说明参数是否必选说明 connector.type 是 connector类型，对于upsert kafka，需配置为'upsert-kafka' connector.version 否 Kafka版本，仅支持：'0.11' format.type 是数据序列化格式，支持：'csv', 'json'及'avro'等 connector.topic 是 kafka topic名 connector.properties.bootstrap.servers 是 kafka brokers地址，以逗号分隔 connector.sink-partitioner 否记录分区方式，支持：'fixed', 'round-robin'及'custom' connector.sink-partitioner-class 否 'sink-partitioner'为'custom'时，需配置，如'org.mycompany.MyPartitioner' connector.sink.ignore-retraction 否是否忽略回撤消息，默认为false。回撤消息将以null值写入kafka update-mode 否支持：'append', 'retract'及'upsert'三种写入模式 connector.properties.* 否配置kafka任意原生属性

数据湖探索 DLI
数据湖探索 DLI-Upsert Kafka结果表:示例

示例 create table upsertKafkaSink( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT, primary key (car_id) not enforced ) with ( 'connector.type' = 'upsert-kafka', 'connector.version' = '0.11', 'connector.topic' = 'test-topic', 'connector.properties.bootstrap.servers' = 'xx.xx.xx.xx:9092', 'format.type' = 'csv' );

数据湖探索 DLI
数据湖探索 DLI-MySQL CDC源表:常见问题

常见问题 Q：MySQL CDC源表不支持定义Watermark，怎么进行窗口聚合？ A：可以采用非窗口聚合的方式，即将时间字段转换成窗口值，然后根据窗口值进行GROUP BY聚合。例如：基于上述示例，统计每分钟的订单数，脚本如下（其中order_time为string类型，表示订单的时间）。 insert into printSink select DATE_FORMAT(order_time, 'yyyy-MM-dd HH:mm'), count(*) from mysqlCdcSource group by DATE_FORMAT(order_time, 'yyyy-MM-dd HH:mm');

数据湖探索 DLI
数据湖探索 DLI-MySQL CDC源表:前提条件

前提条件 MySQL CDC要求MySQL版本为5.7或8.0.x。该场景作业需要DLI与MySQL建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。 Flink跨源开发场景中直接配置跨源认证信息存在密码泄露的风险，优先推荐您使用DLI提供的跨源认证。跨源认证简介及操作方法请参考跨源认证简介。 MySQL已开启了Binlog，并且binlog_row_image设置为FULL。已创建MySQL用户，并授予了SELECT、 SHOW DATABASES 、REPLICATION SLAVE和REPLICATION CLIENT权限。

数据湖探索 DLI
数据湖探索 DLI-MySQL CDC源表:参数说明

参数说明表1 参数说明参数是否必选默认值数据类型说明 connector 是无 String connector类型，需配置为'mysql-cdc'。 hostname 是无 String MySQL数据库的IP地址或者Hostname。 username 是无 String MySQL数据库的用户名。 password 是无 String MySQL数据库的密码。 database-name 是无 String 访问的数据库名称。数据库名称支持正则表达式以读取多个数据库的数据，例如flink(.)*表示以flink开头的数据库名。 table-name 是无 String 访问的表名。表名支持正则表达式以读取多个表的数据，例如cdc_order(.)*表示以cdc_order开头的表名。 port 否 3306 Integer MySQL数据库的端口号。 server-id 否 5400~6000随机值 String 数据库客户端的一个数字ID，该ID必须是MySQL集群中全局唯一的。建议针对同一个数据库的每个作业都设置一个不同的ID。默认会随机生成一个5400~6400的值。 scan.startup.mode 否 initial String 消费数据时的启动模式。 initial（默认）：在第一次启动时，会先扫描历史全量数据，然后读取最新的Binlog数据。 latest-offset：在第一次启动时，不会扫描历史全量数据，直接从Binlog的末尾（最新的Binlog处）开始读取，即只读取该Connector启动以后的最新变更。 server-time-zone 否无 String 数据库在使用的会话时区。例如：Asia/Shanghai。 pwd_auth_name 否无 String DLI侧创建的Password类型的跨源认证名称。使用跨源认证则无需在作业中配置置账号和密码。

数据湖探索 DLI
数据湖探索 DLI-MySQL CDC源表:注意事项

注意事项创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。同步数据库数据的客户端，都会有一个唯一ID，即Server ID。同一个数据库下，建议每个MySQL CDC作业配置不同的Server ID。主要原因如下： MySQL SERVER会根据该ID来维护网络连接以及Binlog位点。因此如果有大量相同的Server ID的客户端一起连接MySQL SERVER，可能导致MySQL SERVER的CPU陡增，影响线上业务稳定性。此外，多个作业共享相同的Server ID，会导致Binlog位点错乱，多读或少读数据，因此建议每个CDC作业都配置不同的Server ID。 MySQL CDC源表暂不支持定义Watermark。如果您需要进行窗口聚合，请参考常见问题描述。若连接DWS、MySQL等支持upsert的sink源，需要在sink表的创建语句中定义主键，请参考示例中printSink建表语句。

数据湖探索 DLI
数据湖探索 DLI-MySQL CDC源表:语法格式

语法格式 create table mySqlCdcSource ( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'mysql-cdc', 'hostname' = 'mysqlHostname', 'username' = 'mysqlUsername', 'password' = 'mysqlPassword', 'database-name' = 'mysqlDatabaseName', 'table-name' = 'mysqlTableName' );

数据湖探索 DLI
数据湖探索 DLI-自拓展输入流:语法格式

语法格式 1 2 3 4 5 6 7 CREATE SOURCE STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* ) WITH ( type = "user_defined", type_class_name = "", type_class_parameter = "" ) (TIMESTAMP BY timeindicator (',' timeindicator)?);timeindicator:PROCTIME '.' PROCTIME| ID '.' ROWTIME

数据湖探索 DLI 自拓展生态
数据湖探索 DLI-CloudTable HBase输入流:示例

示例从CloudTable的HBase中读取对象为car_infos的表。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 CREATE SOURCE STREAM car_infos ( car_id STRING, car_owner STRING, car_age INT, average_speed INT, total_miles INT ) WITH ( type = "cloudtable", region = "xxx", cluster_id = "209ab1b6-de25-4c48-8e1e-29e09d02de28", table_name = "carinfo", table_columns = "rowKey,info:owner,info:age,car:speed,car:miles" );

数据湖探索 DLI
数据湖探索 DLI-CloudTable HBase输入流:语法格式

语法格式 1 2 3 4 5 6 7 8 CREATE SOURCE STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* ) WITH ( type = "cloudtable", region = "", cluster_id = "", table_name = "", table_columns = "" );

数据湖探索 DLI
数据湖探索 DLI-CloudTable HBase输入流:功能描述

功能描述创建source流从表格存储服务CloudTable的HBase中获取数据，作为作业的输入数据。HBase是一个稳定可靠，性能卓越、可伸缩、面向列的分布式云存储系统，适用于海量数据存储以及分布式计算的场景，用户可以利用HBase搭建起TB至PB级数据规模的存储系统，对数据轻松进行过滤分析，毫秒级得到响应，快速发现数据价值。DLI可以从HBase中读取数据，用于过滤分析、数据转储等场景。表格存储服务（CloudTable），是基于Apache HBase提供的分布式、可伸缩、全托管的KeyValue数据存储服务，为DLI提供了高性能的随机读写能力，适用于海量结构化数据、半结构化数据以及时序数据的存储和查询应用，适用于物联网IOT应用和通用海量KeyValue数据存储与查询等场景。CloudTable的更多信息，请参见《表格存储服务用户指南》。

数据湖探索 DLI
数据湖探索 DLI-CloudTable HBase输入流:关键字

关键字表1 关键字说明参数是否必选说明 type 是数据源类型，“CloudTable”表示数据源为表格存储服务。 region 是表格存储服务所在区域。 cluster_id 是待读取数据表所属集群id。如何查看CloudTable的集群id，请参见《表格存储服务用户指南》中“查看集群基本信息”章节。 table_name 是待读取数据的表名，如需指定namespace，可表示为：namespace_name:table_name 。 table_columns 是待读取的列，具体形式如："rowKey,f1:c1,f1:c2,f2:c1"，并且保证与source相同的列数。

数据湖探索 DLI
数据湖探索 DLI-Hbase源表:数据类型映射

数据类型映射 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 不支持

数据湖探索 DLI
数据湖探索 DLI-Hbase源表:常见问题

常见问题 Q：Flink作业运行失败，作业运行日志中如下报错信息，应该怎么解决？ java.lang.IllegalArgumentException: offset (0) + length (8) exceed the capacity of the array: 6 A：如果HBase表中的数据是以其他方式导入的话，那么其存储是以String格式存储的，所以使用其他的数据格式将会报该错误。需要将Flink创建HBase源表中非string类型的字段的字段类型重新改为String即可。

数据湖探索 DLI
数据湖探索 DLI-Hbase源表:注意事项

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

数据湖探索 DLI
数据湖探索 DLI-Hbase源表:语法格式

语法格式 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' = '' );

数据湖探索 DLI
数据湖探索 DLI-Hbase源表:前提条件

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

数据湖探索 DLI
数据湖探索 DLI-开源Kafka输出流:示例

示例将流kafka_sink的数据输出到Kafka中。 1 2 3 4 5 6 7 CREATE SINK STREAM kafka_sink (name STRING) WITH ( type="kafka", kafka_bootstrap_servers = "ip1:port1,ip2:port2", kafka_topic = "testsink", encode = "json" );

数据湖探索 DLI
数据湖探索 DLI-开源Kafka输出流:关键字

关键字表1 关键字说明参数是否必选说明 type 是输出通道类型，"kafka"表示输出到Kafka中。 kafka_bootstrap_servers 是 Kafka的连接端口，需要确保能连通（需要通过增强型跨源开通DLI队列和Kafka集群的连接）。 kafka_topic 是写入的topic encode 是数据编码格式，可选为“csv”、“json”和“user_defined”。若编码格式为“csv”，则需配置“field_delimiter”属性。若编码格式为“user_defined”，则需配置“encode_class_name”和“encode_class_parameter”属性。 filed_delimiter 否当encode为csv时，用于指定各字段分隔符，默认为逗号。 encode_class_name 否当encode为user_defined时，需配置该参数，指定用户自实现编码类的类名（包含完整包路径），该类需继承类DeserializationSchema。 encode_class_parameter 否当encode为user_defined时，可以通过配置该参数指定用户自实现编码类的入参，仅支持一个string类型的参数。 kafka_properties 否可通过该参数配置kafka的原生属性，格式为"key1=value1;key2=value2" kafka_certificate_name 否跨源认证信息名称。跨源认证信息类型为“Kafka_SSL”时，该参数有效。说明：指定该配置项时，服务仅加载该认证下指定的文件和密码，系统将自动设置到“kafka_properties”属性中。 Kafka SSL认证需要的其他配置信息，需要用户手动在“kafka_properties”属性中配置。

数据湖探索 DLI
数据湖探索 DLI-开源Kafka输出流:语法格式

语法格式 1 2 3 4 5 6 7 CREATE SINK STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* ) WITH( type = "kafka", kafka_bootstrap_servers = "", kafka_topic = "", encode = "json" )

数据湖探索 DLI
数据湖探索 DLI-开源Kafka输出流:前提条件

前提条件 Kafka服务端的端口如果监听在hostname上，则需要将Kafka Broker节点的hostname和IP的对应关系添加到DLI队列中。Kafka Broker节点的hostname和IP请联系Kafka服务的部署人员。如何添加IP域名映射，请参见《数据湖探索用户指南》中修改主机信息章节。 Kafka是线下集群，需要通过增强型跨源连接功能将Flink作业与Kafka进行对接。且用户可以根据实际所需设置相应安全组规则。如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

数据湖探索 DLI

共100000条

undefined

意见反馈

0/200

提交取消

提交成功！非常感谢您的反馈，我们会继续努力做到更好反馈提交失败！请稍后重试！

华为云用户手册

7*24

备案

专业服务

退订

建议反馈

售前咨询热线