DCS输出流

功能描述

DLI将Flink作业的输出数据输出到分布式缓存服务（DCS）的Redis中。Redis是一种支持Key-Value等多种数据结构的存储系统。可用于缓存、事件发布或订阅、高速队列等场景，提供字符串、哈希、列表、队列、集合结构直接存取，基于内存，可持久化。有关Redis的详细信息，请访问Redis官方网站https://redis.io/。

分布式缓存服务（DCS）为DLI提供兼容Redis的即开即用、安全可靠、弹性扩容、便捷管理的在线分布式缓存能力，满足用户高并发及快速数据访问的业务诉求。

DCS的更多信息，请参见《分布式缓存服务用户指南》。

前提条件

• 请务必确保您的账户下已在分布式缓存服务（DCS）里创建了Redis类型的缓存实例。

  如何创建Redis类型的缓存实例，请参考《分布式缓存服务用户指南》中“申请Redis缓存实例”章节。

• 该场景作业需要运行在DLI的独享队列上，因此要与DCS集群建立跨源连接，且用户可以根据实际所需设置相应安全组规则。

  如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。

  如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

• 用户通过VPC对等访问DCS实例时，除了满足VPC对等网跨VPC访问的约束之外，还存在如下约束：
  • 当创建DCS实例时使用了172.16.0.0/12~24网段时，DLI队列不能在192.168.1.0/24、192.168.2.0/24、192.168.3.0/24网段。
  • 当创建DCS实例时使用了192.168.0.0/16~24网段时，DLI队列不能在172.31.1.0/24、172.31.2.0/24、172.31.3.0/24网段。
  • 当创建DCS实例时使用了10.0.0.0/8~24网段时，DLI队列不能在172.31.1.0/24、172.31.2.0/24、172.31.3.0/24网段。

语法格式

1
2
3
4
5
6
7
8

CREATE SINK STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* )
  WITH (
    type = "dcs_redis",
    region = "",
    cluster_address = "",
    password = "",
    value_type= "",key_value= ""
  );

关键字

注意事项

• 当配置项支持参数化时，表示将记录中的一列或者多列作为该配置项的一部分。例如当配置项设置为car_${car_brand}时，如果一条记录的car_brand列值为BMW，则该配置项在该条记录下为car_BMW。
• 字符":", ",", ";", "$", "{", "}"已被征用为特殊分隔符，暂时没有提供转义功能，禁止在key和value中作为普通字符使用，否则会影响解析，导致程序异常。

示例

 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}}"
  );