华为云用户手册

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 创建HBase源表的列簇必须定义为ROW类型，字段名对应列簇名（column family），嵌套的字段名对应列限定符名（column qualifier）。 用户只需在表结构中声明查询中使用的的列簇和列限定符。除了ROW类型的列，剩下的原子数据类型字段（比如，STRING, BIGINT）将被识别为HBase的rowkey，一张表中只能声明一个rowkey。rowkey字段的名字可以是任意的，如果是保留关键字，需要用反引号进行转义。

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

示例 示例1：该示例是从DMS Kafka数据源中读取数据，并写入到Print结果表中。 参考增强型跨源连接，根据Kafka所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 设置Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根据Kafka的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业脚本，提交运行作业。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE upsertKafkaSource ( 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, PRIMARY KEY (order_id) NOT ENFORCED) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'KafkaTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkAddress2:KafkaPort', 'key.format' = 'csv', 'value.format' = 'json');CREATE TABLE printSink ( 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, PRIMARY KEY (order_id) NOT ENFORCED) WITH ( 'connector' = 'print');INSERT INTO printSink SELECT * FROM upsertKafkaSource; 向Kafka中的指定topic中插入如下数据（注意：kafka插入数据时请指定key）。 {"order_id":"202303251202020001", "order_channel":"miniAppShop", "order_time":"2023-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2023-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"}{"order_id":"202303251505050001", "order_channel":"appshop", "order_time":"2023-03-25 15:05:05", "pay_amount":"500.00", "real_pay":"400.00", "pay_time":"2023-03-25 15:10:00", "user_id":"0003", "user_name":"Cindy", "area_id":"330108"}{"order_id":"202303251202020001", "order_channel":"miniAppShop", "order_time":"2023-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2023-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330111"} 查看taskmanager的out文件，数据结果参考如下： +I(202303251202020001,miniAppShop,2023-03-2512:02:02,60.0,60.0,2023-03-2512:03:00,0002,Bob,330110)+I(202303251505050001,appshop,2023-03-25 15:05:05,500.0,400.0,2023-03-2515:10:00,0003,Cindy,330108)-U(202303251202020001,miniAppShop,2023-03-2512:02:02,60.0,60.0,2023-03-2512:03:00,0002,Bob,330110)+U(202303251202020001,miniAppShop,2023-03-2512:02:02,60.0,60.0,2023-03-2512:03:00,0002,Bob,330111)

