华为云用户手册

语法格式 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

DMS输入流 分布式消息服务（Distributed Message Service，简称DMS）是一项基于高可用分布式集群技术的消息中间件服务，提供了可靠且可扩展的托管消息队列，用于收发消息和存储消息。 分布式消息服务Kafka是一款基于开源社区版Kafka提供的消息队列服务，向用户提供可靠的全托管式的Kafka消息队列。 DLI 支持创建输入流从DMS的Kafka获取数据，作为作业的输入数据。 创建DMS Kafka输入流的语法与创建开源Apache Kafka输入流一样，具体请参见开源Kafka输入流。 父主题： 创建输入流

注意事项 使用文件系统输出流的Flink作业必须开启checkpoint，保证作业的一致性。 为了避免数据丢失或者数据被覆盖，开启作业异常自动重启或者手动重启，需要配置为“从checkpoint恢复”。 checkpoint间隔设置需在输出文件实时性、文件大小和恢复时长之间进行权衡，比如10分钟。 checkpoint支持如下两种模式： AtLeastOnce：事件至少被处理一次。 ExactlyOnce：事件仅被处理一次。 使用文件系统输出流写入数据到OBS时，应避免多个作业写同一个目录的情况。 OBS对象存储桶的默认行为为覆盖写，可能导致数据丢失。 OBS并行文件系统桶的默认行为追加写，可能导致数据混淆。 因为以上OBS桶类型行为的区别，为避免作业异常重启可能导致的数据异常问题，请根据您的业务需求选择OBS桶类型。

HDFS代理用户配置 登录 MRS 管理页面。 选择MRS的HDFS Namenode配置，在“自定义”中添加配置参数。 图1 HDFS服务配置 其中，core-site值名称“hadoop.proxyuser.myname.hosts”和“hadoop.proxyuser.myname.groups”中的“myname”为传入的krb认证用户名称。 需要保证写入HDFS数据路径权限为777。 配置完成后，单击“保存配置”进行保存。

示例 示例一： 该示例将car_info数据，以buyday字段为分区字段，parquet为编码格式，转储数据到OBS。 1 2 3 4 5 6 7 8 9 10 11 12 13 create sink stream car_infos ( carId string, carOwner string, average_speed double, buyday string ) partitioned by (buyday) with ( type = "filesystem", file.path = "obs://obs-sink/car_infos", encode = "parquet", ak = "{{myAk}}", sk = "{{mySk}}" ); 数据最终在OBS中的存储目录结构为：obs://obs-sink/car_infos/buyday=xx/part-x-x。 数据生成后，可通过如下SQL语句建立OBS分区表，用于后续批处理： 创建OBS分区表。 1 2 3 4 5 6 7 8 create table car_infos ( carId string, carOwner string, average_speed double ) partitioned by (buyday string) stored as parquet location 'obs://obs-sink/car_infos'; 从关联OBS路径中恢复分区信息。 1 alter table car_infos recover partitions; 示例二 该示例将car_info数据，以buyday字段为分区字段，csv为编码格式，转储数据到HDFS。 1 2 3 4 5 6 7 8 9 10 11 12 create sink stream car_infos ( carId string, carOwner string, average_speed double, buyday string ) partitioned by (buyday) with ( type = "filesystem", file.path = "hdfs://node-master1sYAx:9820/user/car_infos", encode = "csv", field_delimiter = "," ); 数据最终在HDFS中的存储目录结构为：/user/car_infos/buyday=xx/part-x-x。

功能描述 创建sink流将数据输出到分布式文件系统(HDFS)或者 对象存储服务 （OBS）等文件系统。数据生成后，可直接对生成的目录创建非DLI表，通过DLI SQL进行下一步处理分析，并且输出数据目录支持分区表结构。适用于数据转储、大数据分析、备份或活跃归档、深度或冷归档等场景。 对象存储服务（Object Storage Service，简称OBS）是一个基于对象的海量存储服务，为客户提供海量、安全、高可靠、低成本的数据存储能力。

