2.5 创建中间流
功能描述
中间流用来简化sql逻辑，若sql逻辑比较复杂，可以写多个sql语句，用中间流进行串 接。中间流仅为逻辑意义上的流，不会产生数据存储。
语法格式
CREATE TEMP STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* )
示例
create temp stream a2(attr1 int, attr2 string);
2.6 创建维表
2.6.1 创建 Redis 表
创建Redis表用于与输入流连接。
DCS的详细信息请参见《分布式缓存服务用户指南》。
流表JOIN语法请参见流表JOIN。
语法格式
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="")?);
关键字
表2-25 关键字说明
参数 是否必
选
说明
type 是 输出通道类型，dcs_redis表示输出到分布式缓存服务的 Redis存储系统中。
cluster_add
ress 是 Redis实例连接地址。
password 否 Redis实例连接密码，当设置为免密访问时，省略该配置 项。
value_type 是 指定数据类型。支持的数据类型包括：string, list, hash, set, zset。
key_colum
n 是 指定代表Redis key属性的列名。
hash_key_c
olumn 否 当value_type设置为hash时，需要指定本字段作为第二级 key属性的列名。
cache_max
_num 否 表示最大缓存的查询结果数，默认值为32768。
cache_time 否 表示数据库查询结果在内存中缓存的最大时间。单位为毫 秒，默认值为10000，当值为0时表示不缓存。
注意事项
● 不支持Redis集群。
● 请务必确保您的账户下已在分布式缓存服务（DCS）里创建了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"
);
2.6.2 创建 RDS 表
创建RDS/DWS表用于与输入流连接。
RDS的更多信息，请参见《关系型数据库用户指南》。
流表JOIN语法请参见流表JOIN。
前提条件
● 请务必确保您的账户下已在关系型数据库（RDS）里创建了PostgreSQL或MySQL 类型的RDS实例。
如何创建RDS实例，请参见《关系型数据库快速入门》中“购买实例”章节。
● 该场景作业需要运行在DLI的独享队列上，因此要与RDS实例建立增强型跨源连 接，且用户可以根据实际所需设置相应安全组规则。
如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章 节。
如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。
语法格式
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 = ""
);
关键字
表2-26 关键字说明
参数 是否
必选
说明
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驱动连接，
x_num 否 表示最大缓存的查询结果数，默认值为32768。
cache_tim
e 否 表示数据库查询结果在内存中缓存的最大时间。单位为毫秒，
默认值为10000，当值为0时表示不缓存。
示例
RDS表用于与输入流连接。
CREATE SOURCE STREAM car_infos ( car_id STRING,
car_owner STRING, car_brand STRING, car_price INT ) WITH (
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驱动进行连接。