参数说明 表2 参数说明 参数 是否必选 默认参数 数据类型 说明 connector 是 无 String connector类型，对于upsert kafka连接器，需配置为'upsert-kafka'。 topic 是 无 String Kafka topic名。 properties.bootstrap.servers 是 无 String Kafka brokers地址，以逗号分隔。 key.format 是 无 String 用于对Kafka消息中key部分序列化和反序列化的格式。key字段由PRIMARY KEY语法指定。支持的格式如下： csv json avro 请参考Format页面以获取更多详细信息和格式参数。 key.fields-prefix 否 无 String 为键格式的所有字段定义自定义前缀，以避免与值格式的字段发生名称冲突。 默认情况下，前缀为空。如果定义了自定义前缀，则表架构和'key.fields'都将使用前缀名称。在构造密钥格式的数据类型时，将删除前缀，并在密钥格式中使用无前缀的名称。请注意，此选项要求'value.fields-include' 必须设置为'EXCEPT_KEY'。 value.format 是 无 String 用于对 Kafka消息中 value 部分序列化和反序列化的格式。支持的格式： csv json avro 请参考Format页面以获取更多详细信息和格式参数。 value.fields-include 是 ALL String 控制哪些字段应该出现在值中。取值范围如下： ALL：消息的value部分将包含schema的所有字段，包括定义中键的字段。 EXCEPT_KEY：记录的value部分包含schema的所有内容，定义为主键的字段除外。 properties.* 否 无 String 该选项可以传递任意的Kafka参数。 “properties.”后的后缀名必须匹配定义在 kafka参数文档中的参数名。 Flink会自动移除选项名中的 "properties." 前缀，并将转换后的键名以及值传入KafkaClient。 例如：您可以通过 'properties.allow.auto.create.topics' = 'false' 来禁止自动创建 topic。 但是'key.deserializer' 和 'value.deserializer' 是不允许通过该方式传递参数，因为Flink会重写这些参数的值。 sink.parallelism 否 无 Integer 定义upsert-kafka sink 算子的并行度。默认情况下，由框架确定并行度，与上游连接算子的并行度保持一致。 sink.buffer-flush.max-rows 否 0 Integer 缓存刷新前，最多能缓存的记录条数。 当sink收到很多同key上的更新时，缓存将保留同 key 的最后一条记录，因此sink缓存能帮助减少发往Kafka topic的数据量，以及避免发送潜在的tombstone消息。可以通过设置为'0'来禁用它。 默认情况下，该选项是未开启的。如果要开启 sink 缓存，需要同时设置'sink.buffer-flush.max-rows'和'sink.buffer-flush.interval'两个选项为大于零的值。 sink.buffer-flush.interval 否 0 Duration 缓存刷新的间隔时间，超过该时间后异步线程将刷新缓存数据。单位可以为毫秒（ms）、秒（s）、分钟（min）或小时（h）。例如'sink.buffer-flush.interval'='10 ms'。 默认情况下，该选项是未开启的。如果要开启 sink 缓存，需要同时设置'sink.buffer-flush.max-rows'和'sink.buffer-flush.interval'两个选项为大于零的值。