语法格式 1 2 3 4 5 6 7 8 9 CREATE SINK STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* ) [PARTITIONED BY (attr_name (',' attr_name)*] WITH ( type = "filesystem", file.path = "obs://bucket/xx", encode = "parquet", ak = "", sk = "" );

关键字 表1 关键字说明 参数 是否必选 说明 type 是 输出流类型。“type”为“filesystem”，表示输出数据到文件系统。 file.path 是 输出目录，格式为: schema://file.path。 当前schame只支持obs和hdfs。 当schema为obs时，表示输出到对象存储服务OBS。 当schema为hdfs时，表示输出到HDFS。HDFS需要配置代理用户，具体请参考HDFS代理用户配置。 示例：hdfs://node-master1sYAx:9820/user/car_infos，其中node-master1sYAx:9820为MRS集群NameNode所在节点信息。 encode 是 输出数据编码格式，当前支持“parquet”格式和“csv”格式。 当schema为obs时，输出数据编码格式仅支持“parquet”格式。 当schema为hdfs时，输出数据编码格式支持“parquet”格式和“csv”格式。 ak 否 输出到OBS时该参数必填。用于访问OBS认证的accessKey，可使用全局变量，屏蔽敏感信息。 关于全局变量在控制台上的使用方法，请参考《 数据湖探索 用户指南》。 sk 否 输出到OBS时该参数必填。用于访问OBS认证的secretKey，可使用全局变量，屏蔽敏感信息。 关于全局变量在控制台上的使用方法，请参考《 数据湖 探索用户指南》。 krb_auth 否 创建跨源认证的认证名。开启kerberos认证时，需配置该参数。如果创建的MRS集群未开启kerb认证的集群，请确保在DLI队列host文件中添加MRS集群master节点的“/etc/hosts”信息。 field_delimiter 否 属性分隔符。 当编码格式为“csv”时，需要设置属性分隔符，用户可以自定义，如：“,”。

示例 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 SOURCE STREAM car_infos ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( type = "dis", region = "", channel = "dliinput", encode = "csv", field_delimiter = "," ); CREATE TABLE db_info ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( type = "rds", username = "root", password = "******", db_url = "postgresql://192.168.0.0:2000/test1", table_name = "car" ); CREATE SINK STREAM audi_cheaper_than_30w ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( type = "dis", region = "", channel = "dlioutput", partition_key = "car_owner", encode = "csv", field_delimiter = "," ); 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 as b on a.car_id = b.car_id; 将数据库连接地址设置为DWS数据库地址，即可创建DWS维表。DWS数据库版本大于8.1.0后，无法用开源的postgresql驱动连接，需要用gaussdb驱动进行连接。

前提条件 请务必确保您的账户下已在关系型数据库（RDS）里创建了PostgreSQL或MySQL类型的RDS实例。 如何创建RDS实例，请参见《关系型数据库快速入门》中“购买实例”章节。 该场景作业需要运行在DLI的独享队列上，因此要与RDS实例建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

关键字 表1 关键字说明 参数 是否必选 说明 type 是 输出通道类型，rds表示输出到关系型数据库中。 username 是 数据库连接用户名。 password 是 数据库连接密码。 db_url 是 数据库连接地址，格式为："{database_type}://ip:port/database" 目前支持两种数据库连接：MySQL和PostgreSQL MySQL: 'mysql://ip:port/database' PostgreSQL: 'postgresql://ip:port/database' 说明： 将数据库连接地址设置为DWS数据库地址，即可创建DWS维表。DWS数据库版本大于8.1.0后，无法用开源的postgresql驱动连接，需要用gaussdb驱动进行连接。 table_name 是 用于查询数据的数据库表名。 db_columns 否 流属性和数据库表的字段对应关系。当sink流中流属性字段和数据库表中的流属性字段不完全匹配时，该参数必配。格式为“dbtable_attr1,dbtable_attr2,dbtable_attr3“。 cache_max_num 否 表示最大缓存的查询结果数，默认值为32768。 cache_time 否 表示数据库查询结果在内存中缓存的最大时间。单位为毫秒，默认值为10000，当值为0时表示不缓存。

语法格式 1 2 3 4 5 6 7 8 9 10 11 12 13 CREATE TABLE table_id ( car_id STRING, car_owner STRING, car_brand STRING, car_price INT ) WITH ( type = "rds", username = "", password = "", db_url = "", table_name = "" );

关键字 表1 关键字说明 参数 是否必选 说明 type 是 输出通道类型，dcs_redis表示输出到分布式缓存服务的Redis存储系统中。 cluster_address 是 Redis实例连接地址。 password 否 Redis实例连接密码，当设置为免密访问时，省略该配置项。 value_type 是 指定数据类型。支持的数据类型包括：string, list, hash, set, zset。 key_column 是 指定代表Redis key属性的列名。 hash_key_column 否 当value_type设置为hash时，需要指定本字段作为第二级key属性的列名。 cache_max_num 否 表示最大缓存的查询结果数，默认值为32768。 cache_time 否 表示数据库查询结果在内存中缓存的最大时间。单位为毫秒，默认值为10000，当值为0时表示不缓存。

注意事项 不支持Redis集群。 请务必确保您的账户下已在分布式缓存服务（D CS ）里创建了Redis类型的缓存实例。 如何创建Redis类型的缓存实例请参考《分布式缓存服务用户指南》。 该场景作业需要运行在DLI的独享队列上，因此要与DCS实例建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

示例 Redis表用于与输入流连接。 CREATE TABLE table_a (attr1 string, attr2 string, attr3 string) WITH ( type = "dcs_redis", value_type = "hash", key_column = "attr1", hash_key_column = "attr2", cluster_address = "192.168.1.238:6379", password = "xxxxxxxx" );

语法格式 1 2 3 4 5 6 CREATE TABLE table_id (key_attr_name STRING(, hash_key_attr_name STRING)?, value_attr_name STRING) WITH ( type = "dcs_redis", cluster_address = ""(,password = "")?, value_type= "", key_column= ""(,hash_key_column="")?);

示例 将流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" );

语法格式 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" )

关键字 表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”属性中配置。

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

关键字 表1 关键字说明 参数 是否必选 说明 type 是 输出通道类型，"mrs_hbase"表示输出到MRS的HBase中。 region 是 MRS服务所在区域。 cluster_address 是 待插入数据表所属集群zookeeper地址，形如：ip1,ip2:port。 table_name 是 待插入数据的表名。 支持参数化，例如当需要某一列或者几列作为表名的一部分时，可表示为”car_pass_inspect_with_age_${car_age}“，其中car_age为列名。 table_columns 是 待插入的列，具体形式如："rowKey,f1:c1,f1:c2,f2:c1"，其中必须指定rowKey，当某列不需要加入数据库时，以第三列为例，可表示为"rowKey,f1:c1,,f2:c1"。 illegal_data_table 否 如果指定该参数，异常数据（比如：rowKey不存在）会写入该表（rowKey为taskNo加下划线加时间戳加六位随机数字，schema为info:data, info:reason），否则会丢弃。 batch_insert_data_num 否 表示一次性批量写入的数据条数，值必须为正整数，上限为1000，默认值为10。 action 否 表示数据是插入还是删除，可选值为add和delete，默认值为add。 krb_auth 否 创建跨源认证的认证名。开启kerberos认证时，需配置该参数，填写对应的跨源认证名称。跨源认证创建详见创建跨源认证。 说明： 请确保在DLI队列host文件中添加MRS集群master节点的“/etc/hosts”信息。

示例 将数据输出到MRS的HBase中。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 CREATE SINK STREAM qualified_cars ( car_id STRING, car_owner STRING, car_age INT, average_speed INT, total_miles INT ) WITH ( type = "mrs_hbase", region = "xxx", cluster_address = "192.16.0.88,192.87.3.88:2181", table_name = "car_pass_inspect_with_age_${car_age}", table_columns = "rowKey,info:owner,,car:speed,car:miles", illegal_data_table = "illegal_data", batch_insert_data_num = "20", action = "add", krb_auth = "KRB_AUTH_NAME" );

语法格式 1 2 3 4 5 6 7 8 9 10 11 CREATE SINK STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* ) WITH ( type = "mrs_hbase", region = "", cluster_address = "", table_name = "", table_columns = "", illegal_data_table = "", batch_insert_data_num = "", action = "" )

