华为云用户手册

语法说明 ROW_NUMBER(): 从第一行开始，依次为每一行分配一个唯一且连续的号码。 PARTITION BY col1[, col2...]: 指定分区的列，例如去重的键。 ORDER BY time_attr [asc|desc]: 指定排序的列。所制定的列必须为时间属性。目前仅支持proctime。升序（ ASC ）排列指只保留第一行，而降序排列（ DESC ）则指保留最后一行。 WHERE rownum = 1: Flink 需要 rownum = 1 以确定该查询是否为去重查询。

示例 根据order_id对数据进行去重，其中proctime为事件时间属性列 SELECT order_id, user, product, number FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY proctime ASC) as row_num FROM Orders) WHERE row_num = 1;

使用示例 UDTF支持CROSS JOIN和LEFT JOIN，在使用UDTF时需要带上 LATERAL 和TABLE 两个关键字。 CROSS JOIN：对于左表的每一行数据，假设UDTF不产生输出，则这一行不进行输出。 LEFT JOIN：对于左表的每一行数据，假设UDTF不产生输出，这一行仍会输出，UDTF相关字段用null填充。 CREATE FUNCTION udtf_test AS 'com.huaweicompany.udf.TableFunction';-- CROSS JOIN INSERT INTO sink_stream select subValue, length FROM source_stream, LATERAL TABLE(udtf_test(attr, ',')) as T(subValue, length);-- LEFT JOIN INSERT INTO sink_stream select subValue, length FROM source_stream LEFT JOIN LATERAL TABLE(udtf_test(attr, ',')) as T(subValue, length) ON TRUE;

操作场景 类型推导包含了验证输入值、派生参数和返回值数据类型。从逻辑角度看，Planner需要知道数据类型、精度和小数位数；从 JVM 角度来看，Planner 在调用自定义函数时需要知道如何将内部数据结构表示为JVM对象。 Flink 自定义函数实现了自动的类型推导提取，通过反射从函数的类及其求值方法中派生数据类型。然而以反射方式提取数据类型并不总是成功的，比如UDTF中常见的Row类型。 由于 Flink 1.11 起引入了新的自定义函数注册接口，使用了新的自定义函数类型推断机制，因此原先1.10 重载 getResultType 声明返回字段类型的方式将不再可用。继续使用会抛出如下异常： Caused by: org.apache.flink.table.api.ValidationException: Cannot extract a data type from a pure 'org.apache.flink.types.Row' class. Please use annotations to define field names and field types. 目前 Flink 1.15 可以通过使用DataTypeHint 和FunctionHint 注解相关参数、类或方法来支持提取过程。

函数说明 表1 集合函数说明 集合函数 函数说明 CARDINALITY(array) 返回数组中元素的数量。 array '[' INT ']' 返回数组中 INT 位置的元素。索引从 1 开始。 ELEMENT(array) 返回数组的唯一元素（其基数应为 1）；如果数组为空，则返回 NULL。如果数组有多个元素，则抛出异常。 CARDINALITY(map) 返回 map 中的 entries 数量。 map ‘[’ value ‘]’ 返回 map 中指定 key 对应的值。

