云服务器内容精选
-
前提条件 确保已创建kafka集群。 该场景作业需要运行在DLI的独享队列上，因此要与Kafka集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。 Flink跨源开发场景中直接配置跨源认证信息存在密码泄露的风险，优先推荐您使用DLI提供的跨源认证。 跨源认证简介及操作方法请参考跨源认证简介。
-
示例（适用于Kafka集群未开启SASL_SSL场景） 该示例是从Kafka的一个topic中读取数据，并使用Kafka结果表将数据写入到kafka的另一个topic中。 参考增强型跨源连接，根据Kafka所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根据Kafka的地址测试队列连通性。若能连通，则表示跨源已经绑定成功，否则表示未成功。 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业脚本，提交运行作业。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE kafkaSource ( 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 ) WITH ( 'connector' = 'kafka', 'topic' = 'KafkaTopic', '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 string, 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集群，向Kafka的source topic中插入如下测试数据： {"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"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.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"} 连接Kafka集群，在Kafka的sink topic读取数据，参考如下： {"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"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.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"}
-
语法格式 1 2 3 4 5 6 7 8 9 10 11 create table kafkaSink( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'kafka', 'topic' = '', 'properties.bootstrap.servers' = '', 'format' = '' );
-
参数说明 表1 参数说明 参数 是否必选 默认值 数据类型 说明 connector 是 无 String connector类型，需配置为'redis'。 host 是 无 String redis连接地址。 port 否 6379 Integer redis连接端口。 password 否 无 String redis认证密码。 namespace 否 无 String redis key的namespace。 例如设置该值为"person"，假设key为"jack"则redis中会是"person:jack"。 delimiter 否 : String redis的key和namespace之间的分隔符。 data-type 否 hash String redis的数据类型，有下列选项，与redis的数据类型相对应： hash list set sorted-set string data-type取值约束详见data-type取值约束说明。 schema-syntax 否 fields String redis的schema语义，包含以下值： fields：适用于所有数据类型。fields类型是指可以设置多个字段，写入时会取每个字段的值。 fields-scores：适用于sorted set数据类型，表示对每个字段都设置一个字段作为其独立的score。 array：适用于list、set、sorted set数据类型 array-scores：适用于sorted set数据类型 map：适用于hash、sorted set数据类型。 schema-syntax取值约束详见schema-syntax取值约束说明。 deploy-mode 否 standalone String redis集群的部署模式，支持standalone、master-replica、cluster，默认standalone。 该值可参考redis集群的实例类型介绍。 retry-count 否 5 Integer 连接redis集群的尝试次数。 connection-timeout-millis 否 10000 Integer 尝试连接redis集群时的最大超时时间。 commands-timeout-millis 否 2000 Integer 等待操作完成响应的最大时间。 rebalancing-timeout-millis 否 15000 Integer redis集群失败时的休眠时间。 default-score 否 0 Double 当data-type设置为“sorted-set”数据类型的默认score。 ignore-retraction 否 false Boolean 是否忽略retract消息。 skip-null-values 否 true Boolean 是否跳过null。若为false，则设置为字符串"null"。 pwd_auth_name 否 无 String DLI侧创建的Password类型的跨源认证名称。 使用跨源认证则无需在作业中配置账号和密码。 key-ttl-mode 否 no-ttl String key-ttl-mode是开启Redis sink TTL的功能参数，key-ttl-mode的限制为：no-ttl、expire-msec、expire-at-date、expire-at-timestamp。 no-ttl：不设置过期时间。 expire-msec：设置key多久过期，参数为long类型字符串，单位为毫秒。 expire-at-date：设置key到某个时间点过期，参数为UTC时间。 expire-at-timestamp：设置key到某个时间点过期，参数为时间戳。 key-ttl 否 无 String key-ttl是key-ttl-mode的补充参数，有以下几种参数值： 当key-ttl-mode取值为no-ttl时，不需要配置此参数。 当key-ttl-mode取值为expire-msec时，需要配置为可以解析成Long型的字符串。例如5000，表示5000ms后key过期。 当key-ttl-mode取值为expire-at-date时，需要配置为Date类型字符串，例如2011-12-03T10:15:30，表示到期时间为北京时间2011-12-03 18:15:30。 当key-ttl-mode取值为expire-at-timestamp时，需要配置为timestamp类型字符串，单位为毫秒。例如1679385600000，表示到期时间为2023-03-21 16:00:00。
-
参数说明 表1 参数说明 参数 是否必选 默认值 数据类型 说明 connector 是 无 String connector类型，需配置为'redis'。 host 是 无 String redis连接地址。 port 否 6379 Integer redis连接端口。 password 否 无 String redis认证密码。 namespace 否 无 String redis key的namespace。 例如设置该值为"person"，假设key为"jack"则redis中会是"person:jack"。 delimiter 否 : String redis的key和namespace之间的分隔符。 data-type 否 hash String redis的数据类型，有下列选项，与redis的数据类型相对应： hash list set sorted-set string data-type取值约束详见data-type取值约束说明。 schema-syntax 否 fields String redis的schema语义，包含以下值： fields：适用于所有数据类型。fields类型是指可以设置多个字段，写入时会取每个字段的值。 fields-scores：适用于sorted set数据类型，表示对每个字段都设置一个字段作为其独立的score。 array：适用于list、set、sorted set数据类型 array-scores：适用于sorted set数据类型 map：适用于hash、sorted set数据类型。 schema-syntax取值约束详见schema-syntax取值约束说明。 deploy-mode 否 standalone String redis集群的部署模式，支持standalone、master-replica、cluster，默认standalone。 该值可参考redis集群的实例类型介绍。 retry-count 否 5 Integer 连接redis集群的尝试次数。 connection-timeout-millis 否 10000 Integer 尝试连接redis集群时的最大超时时间。 commands-timeout-millis 否 2000 Integer 等待操作完成响应的最大时间。 rebalancing-timeout-millis 否 15000 Integer redis集群失败时的休眠时间。 default-score 否 0 Double 当data-type设置为“sorted-set”数据类型的默认score。 ignore-retraction 否 false Boolean 是否忽略retract消息。 skip-null-values 否 true Boolean 是否跳过null。若为false，则设置为字符串"null"。 pwd_auth_name 否 无 String DLI侧创建的Password类型的跨源认证名称。 使用跨源认证则无需在作业中配置账号和密码。 key-ttl-mode 否 no-ttl String key-ttl-mode是开启Redis sink TTL的功能参数，key-ttl-mode的限制为：no-ttl、expire-msec、expire-at-date、expire-at-timestamp。 no-ttl：不设置过期时间。 expire-msec：设置key多久过期，参数为long类型字符串，单位为毫秒。 expire-at-date：设置key到某个时间点过期，参数为UTC时间。 expire-at-timestamp：设置key到某个时间点过期，参数为时间戳。 key-ttl 否 无 String key-ttl是key-ttl-mode的补充参数，有以下几种参数值： 当key-ttl-mode取值为no-ttl时，不需要配置此参数。 当key-ttl-mode取值为expire-msec时，需要配置为可以解析成Long型的字符串。例如5000，表示5000ms后key过期。 当key-ttl-mode取值为expire-at-date时，需要配置为Date类型字符串，例如2011-12-03T10:15:30，表示到期时间为北京时间2011-12-03 18:15:30。 当key-ttl-mode取值为expire-at-timestamp时，需要配置为timestamp类型字符串，单位为毫秒。例如1679385600000，表示到期时间为2023-03-21 16:00:00。
-
示例 该示例是从kafka数据源中读取数据，并以insert模式写入DWS结果表中，其具体步骤如下： 参考增强型跨源连接，在DLI上根据DWS和Kafka所在的虚拟私有云和子网分别创建相应的增强型跨源连接，并绑定所要使用的Flink弹性资源池。 设置DWS和Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性分别根据DWS和Kafka的地址测试队列连通性。若能连通，则表示跨源已经绑定成功，否则表示未成功。 连接DWS数据库，在DWS中创建相应的表，表名为dws_order，SQL语句参考如下： create table public.dws_order( order_id VARCHAR, order_channel VARCHAR, order_time VARCHAR, pay_amount FLOAT8, real_pay FLOAT8, pay_time VARCHAR, user_id VARCHAR, user_name VARCHAR, area_id VARCHAR); 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。该作业脚本将Kafka作业数据源，将DWS作为结果表。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE kafkaSource ( 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 ) WITH ( 'connector' = 'kafka', 'topic' = 'KafkaTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort', 'properties.group.id' = 'GroupId', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ); CREATE TABLE dwsSink ( 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 ) WITH ( 'connector' = 'gaussdb', 'url' = 'jdbc:postgresql://DWSAddress:DWSPort/DWSdbName', 'table-name' = 'dws_order', 'driver' = 'org.postgresql.Driver', 'username' = 'DWSUserName', 'password' = 'DWSPassword', 'write.mode' = 'insert' ); insert into dwsSink select * from kafkaSource; 连接Kafka集群，向Kafka中输入以下测试数据。 {"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"} 从DWS中使用如下SQL语句查看数据结果。 select * from dws_order 数据结果参考如下： 202103241000000001 webShop 2021-03-24 10:00:00 100.0 100.0 2021-03-24 10:02:03 0001 Alice 330106
-
参数说明 表1 参数说明 参数 是否必选 默认值 类型 说明 connector 是 无 String 指定要使用的连接器，这里是'gaussdb' url 是 无 String jdbc连接地址 。 使用gsjdbc4驱动连接时，格式为：jdbc:postgresql://${ip}:${port}/${dbName} 。 使用gsjdbc200驱动连接时，格式为：jdbc:gaussdb://${ip}:${port}/${dbName}。 table-name 是 无 String 操作的表名。如果该DWS表在某schema下，则格式为：'schema\".\"具体表名'，具体可以参考常见问题说明。 driver 否 org.postgresql.Driver String jdbc连接驱动，默认为: org.postgresql.Driver。 使用gsjdbc4驱动连接时，加载的数据库驱动类为：org.postgresql.Driver。 使用gsjdbc200驱动连接时，加载的数据库驱动类为：com.huawei.gauss200.jdbc.Driver。 username 否 无 String DWS数据库认证用户名，需要和'password'一起配置 password 否 无 String DWS数据库认证密码，需要和'username'一起配置 write.mode 否 无 String 数据写入模式，支持: copy, insert以及upsert三种。默认值为upsert。 该参数与'primary key'配合使用。 未配置'primary key'时，支持copy及insert两种模式追加写入。 配置'primary key'，支持copy、upsert以及insert三种模式更新写入。 注意：由于dws不支持更新分布列，因而配置的更新主键必须包含dws表中定义的所有分布列。 sink.buffer-flush.max-rows 否 100 Integer 每次写入请求缓存的最大行数。 它能提升写入数据的性能，但是也可能增加延迟。 设置为 "0" 关闭此选项。 sink.buffer-flush.interval 否 1s Duration 刷新缓存的间隔，在这段时间内以异步线程刷新数据。 它能提升写入数据库的性能，但是也可能增加延迟。 设置为 "0" 关闭此选项。 注意："sink.buffer-flush.max-size" 和 "sink.buffer-flush.max-rows" 同时设置为 "0"，并设置刷新缓存的间隔，则以完整的异步处理方式刷新缓存。 格式为：{length value}{time unit label}，如123ms, 321s，支持的时间单位包括: d,h,min,s,ms等，默认为ms。 sink.max-retries 否 3 Integer 写入最大重试次数。 write.escape-string-value 否 false Boolean 是否对string类型值进行转义。该参数仅用于write.mode为copy模式下。 pwd_auth_name 否 无 String DLI侧创建的Password类型的跨源认证名称。 使用跨源认证则无需在作业中配置置账号和密码。
-
功能描述 DLI将Flink作业的输出数据输出到数据仓库服务（DWS）中。DWS数据库内核兼容PostgreSQL，PostgreSQL数据库可存储更加复杂类型的数据，支持空间信息服务、多版本并发控制（MVCC）、高并发，适用场景包括位置应用、金融保险、互联网电商等。 数据仓库服务（Data Warehouse Service，简称DWS）是一种基于基础架构和平台的在线数据处理数据库，为用户提供海量数据挖掘和分析服务。DWS的更多信息，请参见《数据仓库服务管理指南》。
-
前提条件 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 请务必确保您的账户下已在数据仓库服务（DWS）里创建了DWS集群。如何创建DWS集群，请参考《数据仓库服务管理指南》中“创建集群”章节。 请确保已创建DWS数据库表。 该场景作业需要运行在DLI的独享队列上，因此要与DWS集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。 Flink跨源开发场景中直接配置跨源认证信息存在密码泄露的风险，优先推荐您使用DLI提供的跨源认证。 跨源认证简介及操作方法请参考跨源认证简介。
-
语法格式 DWS结果表中不允许指定所有属性为PRIMARY KEY。 1 2 3 4 5 6 7 8 9 10 11 12 13 create table dwsSink ( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'gaussdb', 'url' = '', 'table-name' = '', 'driver' = '', 'username' = '', 'password' = '' );
-
注意事项 若需要使用upsert模式，则必须在DWS结果表和该结果表连接的DWS表都定义主键。 若DWS在不同的schema中存在相同名称的表，则在flink opensource sql中需要指定相应的schema。 提交Flink作业前，建议勾选“保存作业日志”参数，在OBS桶选项中选择日志保存的位置，方便后续作业提交失败或运行异常时，查看日志并分析问题原因。 使用gsjdbc4驱动连接时，加载的数据库驱动类为：org.postgresql.Driver。该驱动为默认，创建表时可以不填该驱动参数。 例如，使用gsjdbc4驱动连接、upsert模式写入数据到DWS中。 1 2 3 4 5 6 7 8 9 10 11 12 13 create table dwsSink( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT ) with ( 'connector' = 'gaussdb', 'url' = 'jdbc:postgresql://DwsAddress:DwsPort/DwsDatabase', 'table-name' = 'car_info', 'username' = 'DwsUserName', 'password' = 'DwsPasswrod', 'write.mode' = 'upsert' ); 使用gsjdbc200驱动连接时，加载的数据库驱动类为：com.huawei.gauss200.jdbc.Driver。 当DWS表test在名为ads_game_sdk_base的schema下时，可以参考如下样例创建DWS结果表。 create table dwsSink( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT ) with ( 'connector' = 'gaussdb', 'table-name' = 'ads_game_sdk_base\".\"test', 'driver' = 'com.huawei.gauss200.jdbc.Driver', 'url' = 'jdbc:gaussdb://DwsAddress:DwsPort/DwsDatabase', 'username' = 'DwsUserName', 'password' = 'DwsPasswrod', 'write.mode' = 'upsert' );
-
参数说明 表1 参数说明 参数 是否必选 说明 connector.type 是 connector类型，对于redis，需配置为'redis'。 connector.host 是 redis连接地址。 connector.port 是 redis连接端口。 connector.password 否 redis认证密码。 connector.deploy-mode 否 redis部署模式，支持standalone/cluster，默认standalone connector.table-name 否 table存储模式下必配，redis中存储表名。在table存储模式下，数据将以hash类型存储到redis，其中key为：${table-name}:${ext-key}，field名为列名。 说明： table存储模式：将connector.table-name、connector.key-column作为redis的key。redis的hash类型，每个key对应一个hashmap，hashmap的hashkey为源表的字段名，hashvalue为源表的字段值。 connector.key-column 否 table存储模式下可配置，将该字段值作为redis中的ext-key，未配置时，ext-key为生成的uuid connector.write-schema 否 table存储模式下可配置，是否将当前schema写入到redis，默认为false connector.data-type 否 数据存储类型，用户自定义存储模式必配。支持：string, list, hash, set类型。其中string/list以及sets中schema字段数必须为2，hash字段数必须为3 connector.ignore-retraction 否 是否忽略retraction消息，默认为false
-
语法格式 1 2 3 4 5 6 7 8 9 10 11 12 13 create table dwsSink ( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector.type' = 'redis', 'connector.host' = '', 'connector.port' = '', 'connector.password' = '', 'connector.table-name' = '', 'connector.key-column' = '' );
-
示例 配置“connector.table-name”参数时的table存储模式示例。 table模式采用hash类型存储数据，与基本hash类型将表的三个字段分别作为key、hash_key、hash_value不同，table模式下的key值可以通过“connector.table-name”和“connector.key-column”两个参数设置，将表中的所有字段名作为hash_key，字段值作为hash_value写入到hash中。 create table redisSink( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT ) with ( 'connector.type' = 'redis', 'connector.host' = 'xx.xx.xx.xx', 'connector.port' = '6379', 'connector.password' = 'xx', 'connector.table-name'='car_info', 'connector.key-column'='car_id' ); insert into redisSink (car_id,car_owner,car_brand,car_speed) VALUES ("A1234","OwnA","A1234",30); 以下示例演示“connector.data-type”为string, list, hash, set类型时的建表语句。 “connector.data-type”为string类型。 表为2列：第一列为key，第二列为value。 create table redisSink( attr1 STRING, attr2 STRING ) with ( 'connector.type' = 'redis', 'connector.host' = 'xx.xx.xx.xx', 'connector.port' = '6379', 'connector.password' = 'xx', 'connector.data-type' = 'string' ); insert into redisSink (attr1,attr2) VALUES ("car_id","A1234"); “connector.data-type”为list类型。 表为2列：第一列为key，第二列为value。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 create table redisSink( attr1 STRING, attr2 STRING ) with ( 'connector.type' = 'redis', 'connector.host' = 'xx.xx.xx.xx', 'connector.port' = '6379', 'connector.password' = 'xx', 'connector.data-type' = 'list' ); insert into redisSink (attr1,attr2) VALUES ("car_id","A1234"); “connector.data-type”为set类型。 表为2列：第一列为key，第二列为value。 create table redisSink( attr1 STRING, attr2 STRING ) with ( 'connector.type' = 'redis', 'connector.host' = 'xx.xx.xx.xx', 'connector.port' = '6379', 'connector.password' = 'xx', 'connector.data-type' = 'set' ); insert into redisSink (attr1,attr2) VALUES ("car_id","A1234"); “connector.data-type”为hash类型。 表为3列：第一列为key，第二列为hash_key，第三列为hash_value。 create table redisSink( attr1 STRING, attr2 STRING, attr3 STRING ) with ( 'connector.type' = 'redis', 'connector.host' = 'xx.xx.xx.xx', 'connector.port' = '6379', 'connector.password' = 'xx', 'connector.data-type' = 'hash' ); insert into redisSink (attr1,attr2,attr3) VALUES ("car_info","car_id","A1234");
更多精彩内容
CDN加速
GaussDB
文字转换成语音
免费的服务器
如何创建网站
域名网站购买
私有云桌面
云主机哪个好
域名怎么备案
手机云电脑
SSL证书申请
云点播服务器
免费OCR是什么
电脑云桌面
域名备案怎么弄
语音转文字
文字图片识别
云桌面是什么
网址安全检测
网站建设搭建
国外CDN加速
SSL免费证书申请
短信批量发送
图片OCR识别
云数据库MySQL
个人域名购买
录音转文字
扫描图片识别文字
OCR图片识别
行驶证识别
虚拟电话号码
电话呼叫中心软件
怎么制作一个网站
Email注册网站
华为VNC
图像文字识别
企业网站制作
个人网站搭建
华为云计算
免费租用云托管
云桌面云服务器
ocr文字识别免费版
HTTPS证书申请
图片文字识别转换
国外域名注册商
使用免费虚拟主机
云电脑主机多少钱
鲲鹏云手机
短信验证码平台
OCR图片文字识别
SSL证书是什么
申请企业邮箱步骤
免费的企业用邮箱
云免流搭建教程
域名价格