前提条件 确保您的账户下已在 MapReduce服务 （MRS）里创建了您配置的集群。DLI支持与开启kerberos的hbase集群对接。 该场景作业需要运行在DLI的独享队列上，请确保已创建DLI独享队列。 关于如何创建DLI独享队列，在购买队列时，选择“按需计费”，勾选“专属资源模式”即可。具体操作请参见《数据湖探索用户指南》中创建队列章节。 确保DLI独享队列与MRS集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。 若使用MRS HBase，请在增强型跨源的主机信息中添加MRS集群所有节点的主机ip信息。 如何添加IP域名映射，请参见《数据湖探索用户指南》中修改主机信息章节。

示例 将数据输出到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" ); 示例二 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 CREATE SINK STREAM kafka_sink ( a1 string, a2 string, a3 string, a4 INT ) // 输出字段 WITH ( type="kafka", kafka_bootstrap_servers = "192.x.x.x:9093, 192.x.x.x:9093, 192.x.x.x:9093", kafka_topic = "testflink", // 写入的topic encode = "csv", // 编码格式，支持json/csv kafka_certificate_name = "Flink", kafka_properties_delimiter = ",", kafka_properties = "sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"xxx\" password=\"xxx\";,sasl.mechanism=PLAIN,security.protocol=SASL_SSL" );