聚合函数 聚合函数将所有的行作为输入，并返回单个聚合值作为结果。 表1 聚合函数 函数 描述 COUNT([ ALL ] expression | DISTINCT expression1 [, expression2]*) 默认情况下或使用关键字 ALL，返回不为 NULL 的表达式的输入行数。使用 DISTINCT 则对所有值去重后计算。 COUNT(*) | COUNT(1) 返回输入行数。 AVG([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回所有输入行中表达式的平均值（算术平均值）。使用 DISTINCT 则对所有值去重后计算。 SUM([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回所有输入行的表达式总和。使用 DISTINCT 则对所有值去重后计算。 MAX([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回所有输入行中表达式的最大值。使用 DISTINCT 则对所有值去重后计算。 MIN([ ALL | DISTINCT ] expression ) 默认情况下或使用关键字 ALL，返回所有输入行中表达式的最小值。使用 DISTINCT 则对所有值去重后计算。 STDDEV_POP([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回所有输入行中表达式的总体标准偏差。使用 DISTINCT 则对所有值去重后计算。 STDDEV_SAMP([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回所有输入行中表达式的样本标准偏差。使用 DISTINCT 则对所有值去重后计算。 VAR_POP([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回所有输入行中表达式的总体方差（总体标准差的平方）。使用 DISTINCT 则对所有值去重后计算。 VAR_SAMP([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回所有输入行中表达式的样本方差（样本标准差的平方）。使用 DISTINCT 则对所有值去重后计算。 COLLECT([ ALL | DISTINCT ] expression) 默认情况下或使用关键字 ALL，返回跨所有输入行的多组表达式。NULL 值将被忽略。使用 DISTINCT 则对所有值去重后计算 VARIANCE([ ALL | DISTINCT ] expression) VAR_SAMP() 的同义方法。 RANK() 返回值在一组值中的排名。结果是 1 加上分区顺序中当前行之前或等于当前行的行数。排名在序列中不一定连续。 DENSE_RANK() 返回值在一组值中的排名。结果是一加先前分配的等级值。与函数 rank 不同，dense_rank 不会在排名序列中产生间隙。 ROW_NUMBER() 在窗口分区内根据 rows 的排序为每一行分配一个唯一的序列号，从一开始。ROW_NUMBER 和 RANK 相似。ROW_NUMBER 按 顺序对所有行进行编号（例如 1，2，3，4，5）。RANK 为等值 row 提供相同的序列值（例如 1，2，2，4，5）。 LEAD(expression [, offset] [, default]) 返回窗口中当前行之后第 offset 行处的表达式值。offset 的默认值为 1，default 的默认值为 NULL。 LAG(expression [, offset] [, default]) 返回窗口中当前行之前第 offset 行处的表达式值。offset 的默认值为 1，default 的默认值为 NULL。 FIRST_VALUE(expression) 返回一组有序值中的第一个值。 LAST_VALUE(expression) 返回一组有序值中的最后一个值。 LISTAGG(expression [, separator]) 连接字符串表达式的值并在它们之间放置分隔符值。字符串末尾不添加分隔符时则分隔符的默认值为“,”。 父主题： 内置函数

示例 查询为每个订单计算前一个小时之内接收到的同一产品所有订单的总金额。 1 2 3 4 5 6 7 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_sum FROM Orders

注意事项 当前仅支持 PRECEDING (无界或有界) 到 CURRENT ROW 范围内的窗口、FOLLOWING 所描述的区间并未支持。 ORDER BY 必须指定于单个的时间属性。 可以在一个 SELECT 子句中定义多个 OVER 窗口聚合。然而，对于流式查询，由于目前的限制，所有聚合的 OVER 窗口必须是相同的。 OVER 窗口需要数据是有序的。因为表没有固定的排序，所以 ORDER BY 子句是强制的。对于流式查询，Flink 目前只支持 OVER 窗口定义在升序（asc）的时间属性上。其他的排序不支持。

语法说明 SELECT order_id, order_time, amount, SUM(amount) OVER w AS sum_amount, AVG(amount) OVER w AS avg_amount FROM Orders WINDOW 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 窗口。它让查询可读性更好，也允许多个聚合共用一个窗口定义。

逻辑函数 表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。 父主题： 内置函数

数据类型映射 目前，Parquet 格式类型映射与 Apache Hive 兼容，但与 Apache Spark 有所不同： Timestamp：不论精度，映射 timestamp 类型至 int96。 Decimal：根据精度，映射 decimal 类型至固定长度字节的数组。 下表列举了 Flink 中的数据类型与 JSON 中的数据类型的映射关系。 注意：复合数据类型暂只支持写不支持读（Array、Map 与 Row）。 表2 数据类型映射 Flink数据类型 Parquet类型 Parquet逻辑类型 CHAR / VARCHAR / STRING BINARY UTF8 BOOLEAN BOOLEAN - BINARY / VARBINARY BINARY - DECIMAL FIXED_LEN_BYTE_ARRAY DECIMAL TINYINT INT32 INT_8 SMALLINT INT32 INT_16 INT INT32 - BIGINT INT64 - FLOAT FLOAT - DOUBLE DOUBLE - DATE INT32 DATE TIME INT32 TIME_MILLIS TIMESTAMP INT96 - ARRAY - LIST MAP - MAP ROW - STRUCT

语法定义 INSERT INTO table_name [PARTITION part_spec] query part_spec: (part_col_name1=val1 [, part_col_name2=val2, ...]) query: values | { select | selectWithoutFrom | query UNION [ ALL ] query | query EXCEPT query | query INTERSECT query } [ ORDER BY orderItem [, orderItem ]* ] [ LIMIT { count | ALL } ] [ OFFSET start { ROW | ROWS } ] [ FETCH { FIRST | NEXT } [ count ] { ROW | ROWS } ONLY] orderItem: expression [ ASC | DESC ] select: SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } FROM tableExpression [ WHERE booleanExpression ] [ GROUP BY { groupItem [, groupItem ]* } ] [ HAVING booleanExpression ] [ WINDOW windowName AS windowSpec [, windowName AS windowSpec ]* ] selectWithoutFrom: SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } projectItem: expression [ [ AS ] columnAlias ] | tableAlias . * tableExpression: tableReference [, tableReference ]* | tableExpression [ NATURAL ] [ LEFT | RIGHT | FULL ] JOIN tableExpression [ joinCondition ] joinCondition: ON booleanExpression | USING '(' column [, column ]* ')' tableReference: tablePrimary [ matchRecognize ] [ [ AS ] alias [ '(' columnAlias [, columnAlias ]* ')' ] ] tablePrimary: [ TABLE ] [ [ catalogName . ] schemaName . ] tableName | LATERAL TABLE '(' functionName '(' expression [, expression ]* ')' ')' | UNNEST '(' expression ')' values: VALUES expression [, expression ]* groupItem: expression | '(' ')' | '(' expression [, expression ]* ')' | CUBE '(' expression [, expression ]* ')' | ROLLUP '(' expression [, expression ]* ')' | GROUPING SETS '(' groupItem [, groupItem ]* ')' windowRef: windowName | windowSpec windowSpec: [ windowName ] '(' [ ORDER BY orderItem [, orderItem ]* ] [ PARTITION BY expression [, expression ]* ] [ RANGE numericOrIntervalExpression {PRECEDING} | ROWS numericExpression {PRECEDING} ] ')' matchRecognize: MATCH_RECOGNIZE '(' [ PARTITION BY expression [, expression ]* ] [ ORDER BY orderItem [, orderItem ]* ] [ MEASURES measureColumn [, measureColumn ]* ] [ ONE ROW PER MATCH ] [ AFTER MATCH ( SKIP TO NEXT ROW | SKIP PAST LAST ROW | SKIP TO FIRST variable | SKIP TO LAST variable | SKIP TO variable ) ] PATTERN '(' pattern ')' [ WITHIN intervalLiteral ] DEFINE variable AS condition [, variable AS condition ]* ')' measureColumn: expression AS alias pattern: patternTerm [ '|' patternTerm ]* patternTerm: patternFactor [ patternFactor ]* patternFactor: variable [ patternQuantifier ] patternQuantifier: '*' | '*?' | '+' | '+?' | '?' | '??' | '{' { [ minRepeat ], [ maxRepeat ] } '}' ['?'] | '{' repeat '}'

约束限制 Flink SQL 对于标识符（表、属性、函数名）有类似于 Java 的词法约定: 不管是否引用标识符，都保留标识符的大小写。 且标识符需区分大小写。 与 Java 不一样的地方在于，通过反引号，可以允许标识符带有非字母的字符（如："SELECT a AS `my field` FROM t"）。 字符串文本常量需要被单引号包起来（如 SELECT 'Hello World' ）。两个单引号表示转义（如 SELECT 'It''s me.'）。字符串文本常量支持 Unicode 字符，如需明确使用 Unicode 编码，请使用以下语法： 使用反斜杠（\）作为转义字符（默认）：SELECT U&'\263A' 使用自定义的转义字符： SELECT U&'#263A' UESCAPE '#'

参数说明 表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产生的数据。

语法格式 1 2 3 4 5 6 7 8 9 10 11 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 使用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;

功能描述 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算子的记录数。

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 TVF ON L.window_start = R.window_start AND L.window_end = R.window_end AND ...

算术函数 表1 算术函数 运算符 描述 + numeric 返回 numeric。 - numeric 返回 numeric 的相反数。 numeric1 + numeric2 返回 numeric1 加 numeric2 numeric1 - numeric2 返回 numeric1 减 numeric2。 numeric1 * numberic2 返回 numeric1 乘以 numeric2。 numeric1 / numeric2 返回 numeric1 除以 numeric2。 numeric1 % numeric2 返回 numeric1 除以 numeric2 的余数（模数）。仅当 numeric1 为负时，结果才为负。 POWER(numeric1, numeric2) 返回 numeric1 的 numeric2 次方。 ABS(numeric) 返回 numeric 的绝对值。 SQRT(numeric) 返回 numeric 的平方根。 LN(numeric) 返回 numeric 的自然对数（以 e 为底）。 LOG10(numeric) 返回以 10 为底的 numeric 的对数。 LOG2(numeric) 返回以 2 为底的 numeric 的对数。 LOG(numeric2) LOG(numeric1, numeric2) 当用一个参数调用时，返回 numeric2 的自然对数。当使用两个参数调用时，此函数返回 numeric2 以 numeric1 为底的对数。numeric2 必须大于 0，numeric1 必须大于 1。 EXP(numeric) 返回 e 的 numeric 次幂。 CEIL(numeric) CEILING(numeric) 向上取整，并返回大于或等于 numeric 的最小整数。 FLOOR(numeric) 向下取整，并返回小于或等于 numeric 的最大整数。 SIN(numeric) 返回 numeric 的正弦值。 SINH(numeric) 返回 numeric 的双曲正弦值。返回类型为 DOUBLE。 COS(numeric) 返回 numeric 的正切值。 TAN(numeric) 计算给定A的正切值。 TANH(numeric) 返回 numeric 的双曲正切值。返回类型为 DOUBLE。 COT(numeric) 返回 numeric 的余切值。 ASIN(numeric) 返回 numeric 的反正弦值。 ACOS(numeric) 返回 numeric 的反余弦值。 ATAN(numeric) 返回 numeric 的反正切值。 ATAN2(numeric1, numeric2) 返回坐标 (numeric1, numeric2) 的反正切。 COSH(numeric) 返回 numeric 的双曲余弦值。返回值类型为 DOUBLE。 DEGREES(numeric) 返回弧度 numeric 的度数表示。 RADIANS(numeric) 返回度数 numeric 的弧度表示。 SIGN(numeric) 返回 numeric 的符号。 ROUND(numeric, INT) 返回 numeric 四舍五入保留 INT 小数位的值。 PI() 返回无比接近 pi 的值。 E() 返回无比接近 e 的值。 RAND() 返回 [0.0, 1.0) 范围内的伪随机双精度值。 RAND(INT) 返回范围为 [0.0, 1.0) 的伪随机双精度值，初始种子为 INT。 如果两个 RAND 函数具有相同的初始种子，它们将返回相同的数字序列。 RAND_INTEGER(INT) 返回 [0, INT) 范围内的伪随机整数。 RAND_INTEGER(INT1, INT2) 返回范围为 [0, INT2) 的伪随机整数，初始种子为 INT1。 如果两个 RAND_INTGER 函数具有相同的初始种子和边界，它们将返回相同的数字序列。 UUID() 根据 RFC 4122 类型 4（伪随机生成）UUID，返回 UUID（通用唯一标识符）字符串。 例如“3d3c68f7-f608-473f-b60c-b0c44ad4cc4e”，UUID 是使用加密强的伪随机数生成器生成的。 BIN(INT) 以二进制格式返回 INTEGER 的字符串表示形式。如果 INTEGER 为 NULL，则返回 NULL。 例如 4.bin() 返回“100”，12.bin() 返回“1100”。 HEX(numeric) HEX(string) 以十六进制格式返回整数 numeric 值或 STRING 的字符串表示形式。如果参数为 NULL，则返回 NULL。 例如数字 20 返回“14”，数字 100 返回“64”，字符串“hello,world” 返回“68656C6C6F2C776F726C64”。 TRUNCATE(numeric1, integer2) 返回截取 integer2 位小数的数字。如果 numeric1 或 integer2 为 NULL，则返回 NULL。 如果 integer2 为 0，则结果没有小数点或小数部分。integer2 可以为负数，使值的小数点左边的 integer2 位变为零。 此函数也可以传入只有一个 numeric1 参数且不设置 Integer2 以使用。 如果未设置 Integer2 则 Integer2 为 0。 例如 42.324.truncate(2) 为 42.32，42.324.truncate() 为 42.0。 父主题： 内置函数

功能描述 窗口去重是一种特殊的去重，它根据指定的多个列来删除重复的行，保留每个窗口和分区键的第一个或最后一个数据。 对于流式查询，与普通去重不同，窗口去重只在窗口的最后返回结果数据，不会产生中间结果。它会清除不需要的中间状态。 因此，窗口去重查询在用户不需要更新结果时，性能较好。通常，窗口去重直接用于窗口表值函数上。另外，它可以用于基于窗口表值函数的操作。比如窗口聚合，窗口TopN和窗口关联。 窗口Top-N的语法和普通的Top-N相同。 除此之外，窗口去重需要 PARTITION BY 子句包含表的 window_start 和 window_end 列。 否则优化器无法翻译。 Flink 使用 ROW_NUMBER() 移除重复数据，就像窗口TopN一样。理论上，窗口是一种特殊的窗口 Top-N：N是1并且是根据处理时间或事件时间排序的。 更多介绍和使用请参考开源社区文档：窗口去重。

镜像迁移操作步骤 准备镜像仓库访问权限文件：auth.json。 新建一个auth.json文件，并按照格式修改，如果是匿名访问仓库，则不需要填写用户名、密码等信息。将文件放置在image-migrator所在目录下。 示例： { "quay.io/coreos": { }, "swr.cn-north-4.myhuaweicloud.com": { "username": "cn-north-4@RVHVMX******", "password": "cab4ceab4a1545***************", "insecure": true } } 详细的参数说明请参见“auth.json”文件。 准备镜像列表文件：images.json。 通过kubectl连接源集群。具体方法可参考使用kubectl连接集群。 执行镜像迁移config子命令，生成images.json文件。 您可以参考image-migrator config使用方法中的方法和示例，不指定命名空间，或者指定一个、多个命名空间来获取源集群应用中使用的镜像。 根据需求调整images.json文件内容，但要遵循“images.json”文件中所讲的八项要求。 镜像迁移。 您可以执行默认的./image-migrator-linux-amd64命令进行镜像迁移，也可以根据需要设置image-migrator的参数。 例如以下命令： ./image-migrator-linux-amd64 --workers=5 --auth=./auth.json --images=./images.json --namespace=test --registry=swr.cn-north-4.myhuaweicloud.com --retries=2 示例： $ ./image-migrator-linux-amd64 Start to generate images tasks, please wait ... Start to handle images tasks, please wait ... Images(38) migration finished, 0 images tasks failed, 0 tasks generate failed 结果查看。 上述命令执行完毕后，回显如下类似信息： Images(38) migration finished, 0 images tasks failed, 0 tasks generate failed 表示按照配置，成功将38个镜像迁移到SWR仓库中。

准备工作 在开始迁移之前，请确保您已准备了一台安装了kubectl的服务器，用于连接源集群和目标集群。该服务器需要至少拥有5GB左右的本地磁盘空间和≥8G的内存，以确保迁移工具可以正常运行，并存储相关数据，如源集群的采集数据和目标集群的推荐数据等。 迁移工具支持在Linux（x86、arm）、Windows环境中运行，因此您可以在这些操作系统中任选一种作为服务器的操作系统。 对于Linux操作系统来说，使用image-migrator前，需要运行chmod u+x 工具名 命令（例如chmod u+x image-migrator-linux-amd64），授予可执行权限。 表1 image-migrator工具包获取 image-migrator image-migrator是一个镜像迁移工具，能够自动将基于Docker Registry v2搭建的Docker镜像仓库或第三方云镜像仓库中的镜像迁移到SWR中。 Linux x86：https://ucs-migration.obs.cn-north-4.myhuaweicloud.com/toolkits/image-migrator-linux-amd64 Linux arm：https://ucs-migration.obs.cn-north-4.myhuaweicloud.com/toolkits/image-migrator-linux-arm64 Windows：https://ucs-migration.obs.cn-north-4.myhuaweicloud.com/toolkits/image-migrator-windows-amd64.exe

image-migrator工作原理 图1 image-migrator工作原理 使用image-migrator工具将镜像迁移到SWR时，需要准备两个文件，一个为镜像仓库访问权限文件“auth.json”，该文件中的两个对象分别为源镜像仓库和目标镜像仓库（即Registry）的账号和密码；另一个为镜像列表文件“images.json”，文件内容由多条镜像同步规则组成，每一条规则包括一个源镜像仓库（键）和一个目标镜像仓库（值）。将这两个文件准备好以后，放在image-migrator工具所在目录下，执行一条简单的命令，就可以完成镜像的迁移。关于两个文件的详细介绍如下： “auth.json”文件 “auth.json”为镜像仓库访问权限文件，其中每个对象为一个registry的用户名和密码。通常源镜像仓库需要具有pull以及访问tags权限，目标镜像仓库需要拥有push以及创建仓库权限。如果是匿名访问镜像仓库，则不需要填写用户名、密码等信息。“auth.json”文件的结构如下： { "源镜像仓库地址": { }, "目标镜像仓库地址": { "username": "xxxxxx", "password": "xxxxxx", "insecure": true } } 其中， “源镜像仓库地址”和“目标镜像仓库地址”支持“registry”和“registry/namespace”两种形式，需要跟下述“images.json”中的registry或registry/namespace对应。images中被匹配到的url会使用对应用户名密码进行镜像同步，优先匹配“registry/namespace”的形式。 目标镜像仓库地址如果为“registry”形式，可以从SWR控制台页面获取，具体方法如下：在“总览”页面单击右上角“登录指令”，登录指令末尾的域名即为SWR镜像仓库地址，例如swr.cn-north-4.myhuaweicloud.com。注意每个Region的地址不同，请切换到对应Region获取。如果为“registry/namespace”形式，还要将namespace替换为SWR的组织名称。 username：（可选）用户名，values可以填写具体取值，也可以使用“${env}”或者“$env”类型的字符串引用环境变量。 password：（可选）密码，values可以填写具体取值，也可以使用“${env}”或者“$env”类型的字符串引用环境变量。 insecure：（可选）registry是否为http服务，如果是，insecure为true；默认是false。 目标镜像仓库SWR的用户名形式为：区域项目名称@AK；密码为AK和SK经过加密处理后的登录密钥，详细指导请参考获取长期有效登录指令。 示例： { "quay.io/coreos": { }, "swr.cn-north-4.myhuaweicloud.com": { "username": "cn-north-4@RVHVMX******", "password": "cab4ceab4a1545***************", "insecure": true } } “images.json”文件 该文件本质上是一个待迁移的镜像清单，由多条镜像同步规则组成，每一条规则包括一个源镜像仓库（键）和一个目标镜像仓库（值）。具体的要求如下： 同步的最大单位是仓库（repository），不支持通过一条规则同步整个namespace以及registry。 源仓库和目标仓库的格式与docker pull/push命令使用的镜像url类似（registry/namespace/repository:tag）。 源仓库和目标仓库（如果目标仓库不为空字符串）都至少包含registry/namespace/repository。 源仓库字段不能为空，如果需要将一个源仓库同步到多个目标仓库需要配置多条规则。 目标仓库名可以和源仓库名不同，此时同步功能类似于：docker pull + docker tag + docker push。 当源仓库字段中不包含tag时，表示将该仓库所有tag同步到目标仓库，此时目标仓库不能包含tag。 当源仓库字段中包含tag时，表示只同步源仓库中的一个tag到目标仓库，如果目标仓库中不包含tag，则默认使用源tag。 当目标仓库为空字符串时，会将源镜像同步到默认registry的默认namespace下，并且repository以及tag与源仓库相同，默认registry和默认namespace可以通过命令行参数以及环境变量配置。 示例如下： { "quay.io/coreos/etcd:1.0.0": "swr.cn-north-4.myhuaweicloud.com/test/etcd:1.0.0", "quay.io/coreos/etcd": "swr.cn-north-4.myhuaweicloud.com/test/etcd", "quay.io/coreos/etcd:2.7.3": "swr.cn-north-4.myhuaweicloud.com/test/etcd" } 我们提供了一个自动获取集群中工作负载正在使用的镜像的方法，即image-migrator工具的config子命令，具体用法请参见image-migrator config使用方法。得到images.json文件后，您还可以根据需要进行修改、添加或删除。

image-migrator使用方法 image-migrator工具支持在Linux（x86、arm）和Windows环境中运行，使用方法相似。本文将以Linux（x86）环境为例进行介绍。 若使用Linux（arm）或Windows环境，请将下述命令中的image-migrator-linux-amd64分别替换为image-migrator-linux-arm64或image-migrator-windows-amd64.exe。 在image-migrator工具所在目录下执行./image-migrator-linux-amd64 -h，可以查看image-migrator工具的使用方法。 --auth：指定auth.json的路径，默认在image-migrator所在目录下。 --images：指定images.json的路径，默认在image-migrator所在目录下。 --log：指定image-migrator生成日志的路径，默认是image-migrator当前目录下的image-migrator.log。 --namespace：默认的目标仓库的namespace，也就是说，如果images.json中没有指定目标仓库中的namespace，可以在执行迁移命令时指定。 --registry：默认的目标仓库的registry，也就是说，如果images.json中没有指定目标仓库中的registry，可以在执行迁移命令时指定。 --retries：迁移失败时的重试次数，默认为3。 --workers：镜像搬迁的worker数量（并发数），默认是7。 $ ./image-migrator-linux-amd64 -h A Fast and Flexible docker registry image images tool implement by Go. Usage: image-migrator [flags] Aliases: image-migrator, image-migrator Flags: --auth string auth file path. This flag need to be pair used with --images. (default "./auth.json") -h, --help help for image-migrator --images string images file path. This flag need to be pair used with --auth (default "./images.json") --log string log file path (default "./image-migrator.log") --namespace string default target namespace when target namespace is not given in the images config file, can also be set with DEFAULT_NAMESPACE environment value --registry string default target registry url when target registry is not given in the images config file, can also be set with DEFAULT_REGISTRY environment value -r, --retries int times to retry failed tasks (default 3) -w, --workers int numbers of working goroutines (default 7) $./image-migrator --workers=5 --auth=./auth.json --images=./images.json --namespace=test \ --registry=swr.cn-north-4.myhuaweicloud.com --retries=2 $ ./image-migrator Start to generate images tasks, please wait ... Start to handle images tasks, please wait ... Images(38) migration finished, 0 images tasks failed, 0 tasks generate failed 示例如下： ./image-migrator --workers=5 --auth=./auth.json --images=./images.json --namespace=test --registry=swr.cn-north-4.myhuaweicloud.com --retries=2 该命令表示将“images.json”文件中的镜像迁移至“swr.cn-north-4.myhuaweicloud.com/test”镜像仓库下，迁移失败时可以重试2次，一次可以同时搬迁5个镜像。

image-migrator config使用方法 image-migrator工具的config子命令可用于获取集群应用中使用的镜像，在工具所在目录下生成images.json。执行./image-migrator-linux-amd64 config -h命令可以查看config子命令的使用方法。 -k, --kubeconfig：指定kubectl的kubeConfig位置，默认是$HOME/.kube/config。kubeConfig文件：用于配置对Kubernetes集群的访问，KubeConfig文件中包含访问注册kubernetes集群所需要的认证凭据以及Endpoint（访问地址），详细介绍可参见Kubernetes文档。 -n, --namespaces：指定获取镜像的命名空间，多个命名空间用逗号分隔（如：ns1,ns2,ns3)，默认是""，表示获取所有命名空间的镜像。 -t, --repo：指定目标仓库的地址（registry/namespace）。 $ ./image-migrator-linux-amd64 config -h generate images.json Usage: image-migrator config [flags] Flags: -h, --help help for config -k, --kubeconfig string The kubeconfig of k8s cluster's. Default is the $HOME/.kube/config. (default "/root/.kube/config") -n, --namespaces string Specify a namespace for information collection. If multiple namespaces are specified, separate them with commas (,), such as ns1,ns2. default("") is all namespaces -t, --repo string target repo,such as swr.cn-north-4.myhuaweicloud.com/test 示例如下： 指定一个命名空间 ./image-migrator-linux-amd64 config -n default -t swr.cn-north-4.myhuaweicloud.com/test 指定多个命名空间 ./image-migrator-linux-amd64 config -n default,kube-system -t swr.cn-north-4.myhuaweicloud.com/test 不指定命名空间（表示获取所有命名空间的镜像） ./image-migrator-linux-amd64 config -t swr.cn-north-4.myhuaweicloud.com/test

请求示例 在当前区域创建或者更新Landing Zone。 POST https://{hostname}/v1/landing-zone/setup { "identity_store_email" : "string", "home_region" : "string", "setup_landing_zone_action_type" : "string", "region_configuration_list" : [ { "region" : "string", "region_configuration_status" : "ENABLED" } ], "organization_structure" : [ { "organizational_unit_name" : "string", "organizational_unit_type" : "CORE", "accounts" : [ { "account_name" : "string", "account_id" : "string", "phone" : "18700000000", "account_email" : "string", "account_type" : "LOGGING" } ] } ], "deny_ungoverned_regions" : false, "cloud_trail_type" : true, "kms_key_id" : "", "logging_configuration" : { "logging_bucket" : { "retention_days" : 5475 }, "access_logging_bucket" : { "retention_days" : 5475 } } }

请求参数 表1 请求Body参数 参数 是否必选 参数类型 描述 identity_store_email 是 String 管理员纳管账号创建Identity Center用户所用邮箱。 最小长度：3 最大长度：64 home_region 是 String 主区域。 最小长度：1 最大长度：64 setup_landing_zone_action_type 是 String 设置Landing Zone的类型。包括CREATE、REPAIR以及UPDATE。 最小长度：1 最大长度：256 region_configuration_list 是 Array of regionConfigurationList objects 当前纳管账号纳管的区域。 数组长度：1 - 30 organization_structure 是 Array of OrganizationStructureBaseLine objects 基础环境的纳管账号体系。 数组长度：1 - 2 deny_ungoverned_regions 否 Boolean 是否允许区域拒绝，默认false。 缺省值：false cloud_trail_type 否 Boolean 是否允许设置组织汇聚。 缺省值：true kms_key_id 否 String 加密字段。 logging_configuration 是 logging_configuration object 日志桶设置信息。 表2 regionConfigurationList 参数 是否必选 参数类型 描述 region 是 String 区域名字。 最小长度：1 最大长度：36 region_configuration_status 是 String 纳管账号类型LOGGING,SECURITY。 枚举值： ENABLED DISABLED 表3 OrganizationStructureBaseLine 参数 是否必选 参数类型 描述 organizational_unit_name 是 String 注册OU名称。 最小长度：1 最大长度：64 organizational_unit_type 是 String 可设置的注册OU类型。包括核心注册OU和自定义注册OU。 枚举值： CORE CUSTOM accounts 否 Array of AccountBaseline objects 纳管账号基本信息。 数组长度：0 - 100 表4 AccountBaseline 参数 是否必选 参数类型 描述 account_name 是 String 纳管账号名称。 最小长度：6 最大长度：30 account_id 否 String 纳管账号的唯一标识符（ID）。 最小长度：0 最大长度：64 phone 否 String 手机号码。 最小长度：1 最大长度：32 account_email 否 String 纳管账号邮箱。 最小长度：0 最大长度：64 account_type 是 String 纳管账号类型。类型包括LOGGING，SECURITY和CUSTOM。 枚举值： LOGGING SECURITY CUSTOM 表5 logging_configuration 参数 是否必选 参数类型 描述 logging_bucket 是 LoggingBucketBaseline object 日志基础设置。 access_logging_bucket 是 AccessLoggingBucketBaseline object 访问日志基础设置。 表6 LoggingBucketBaseline 参数 是否必选 参数类型 描述 retention_days 是 Integer 桶保留天数。 最小值：1 最大值：5475 表7 AccessLoggingBucketBaseline 参数 是否必选 参数类型 描述 retention_days 是 Integer 桶保留天数。 最小值：1 最大值：5475

响应参数 状态码： 200 表1 响应Body参数 参数 参数类型 描述 baseline_update_available Boolean Landing Zone基础配置是否可用。 control_update_available Boolean 当前账号下是否有新的控制策略。 landing_zone_update_available Boolean Landing Zone是否可更新。 service_landing_zone_version String Landing Zone当前最新的版本号。 最小长度：1 最大长度：128 user_landing_zone_version String 用户当前的Landing Zone版本。 最小长度：1 最大长度：128

响应示例 状态码： 200 请求成功。 { "baseline_update_available" : true, "control_update_available" : true, "landing_zone_update_available" : true, "service_landing_zone_version" : "string", "user_landing_zone_version" : "string" }