语法格式 1 2 3 4 5 6 7 8 9101112 create table kafkaTable( 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' = '');

功能描述 Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。Upsert Kafka 连接器支持以upsert方式从Kafka topic中读取数据并将数据写入Kafka topic。表类型支持源表和结果表。 作为source，upsert-kafka 连接器生产changelog流，其中每条数据记录代表一个更新或删除事件。 数据记录中的value被解释为同一key的最后一个value的UPDATE，如果有这个key（如果不存在相应的key，则该更新被视为INSERT）。用表来类比，changelog 流中的数据记录被解释为UPSERT，也称为INSERT/UPDATE，因为任何具有相同key的现有行都被覆盖。另外，value为空的消息将会被视作为DELETE消息。 作为sink，upsert-kafka连接器可以消费changelog流。它会将INSERT/UPDATE_AFTER数据作为正常的Kafka消息写入，并将DELETE数据以value为空的Kafka消息写入（表示对应 key 的消息被删除）。Flink将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。 表1 支持类别 类别 详情 支持表类型 源表、结果表

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 Upsert Kafka 始终以upsert方式工作，并且需要在DDL中定义主键。在具有相同主键值的消息按序存储在同一个分区的前提下，在 changlog source 定义主键意味着 在物化后的 changelog 上主键具有唯一性。定义的主键将决定哪些字段出现在Kafka消息的key中。 由于该连接器以 upsert 的模式工作，该连接器作为 source 读入时，可以确保具有相同主键值下仅最后一条消息会生效。 数据类型的使用，请参考Format章节。

功能描述 您可以将Hive表用作时态表，通过时态连接来关联Hive表。有关时态连接的详细信息，请参阅 temporal join。 Flink支持processing-time temporal join Hive Table，processing-time temporal join始终会加入最新版本的时态表。Flink支持分区表和 Hive非分区表的临时连接，对于分区表，Flink 支持自动跟踪Hive表的最新分区。详情可参考：Apache Flink Hive Read & Write

注意事项 Flink目前不支持与Hive表进行基于事件时间event-time的时间关联。 Temporal Join The Latest Partition 特性，仅在 Flink STREAMING 模式下支持。 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 数据类型的使用，请参考Format章节。 Hive 方言支持的 DDL 语句，Flink 1.15 当前仅支持使用Hive语法创建OBS表和使用hive语法的 DLI Lakehouse表。 使用Hive语法创建OBS表 defalut方言： with 属性中需要设置hive.is-external为true。 使用hive 方言：建表语句需要使用EXTERNAL关键字。 使用hive语法的DLI Lakehouse表 使用hive 方言：表属性中需要添加'is_lakehouse'='true'。 创建Flink OpenSource SQL作业时，在作业编辑界面配置开启checkpoint功能。

CAST示例一：将amount值转换成整型 将amount值转换成整型。 insert into temp select cast(amount as INT) from source_stream; 表2 CAST类型转换函数示例 示例 说明 示例 cast(v1 as string) 将v1转换为字符串类型，v1可以是数值类型，TIMESTAMP/DATE/TIME。 表T1: | content (INT) || ------------- || 5 | 语句： SELECT cast(content as varchar)FROM T1; 结果： "5" cast (v1 as int) 将v1转换为int, v1可以是数值类型或字符类。 表T1: | content (STRING) || ------------- || "5" | 语句： SELECT cast(content as int)FROM T1; 结果： 5 cast(v1 as timestamp) 将v1转换为timestamp类型，v1可以是字符串或DATE/TIME。 表T1: | content (STRING) || ------------- || "2018-01-01 00:00:01" | 语句： SELECT cast(content as timestamp)FROM T1; 结果： 1514736001000 cast(v1 as date) 将v1转换为date类型， v1可以是字符串或者TIMESTAMP。 表T1: | content (TIMESTAMP) || ------------- || 1514736001000 | 语句： SELECT cast(content as date)FROM T1; 结果： "2018-01-01" Flink作业不支持使用CAST将“BIGINT”转换为“TIMESTAMP”，可以使用to_timestamp进行转换。

INNER/LEFT/RIGHT/FULL OUTER INNER/LEFT/RIGHT/FULL OUTER 这几种窗口关联的语法非常相似，我们在这里只举一个 FULL OUTER JOIN 的例子。 当执行窗口关联时，所有具有相同 key 和相同滚动窗口的数据会被关联在一起。这里给出一个基于 TUMBLE Window TVF 的窗口连接的例子。 在下面的例子中，通过将 join 的时间区域限定为固定的 5 分钟，数据集被分成两个不同的时间窗口：[12:00,12:05) 和 [12:05,12:10)。L2 和 R2 不能 join 在一起是因为它们不在一个窗口中。 语法格式 SELECT ...FROM L [LEFT|RIGHT|FULL OUTER] JOIN R -- L and R are relations applied windowing TVFON L.window_start = R.window_start AND L.window_end = R.window_end AND ...

示例 从Kafka源表中读取数据，将DWS表作为维表，并将二者生成的宽表信息写入Kafka结果表中，其具体步骤如下： 参考增强型跨源连接，在DLI上根据DWS和Kafka所在的虚拟私有云和子网分别创建相应的增强型跨源连接，并绑定所要使用的Flink弹性资源池。 设置DWS和Kafka的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性分别根据DWS和Kafka的地址测试队列连通性。若能连通，则表示跨源已经绑定成功，否则表示未成功。 连接DWS数据库实例，在DWS中创建相应的表，作为维表，表名为area_info，SQL语句如下： create table public.area_info( area_id VARCHAR, area_province_name VARCHAR, area_city_name VARCHAR, area_county_name VARCHAR, area_street_name VARCHAR, region_name VARCHAR); 连接DWS数据库实例，向DWS维表area_info中插入测试数据，其语句如下： insert into area_info (area_id, area_province_name, area_city_name, area_county_name, area_street_name, region_name) values ('330102', 'a1', 'b1', 'c1', 'd1', 'e1'), ('330106', 'a1', 'b1', 'c2', 'd2', 'e1'), ('330108', 'a1', 'b1', 'c3', 'd3', 'e1'), ('330110', 'a1', 'b1', 'c4', 'd4', 'e1'); 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。该作业脚本将Kafka作为数据源，DWS作为维表，数据输出到Kafka结果表中。 注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。 CREATE TABLE orders ( 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, proctime as Proctime()) WITH ( 'connector' = 'kafka', 'topic' = 'KafkaSourceTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort', 'properties.group.id' = 'dws-order', 'scan.startup.mode' = 'latest-offset', 'format' = 'json');--创建地址维表create table area_info ( area_id string, area_province_name string, area_city_name string, area_county_name string, area_street_name string, region_name string ) WITH ( 'connector' = 'gaussdb', 'driver' = 'org.postgresql.Driver', 'url' = 'jdbc:postgresql://DwsAddress:DwsPort/DwsDbName', 'table-name' = 'area_info', 'username' = 'DwsUserName', 'password' = 'DwsPassword', 'lookup.cache.max-rows' = '10000', 'lookup.cache.ttl' = '2h');--根据地址维表生成详细的包含地址的订单信息宽表create table order_detail( 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, area_province_name string, area_city_name string, area_county_name string, area_street_name string, region_name string) with ( 'connector' = 'kafka', 'topic' = 'KafkaSinkTopic', 'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort', 'format' = 'json');insert into order_detail select orders.order_id, orders.order_channel, orders.order_time, orders.pay_amount, orders.real_pay, orders.pay_time, orders.user_id, orders.user_name, area.area_id, area.area_province_name, area.area_city_name, area.area_county_name, area.area_street_name, area.region_name from orders left join area_info for system_time as of orders.proctime as area on orders.area_id = area.area_id; 连接Kafka集群，向kafka中source topic中插入如下测试数据： {"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}{"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"}{"order_id":"202103251505050001", "order_channel":"qqShop", "order_time":"2021-03-25 15:05:05", "pay_amount":"500.00", "real_pay":"400.00", "pay_time":"2021-03-25 15:10:00", "user_id":"0003", "user_name":"Cindy", "area_id":"330108"} 连接Kafka集群，读取kafka中sink topic中数据，结果参考如下： {"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","area_province_name":"a1","area_city_name":"b1","area_county_name":"c2","area_street_name":"d2","region_name":"e1"}{"order_id":"202103251202020001","order_channel":"miniAppShop","order_time":"2021-03-25 12:02:02","pay_amount":60.0,"real_pay":60.0,"pay_time":"2021-03-25 12:03:00","user_id":"0002","user_name":"Bob","area_id":"330110","area_province_name":"a1","area_city_name":"b1","area_county_name":"c4","area_street_name":"d4","region_name":"e1"}{"order_id":"202103251505050001","order_channel":"qqShop","order_time":"2021-03-25 15:05:05","pay_amount":500.0,"real_pay":400.0,"pay_time":"2021-03-25 15:10:00","user_id":"0003","user_name":"Cindy","area_id":"330108","area_province_name":"a1","area_city_name":"b1","area_county_name":"c3","area_street_name":"d3","region_name":"e1"}

常见问题 Q：若Flink作业日志中有如下报错信息，应该怎么解决？ java.io.IOException: unable to open JDBC writer...Caused by: org.postgresql.util.PSQLException: The connection attempt failed....Caused by: java.net.SocketTimeoutException: connect timed out A：应考虑是跨源没有绑定，或者跨源没有绑定成功。 参考增强型跨源连接章节，重新配置跨源。参考DLI跨源连接DWS失败进行问题排查。 Q：如果该DWS表在某schema下，则应该如何配置？ A：如下示例是使用schema为dbuser2下的表area_info： --创建地址维表create table area_info ( area_id string, area_province_name string, area_city_name string, area_county_name string, area_street_name string, region_name string ) WITH ( 'connector' = 'gaussdb', 'driver' = 'org.postgresql.Driver', 'url' = 'jdbc:postgresql://DwsAddress:DwsPort/DwsDbname', 'table-name' = 'dbuser2.area_info', 'username' = 'DwsUserName', 'password' = 'DwsPassword', 'lookup.cache.max-rows' = '10000', 'lookup.cache.ttl' = '2h');

参数说明 表1 参数说明 参数 是否必选 默认值 数据类型 说明 connector 是 无 String connector类型，需配置为'gaussdb'。 url 是 无 String jdbc连接地址。 使用gsjdbc4驱动连接时，格式为：jdbc:postgresql://${ip}:${port}/${dbName} 。 使用gsjdbc200驱动连接时，格式为：jdbc:gaussdb://${ip}:${port}/${dbName}。 table-name 是 无 String 读取数据库中的数据所在的表名。 driver 否 无 String jdbc连接驱动，默认为: org.postgresql.Driver。 使用gsjdbc4驱动连接时，加载的数据库驱动类为：org.postgresql.Driver。 使用gsjdbc200驱动连接时，加载的数据库驱动类为：com.huawei.gauss200.jdbc.Driver。 username 否 无 String 数据库认证用户名，需要和'password'一起配置。 password 否 无 String 数据库认证密码，需要和'username'一起配置。 scan.partition.column 否 无 String 用于对输入进行分区的列名。 与scan.partition.lower-bound、scan.partition.upper-bound、scan.partition.num必须同时存在或者同时不存在。 scan.partition.lower-bound 否 无 Integer 第一个分区的最小值。 与scan.partition.column、scan.partition.upper-bound、scan.partition.num必须同时存在或者同时不存在。 scan.partition.upper-bound 否 无 Integer 最后一个分区的最大值。 与scan.partition.column、scan.partition.lower-bound、scan.partition.num必须同时存在或者同时不存在。 scan.partition.num 否 无 Integer 分区的个数。 与scan.partition.column、scan.partition.upper-bound、scan.partition.upper-bound必须同时存在或者同时不存在。 scan.fetch-size 否 0 Integer 每次从数据库拉取数据的行数。默认值为0，表示不限制。 scan.auto-commit 否 true Boolean 设置自动提交标志。 它决定每一个statement是否以事务的方式自动提交。 lookup.cache.max-rows 否 无 Integer 维表配置，缓存的最大行数，超过该值时，最先添加的数据将被标记为过期。 默认表示不使用该配置。 lookup.cache.ttl 否 无 Duration 维表配置，缓存超时时间，超过该时间的数据会被剔除。格式为：{length value}{time unit label}，如123ms, 321s，支持的时间单位包括: d,h,min,s,ms等，默认为ms。 默认表示不使用该配置。 lookup.max-retries 否 3 Integer 维表配置，数据拉取最大重试次数。

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 with参数中字段只能使用单引号，不能使用双引号。

语法格式 1 2 3 4 5 6 7 8 91011 create table dwsSource ( attr_name attr_type (',' attr_name attr_type)* )with ( 'connector' = 'gaussdb', 'url' = '', 'table-name' = '', 'username' = '', 'password' = '');

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

示例1 使用datagen随机生成数据写入obs的bucketName桶下的fileName目录中。文件生成时间与checkpoint有关，达到30min或128MB时，生成新文件。 create table orders( name string, num INT) with ( 'connector' = 'datagen', 'rows-per-second' = '100', 'fields.name.kind' = 'random', 'fields.name.length' = '5' );CREATE TABLE sink_table ( name string, num INT) WITH ( 'connector' = 'filesystem', 'path' = 'obs://bucketName/fileName', 'format' = 'csv', 'sink.rolling-policy.file-size'='128m', 'sink.rolling-policy.rollover-interval'='30 min');INSERT into sink_table SELECT * from orders;

示例2 使用datagen随机生成数据写入obs的bucketName桶下的fileName目录中。文件生成时间与checkpoint有关，达到checkpoint间隔或达到100MB时，生成新文件。 create table orders( name string, num INT) with ( 'connector' = 'datagen', 'rows-per-second' = '100', 'fields.name.kind' = 'random', 'fields.name.length' = '5' );CREATE TABLE sink_table ( name string, num INT) WITH ( 'connector' = 'filesystem', 'path' = 'obs://bucketName/fileName', 'format' = 'parquet', 'sink.rolling-policy.file-size'='128m', 'sink.rolling-policy.rollover-interval'='30 min', 'auto-compaction'='true', 'compaction.file-size'='100m');INSERT into sink_table SELECT * from orders;

功能描述 窗口 Top-N 是特殊的 Top-N，它返回每个分区键的每个窗口的N个最小或最大值。 与普通Top-N不同，窗口Top-N只在窗口最后返回汇总的Top-N数据，不会产生中间结果。窗口 Top-N 会在窗口结束后清除不需要的中间状态。 窗口 Top-N 适用于用户不需要每条数据都更新Top-N结果的场景，相对普通Top-N来说性能更好。通常，窗口 Top-N 直接用于窗口表值函数（Windowing TVFs）窗口 Top-N 可以用于基于窗口表值函数（Windowing TVFs）的操作之上，比如窗口聚合，窗口Top-N和 窗口关联。 窗口 Top-N 的语法和普通的 Top-N 相同。 除此之外，窗口 Top-N 需要 PARTITION BY 子句包含窗口表值函数或窗口聚合产生的 window_start 和 window_end。 否则优化器无法翻译。 更多介绍和使用请参考开源社区文档：窗口Top-N。

参数说明 表1 参数说明 参数 是否必选 默认值 类型 说明 connector 是 无 String 固定位filesystem。 path 是 无 String OBS路径。 format 是 无 String 文件格式。 支持csv、parquet格式。 sink.rolling-policy.file-size 否 128MB MemorySize 单个part文件最大大小，超过该数值会滚动产生新文件。 说明： RollingPolicy 定义了何时关闭给定的In-progress Part文件，并将其转换为Pending状态，然后再转换为Finished状态。 Finished状态的文件，可供查看并且可以保证数据的有效性，在出现故障时不会恢复。 在STREAMING模式下，滚动策略结合Checkpoint间隔（到下一个Checkpoint成功时，文件的Pending状态才转换为Finished状态）共同控制Part文件对下游readers是否可见以及这些文件的大小和数量。 sink.rolling-policy.rollover-interval 否 30 min Duration 单个Part文件处于打开状态的最长时间，超过该时间会滚动产生新文件（默认值30分钟，以避免产生大量小文件）。检查频率是通过sink.rolling-policy.check-interval参数控制的。 说明： 该参数数字与单位之间必须要有空格。 支持的时间单位包括: d,h,min,s,ms等。 对于bulk格式的文件(parquet、orc、avro)，checkpoint的时间间隔也会控制单个part文件打开的最长时间。 sink.rolling-policy.check-interval 否 1 min Duration 基于时间的滚动策略的检查间隔。 该属性控制了基于sink.rolling-policy.rollover-interval属性检查文件是否该被滚动的检查频率。 auto-compaction 否 false Boolean 在流式 sink 中是否开启自动合并功能。数据首先会被写入临时文件。当checkpoint完成后，该checkpoint产生的临时文件会被合并。 compaction.file-size 否 `sink.rolling-policy.file-size`的大小 MemorySize 合并目标文件大小，默认值为滚动文件大小。 说明： 只有在同个checkpoint内的文件会被合并，因此最终文件的数量至少等于checkpoint的数量。 如果合并时间较长，可能会引起反压，延长checkpoint所需时间。 开启该功能后，checkpoint时会产生最终文件，并打开新的文件接收下个checkpoint产生的数据。

功能描述 FileSystem sink用于将数据输出到分布式文件系统HDFS或者 对象存储服务 OBS等文件系统。适用于数据转储、大数据分析、备份或活跃归档、深度或冷归档等场景。 考虑到输入流可以是无界的，每个桶中的数据被组织成有限大小的Part文件。完全可以配置为基于时间的方式往桶中写入数据，比如可以设置每个小时的数据写入一个新桶中。即桶中将包含一个小时间隔内接收到的记录。 桶目录中的数据被拆分成多个Part文件。对于相应的接收数据的桶的Sink的每个Subtask，每个桶将至少包含一个Part文件。将根据配置的滚动策略来创建其他Part文件。对于Row Formats默认的策略是根据Part文件大小进行滚动，需要指定文件打开状态最长时间的超时以及文件关闭后的非活动状态的超时时间。对于Bulk Formats在每次创建Checkpoint时进行滚动，并且用户也可以添加基于大小或者时间等的其他条件。更多信息参考文件系统 SQL 连接器 在STREAMING模式下使用FileSink需要开启Checkpoint功能。Part文件只在Checkpoint成功时生成。如果没有开启Checkpoint功能，文件将永远停留在in-progress或者pending的状态，并且下游系统将不能安全读取该文件数据。 sink end算子的接受记录数为checkpoint的个数，非实际的发送数据，实际发送数据量请参考streaming-writer或StreamingFileWriter算子的记录数。

Join Temporal Table Function 功能描述 注意事项 目前仅支持在 Temporal Tables 上的 inner join 示例 假如Rates是一个 Temporal Table Function， join 可以使用 SQL 进行如下的表达: SELECT o_amount, r_rateFROM Orders, LATERAL TABLE (Rates(o_proctime))WHERE r_currency = o_currency;

语法格式 1 2 3 4 5 6 7 8 91011 CREATE TABLE sink_table ( name string, num INT, p_day string, p_hour string) partitioned by (p_day, p_hour) WITH ( 'connector' = 'filesystem', 'path' = 'obs://*** ', 'format' = 'parquet', 'auto-compaction' = 'true');

逻辑函数 表1 逻辑函数 SQL函数 返回类型 描述 boolean1 OR boolean2 BOOLEAN 如果 boolean1 为 TRUE 或 boolean2 为 TRUE 返回 TRUE。支持三值逻辑。 例如 true || Null(BOOLEAN) 返回 TRUE。 boolean1 AND boolean2 BOOLEAN 如果 boolean1 和 boolean2 都为 TRUE 返回 TRUE。支持三值逻辑。 例如 true && Null(BOOLEAN) 返回 UNKNOWN。 NOT boolean BOOLEAN 如果布尔值为 FALSE 返回 TRUE；如果布尔值为 TRUE 返回 FALSE；如果布尔值为 UNKNOWN 返回 UNKNOWN。 boolean IS FALSE BOOLEAN 如果布尔值为 FALSE 返回 TRUE；如果 boolean 为 TRUE 或 UNKNOWN 返回 FALSE。 boolean IS NOT FALSE BOOLEAN 如果 boolean 为 TRUE 或 UNKNOWN 返回 TRUE；如果 boolean 为 FALSE 返回 FALSE。 boolean IS TRUE BOOLEAN 如果 boolean 为 TRUE 返回 TRUE；如果 boolean 为 FALSE 或 UNKNOWN 返回 FALSE。 boolean IS NOT TRUE BOOLEAN 如果 boolean 为 FALSE 或 UNKNOWN 返回 TRUE；如果布尔值为 TRUE 返回 FALSE。 boolean IS UNKNOWN BOOLEAN 如果布尔值为 UNKNOWN 返回 TRUE；如果 boolean 为 TRUE 或 FALSE 返回 FALSE。 boolean IS NOT UNKNOWN BOOLEAN 如果 boolean 为 TRUE 或 FALSE 返回 TRUE；如果布尔值为 UNKNOWN 返回 FALSE。 父主题： 内置函数

Join表函数(UDTF) 功能描述 将表与表函数的结果进行 join 操作。左表（outer）中的每一行将会与调用表函数所产生的所有结果中相关联行进行 join 。 注意事项 针对横向表的左外部连接当前仅支持文本常量 TRUE 作为谓词。 示例 如果表函数返回了空结果，左表（outer）的行将会被删除 SELECT users, tagFROM Orders, LATERAL TABLE(unnest_udtf(tags)) t AS tag; 如果表函数返回了空结果，将会保留相对应的外部行并用空值填充 SELECT users, tagFROM Orders LEFT JOIN LATERAL TABLE(unnest_udtf(tags)) t AS tag ON TRUE;

IN 语法格式 123 SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } FROM tableExpression WHERE column_name IN (value (, value)* ) | query 语法说明 IN操作符允许在where子句中规定多个值。如果表达式在给定的表子查询中存在，则返回 true 。 注意事项 子查询表必须由单个列构成，且该列的数据类型需与表达式保持一致。 示例 输出Orders中NewProducts中product的user和amount信息。 12345 insert into temp SELECT user, amountFROM OrdersWHERE product IN ( SELECT product FROM NewProducts);

Union/Union ALL/Intersect/Except 语法格式 1 query UNION [ ALL ] | Intersect | Except query 语法说明 UNION返回多个查询结果的并集。 Intersect返回多个查询结果的交集。 Except返回多个查询结果的差集。 注意事项 集合运算是以一定条件将表首尾相接，所以其中每一个SELECT语句返回的列数必须相同，列的类型一定要相同，列名不一定要相同。 UNION默认是去重的，UNION ALL是不去重的。 示例 输出Orders1和Orders2的并集，不包含重复记录。 12 insert into temp SELECT * FROM Orders1 UNION SELECT * FROM Orders2;

语法说明 SELECT order_id, order_time, amount, SUM(amount) OVER w AS sum_amount, AVG(amount) OVER w AS avg_amountFROM OrdersWINDOW w AS ( PARTITION BY product ORDER BY order_time RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW) ORDER BY：OVER 窗口需要数据是有序的。因为表没有固定的排序，所以 ORDER BY 子句是强制的。对于流式查询，Flink 目前只支持 OVER 窗口定义在升序（asc）的时间属性上。其他的排序不支持。 PARTITION BY：OVER 窗口可以定义在一个分区表上。PARTITION BY 子句代表着每行数据只在其所属的数据分区进行聚合。 范围（RANGE）定义：范围（RANGE）定义指定了聚合中包含了多少行数据。范围通过 BETWEEN 子句定义上下边界，其内的所有行都会聚合。Flink 只支持 CURRENT ROW 作为上边界。有两种方法可以定义范围：ROWS 间隔 和 RANGE 间隔： RANGE 间隔 RANGE 间隔是定义在排序列值上的，在 Flink 里，排序列总是一个时间属性。下面的 RANG 间隔定义了聚合会在比当前行的时间属性小 30 分钟的所有行上进行。 RANGE BETWEEN INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW ROW 间隔 ROWS 间隔基于计数。它定义了聚合操作包含的精确行数。下面的 ROWS 间隔定义了当前行 + 之前的 10 行（也就是11行）都会被聚合。 ROWS BETWEEN 10 PRECEDING AND CURRENT ROW WINDOW：WINDOW 子句可用于在 SELECT 子句之外定义 OVER 窗口。它让查询可读性更好，也允许多个聚合共用一个窗口定义。

示例 查询为每个订单计算前一个小时之内接收到的同一产品所有订单的总金额。 1234567 SELECT order_id, order_time, amount, SUM(amount) OVER ( PARTITION BY product ORDER BY order_time RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW ) AS one_hour_prod_amount_sumFROM Orders