前提条件 Kafka服务端的端口如果监听在hostname上，则需要将Kafka Broker节点的hostname和IP的对应关系添加到DLI队列中。Kafka Broker节点的hostname和IP请联系Kafka服务的部署人员。如何添加IP域名映射，请参见《数据湖探索用户指南》中修改主机信息章节。 Kafka是线下集群，需要通过增强型跨源连接功能将Flink作业与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" )

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

示例 将流qualified_cars的数据输出到DCS服务的Redis类型的缓存实例中。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 CREATE SINK STREAM qualified_cars ( car_id STRING, car_owner STRING, car_age INT, average_speed DOUBLE, total_miles DOUBLE ) WITH ( type = "dcs_redis", cluster_address = "192.168.0.34:6379", password = "xxxxxxxx", value_type = "string; list; hash; set; zset", key_value = "${car_id}_str: ${car_owner}; name_list: ${car_owner}; ${car_id}_hash: {name:${car_owner}, age: ${car_age}}; name_set: ${car_owner}; math_zset: {${car_owner}:${average_speed}}" );

关键字 表1 关键字说明 参数 是否必选 说明 type 是 输出通道类型，dcs_redis表示输出到分布式缓存服务的Redis存储系统中。 region 是 数据所在的DCS所在区域。 cluster_address 是 Redis实例连接地址。 password 否 Redis实例连接密码，当设置为免密访问时，省略该配置项。 value_type 是 该参数可配置为如下选项或选项的组合： 支持指定插入数据类型，包括：string, list, hash, set, zset； 支持设置key的过期时间，包括expire, pexpire, expireAt, pexpireAt； 支持删除key命令，包括del, hdel； 当需要使用多个命令时，用“;”分隔。 key_value 是 设置具体的key和value，key_value对必须与value_type所指定的类型数相对应，用“;”分隔，且key和value均支持参数化，动态列名采用${列名}表示。