摘要：開發(fā)指南是為簡(jiǎn)化計(jì)算模型，降低用戶使用實(shí)時(shí)計(jì)算的門檻而設(shè)計(jì)的一套符合標(biāo)準(zhǔn)語(yǔ)義的開發(fā)套件。隨后，將為該表生成字段，用于記錄并表示事件時(shí)間。

UFlink SQL 開發(fā)指南

UFlink SQL 是 UCloud 為簡(jiǎn)化計(jì)算模型，降低用戶使用實(shí)時(shí)計(jì)算的門檻而設(shè)計(jì)的一套符合標(biāo)準(zhǔn) SQL 語(yǔ)義的開發(fā)套件。

接下來(lái)，開發(fā)者可以根據(jù)如下內(nèi)容，逐漸熟悉并使用 UFlink SQL 組件所提供的便捷功能。

1 表中受支持的數(shù)據(jù)類型

2 DDL 語(yǔ)句

UFlink 實(shí)時(shí)計(jì)算框架本身不帶有存儲(chǔ)數(shù)據(jù)的功能，因此，創(chuàng)建的表均為外部表，是對(duì)外部數(shù)據(jù)的映射和引用，在程序執(zhí)行時(shí)，數(shù)據(jù)會(huì)動(dòng)態(tài)加載到 UFlink 中。

2.1 創(chuàng)建源表

在 UFlink SQL 中，創(chuàng)建 數(shù)據(jù)源表 遵循如下規(guī)律：

CREATE TABLE table_name(
    columnName dataType
    [ columnName dataType ]*
) WITH (
    type = xxxx
    xxxx = xxxx
);

1. Kafka 數(shù)源表

（1）示例

? 以 Kafka 作為數(shù)據(jù)源表，建表語(yǔ)句示例如下：

CREATE TABLE ut_kafka_source(
    user_id VARCHAR
    pay_cash FLOAT
    pay_time VARCHAR
 )WITH(
    type = kafka11
    bootstrapServers = linux01:9092linux02:9092linux03:9092
    topic = flink-test01
    groupId = group1
    parallelism = 1
 );

上述代碼用于在 UFlink 中創(chuàng)建一張名為 ut_kafka_source 的表，并聲明表中字段信息（屬性名及屬性類型）。

隨后，必須在 WITH 代碼塊中指定數(shù)據(jù)來(lái)源，以及數(shù)據(jù)源所使用的框架的必要參數(shù)。

（2）相關(guān)參數(shù)

? 可配置參數(shù)如下表所示：

2.2 創(chuàng)建結(jié)果表

在 UFlink SQL 中，創(chuàng)建 結(jié)果表 遵循如下規(guī)律：

CREATE TABLE table_name(
    columnName dataType
    [ columnName dataType ]*
) WITH (
    type = xxxx
    xxxx = xxxx
);

1. Kafka 結(jié)果表

（1）示例

? 以 Kafka 作為結(jié)果表，建表語(yǔ)句示例如下：

CREATE TABLE ut_kafka_result(
    uid VARCHAR
    pay_cash FLOAT
    sex VARCHAR
    age VARCHAR
 )WITH(
    type = kafka11
    bootstrapServers = linux01:9092linux02:9092linux03:9092
    topic = flink-test02
    parallelism = 1
 );

上述代碼用于在 UFlink 中創(chuàng)建一張名為 ut_kafka_result 的表，并聲明表中字段信息（屬性名及屬性類型）。

隨后，必須在 WITH 代碼塊中指定數(shù)據(jù)輸出時(shí)的相關(guān)參數(shù)，以及數(shù)據(jù)源所使用的框架的必要參數(shù)。

（2）相關(guān)參數(shù)

? 可配置參數(shù)如下表所示：

2. MySQL 結(jié)果表

（1）示例

? 以 MySQL 作為結(jié)果表，建表語(yǔ)句示例如下：

CREATE TABLE ut_mysql_result(
    uid VARCHAR
    pay_cash FLOAT
    sex VARCHAR
    age VARCHAR
)WITH(
    type = mysql
    url = jdbc:mysql://linux01:3306/db_flink_test?useUnicode=true&characterEncoding=utf-8
    userName = root
    password = 123456
    tableName = ut_mysql_result
    parallelism = 1
);

（2）相關(guān)參數(shù)

? 可配置參數(shù)如下表所示：

3. PostgreSQL 結(jié)果表

（1）示例

? 以 PostgreSQL 作為結(jié)果表，建表語(yǔ)句示例如下：

CREATE TABLE ut_postgre_result(
    uid VARCHAR
    pay_cash FLOAT
    sex VARCHAR
    age VARCHAR
)WITH(
    type = postgre
    url = jdbc:postgresql://10.9.11.122:5432/db_flink_test?useUnicode=true&characterEncoding=utf-8
    userName = root
    password = 123456
    tableName = ut_postgre_result
    parallelism = 1
);

上述代碼展示了如何使用 UFlink SQL 模塊，并通過(guò) PostgreSQL 與 UDW(UCloud 數(shù)據(jù)湖) 產(chǎn)品交互，實(shí)現(xiàn)數(shù)據(jù)輸出到 UDW。

（2）相關(guān)參數(shù)

? 可配置參數(shù)如下表所示：

2.3 創(chuàng)建維表

在 UFlink SQL 中，創(chuàng)建 維度表 遵循如下規(guī)律：

CREATE TABLE table_name(
    columnName dataType
    [ columnName dataType ]*
) WITH (
    type = xxxx
    xxxx = xxxx
    SIDE DIMENSION SIGN
);

其中，維表中最后一個(gè)字段，必須為SIDE DIMENSION SIGN，以此來(lái)標(biāo)識(shí)這是一張維度表。

1. MySQL 維表

（1）示例

? 以 MySQL 作為維度表，建表語(yǔ)句示例如下：

CREATE TABLE ut_mysql_side(
    uid VARCHAR
    sex VARCHAR
    age VARCHAR
    PRIMARY KEY(uid)
    SIDE DIMENSION SIGN
 )WITH(
    type =mysql
    url =jdbc:mysql://linux01:3306/db_flink_test?useUnicode=true&characterEncoding=utf-8
    userName =root
    password =123456
    tableName =ut_mysql_side
    cache =LRU
    cacheSize =2
    cacheTTLMs =2001
    parallelism =1
    partitionedJoin=false
 );

上述代碼用于在 UFlink 中創(chuàng)建一張名為 ut_mysql_side 的表，并聲明表中字段信息（屬性名及屬性類型），并可以指定主鍵對(duì)應(yīng)的列，并且，涉及到 JOIN 操作的列，必須通過(guò)PRIMARY KEY 關(guān)鍵字進(jìn)行聲明，其中可以指定多個(gè)列（逗號(hào)分隔）。

隨后，必須在 WITH 代碼塊中指定數(shù)據(jù)輸出時(shí)的相關(guān)參數(shù)，以及數(shù)據(jù)源所使用的框架的必要參數(shù)。

（2）相關(guān)參數(shù)

可配置參數(shù)如下表所示：

2. PostgreSQL 維表

（1）示例

? 以 MySQL 作為維度表，建表語(yǔ)句示例如下：

CREATE TABLE ut_postgre_side(
    uid VARCHAR
    sex VARCHAR
    age VARCHAR
    PRIMARY KEY(uid)
    SIDE DIMENSION SIGN
 )WITH(
    type =postgre
    url = jdbc:postgresql://10.9.116.184:5432/db_flink_test?useUnicode=true&characterEncoding=utf-8
    userName =root
    password =birthDAY+-0230
    tableName =ut_postgre_side
    parallelism =1
    cache =LRU
    cacheSize =2
    cacheTTLMs =2001
    partitionedJoin=false
 );

上述代碼用于在 UFlink 中創(chuàng)建一張名為 ut_mysql_side 的表，并聲明表中字段信息（屬性名及屬性類型），并可以指定主鍵對(duì)應(yīng)的列，并且，涉及到 JOIN 操作的列，必須通過(guò)PRIMARY KEY 關(guān)鍵字進(jìn)行聲明，其中可以指定多個(gè)列（逗號(hào)分隔）。

隨后，必須在 WITH 代碼塊中指定數(shù)據(jù)輸出時(shí)的相關(guān)參數(shù)，以及數(shù)據(jù)源所使用的框架的必要參數(shù)。

（2）相關(guān)參數(shù)

可配置參數(shù)如下表所示：

2.4 創(chuàng)建視圖

? 在 UFlink 中創(chuàng)建視圖遵循如下邏輯：

CREATE VIEW view_name AS SELECT columnName [ columnName]* FROM table_name[view_name];

在 Flink 中通過(guò)創(chuàng)建臨時(shí)視圖，可以降低復(fù)雜 SQL 的理解成本，從而更好的管理 SQL 邏輯。

3 DML 語(yǔ)句

3.1 INSERT INTO 語(yǔ)句

在 UFlink 中使用 INSERT INTO 語(yǔ)句遵循如下邏輯：

INSERT INTO 
    table_name
    SELECT 
      [ (columnName[  columnName]*) ]
       queryStatement;

例如如下 SQL 操作：

INSERT INTO
    ut_kafka_result
    SELECT
        t3.uid
        t3.pay_cash
        t3.sex
        t3.age
    FROM
        (SELECT
            t1.*
            t2.uid
            t2.sex
            t2.age
        FROM
            ut_kafka_source t1
        JOIN
            ut_mysql_side t2
        ON
            t1.user_id = t2.uid
        WHERE
            t2.sex = 男
        AND
            t1.pay_cash > 100) AS t3

對(duì)于 INSERT INTO 操作，需遵循如下約束：

• 數(shù)據(jù)源表：只能 FROM，不能 INSERT
• 數(shù)據(jù)維表：只能 JOIN，不能 INSERT
• 數(shù)據(jù)結(jié)果表：只能 INSERT
• 視圖：只能 FROM，不能 INSERT

4 QUERY 語(yǔ)句

UFlink 中允許直接對(duì) SELECT 中涉及的字段應(yīng)用 UDXF函數(shù)，從實(shí)現(xiàn)更復(fù)雜的字段到列的映射關(guān)系。

例如下面的數(shù)據(jù)：

{"grade_data":[{"class1":{"小明":"98""老王":"100"}}{"class2":{"凱特琳":"88""凱南":"99"}}]}

通過(guò)如下 SQL 語(yǔ)句進(jìn)行 行轉(zhuǎn)列 操作：

CREATE TABLE FUNCTION ParseArrayUDTF WITH cn.ucloud.sql.ParseArrayUDTF;

INSERT INTO
    tb_class_grade_result
    SELECT
        class
        student
        grade
    FROM
        tb_class_grade LATERAL TABLE(ParseArrayUDTF(grade_data)) as T(class student grade)

執(zhí)行完畢后，輸出效果為：

其中自定義函數(shù)的具體實(shí)現(xiàn)可參考第 5 小節(jié)中的內(nèi)容。

同時(shí)也可以使用 WHERE 和 JOIN 語(yǔ)法，例如如下 SQL 語(yǔ)句：

SELECT
    t3.uid
    t3.pay_cash
    t3.sex
    t3.age
FROM
    (SELECT
        t1.*
        t2.uid
        t2.sex
        t2.age
    FROM
        ut_kafka_source t1
    JOIN
        ut_mysql_side t2
    ON
        t1.user_id = t2.uid
    WHERE
        t2.sex = 男
    AND
        t1.pay_cash > 100) AS t3

另外同時(shí)支持 Flink 原生所有的 SQL 語(yǔ)法。

5 UDXF 自定義函數(shù)

在 Flink 中對(duì)某列數(shù)據(jù)進(jìn)行復(fù)雜處理時(shí)，如果想要實(shí)現(xiàn)更有針對(duì)性的邏輯操作，可以通過(guò)實(shí)現(xiàn) UDXF 函數(shù)來(lái)實(shí)現(xiàn)。

5.1 UDF 自定義標(biāo)量函數(shù)

該類型函數(shù)主要特點(diǎn)為“一進(jìn)一出”，即，依次對(duì)某單列數(shù)據(jù)進(jìn)行處理，并輸出當(dāng)前列的處理結(jié)果（依然為單列）。

1. 實(shí)現(xiàn)函數(shù)

（1） 規(guī)則

編寫 UFlink 的 UDF 函數(shù)代碼遵循如下規(guī)范：

public class XXX extends ScalarFunction{
    public OUT eval(IN columnValue){
        ...
    }
}

（2） 示例

以對(duì)某列數(shù)據(jù)進(jìn)行大寫轉(zhuǎn)換為例，通過(guò)繼承 ScalarFunction 類實(shí)現(xiàn) UDF 函數(shù)，代碼如下：

package cn.ucloud.sql;

public class TransCaseUDF extends ScalarFunction {
    private static final Logger LOG = LoggerFactory.getLogger(TransCaseUDF.class);
    public String eval(object data){
        return new String(data.toString().toLowerCase().getBytes() StandardCharsets.UTF_8);
    }
}

其中 eval 為固定方法名。

2. 注冊(cè)函數(shù)

（1） 規(guī)則

在 UFlink SQL 中，注冊(cè)標(biāo)量函數(shù)（Scalar）方式如下：

CREATE SCALAR FUNCTION [UDFName] WITH [UDF Class Name];

（2） 示例

注冊(cè)示例如下所示：

CREATE SCALAR FUNCTION TransCaseUDF WITH cn.ucloud.sql.TransCaseUDF;

3. 使用函數(shù)

結(jié)合上述示例，使用該函數(shù)方法如下：

INSERT INTO
    tb_name
    SELECT
        TransCaseUDF(name) as upper_name
    FROM
        tb_user_info;

5.2 UDAF 自定義聚合函數(shù)

該類型函數(shù)主要特點(diǎn)為 “多進(jìn)一出”，即，對(duì)某幾列多行數(shù)據(jù)進(jìn)行聚合，并輸出聚合結(jié)果。

1. 實(shí)現(xiàn)函數(shù)

（1） 規(guī)則

編寫 UFlink 的 UDAF 函數(shù)代碼遵循如下規(guī)范：

• 定義累加器

public static class XXXAccum {...)

• 定義類繼承自AggregateFunction

public class XXXUDAF extends AggregateFunction {
    public XXXAccum createAccumulator(){...} //創(chuàng)建累加器
    public T getValue(){...} //獲取聚合結(jié)果
    public void accumulate(){...} //累加
    public void merge(){...} //合并多個(gè)分區(qū)的累加器
    public void resetAccumulator(){...} //重置累加器
}

（2） 示例

以求某列的平均數(shù)為例，實(shí)現(xiàn)方法如下：

package cn.ucloud.sql;
import org.apache.flink.table.functions.AggregateFunction;
import java.util.Iterator;

public class AverageUDAF extends AggregateFunction {
    public static class AverageAccum {
        public float sum count = 0F;
    }

    @Override
    public AverageAccum createAccumulator() {
        return new AverageAccum();
    }

    @Override
    public Float getValue(AverageAccum accumulator) {
        if(accumulator.count == 0F) return null;
        else return accumulator.sum / accumulator.count;
    }

    public void accumulate(AverageAccum accumulator float value) {
        accumulator.sum += value;
        accumulator.count += 1F;
    }

    public void merge(AverageAccum accumulator Iterable it) {
        Iterator iter = it.iterator();

        while(iter.hasNext()) {
            AverageAccum acc = iter.next();
            accumulator.sum += acc.sum;
            accumulator.count += acc.count;
        }
    }

    public void resetAccumulator(AverageAccum accumulator) {
        accumulator.sum = 0F;
        accumulator.count = 0F;
    }
}

2. 注冊(cè)函數(shù)

（1） 規(guī)則

在 UFlink SQL 中，注冊(cè)聚合函數(shù)（Aggregate）方式如下：

CREATE AGGREGATE FUNCTION [UDAFName] WITH [UDAF Class Name];

（2） 示例

結(jié)合上述示例，使用該函數(shù)方法如下：

CREATE AGGREGATE FUNCTION AverageUDAF WITH cn.ucloud.sql.AverageUDAF;

3. 使用函數(shù)

結(jié)合上述示例，使用該函數(shù)方法如下：

CREATE AGGREGATE FUNCTION AverageUDAF WITH cn.ucloud.sql.AverageUDAF;

CREATE TABLE ut_kafka_source(
    sex VARCHAR
    age FLOAT
 )WITH(
    type =kafka11
    bootstrapServers =linux01:9092linux02:9092linux03:9092
    topic =flink-test01
    groupId = group1
    parallelism =1
 );

CREATE TABLE ut_mysql_age_result(
    sex VARCHAR
    average_age FLOAT
 )WITH(
    type =mysql
    url = jdbc:mysql://linux01:3306/db_flink_test?useUnicode=true&characterEncoding=utf-8
    userName =root
    password =123456
    tableName =ut_mysql_age_result
    parallelism =1
 );

INSERT INTO
    ut_mysql_age_result
    SELECT
        sex
        AverageUDAF(age) AS average_age
    FROM
        ut_kafka_source
    GROUP BY
        sex;

5.3 UDTF 自定義表值函數(shù)

該類型函數(shù)主要特點(diǎn)為 “一進(jìn)多出”，即，對(duì)某一行一列數(shù)據(jù)進(jìn)行拆分，并輸出結(jié)果到某列多行。

1. 實(shí)現(xiàn)函數(shù)

（1） 規(guī)則

編寫 UFlink 的 UDTF 函數(shù)代碼遵循如下規(guī)范：

public class ParseArrayUDTF extends TableFunction {
    public void eval(T obj) {
       ...
    }

    @Override
    public TypeInformation getResultType() {
        return Types.ROW(A.class B.class C.class);
    }
}

（2） 示例

以拆分如下數(shù)據(jù)為例：

{"grade_data":[{"class1":{"小明":"98""老王":"100"}}{"class2":{"凱特琳":"88""凱南":"99"}}]}

將其拆分為如下表：

此時(shí)，UDTF 函數(shù)實(shí)現(xiàn)如下：

package cn.ucloud.sql;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import java.util.List;
import java.util.Map;

public class ParseArrayUDTF extends TableFunction {
    //list 數(shù)據(jù)為 [{"class1":{"小明":"98""老王":"100"}}{"class2":{"凱特琳":"88""凱南":"99"}}]
    public void eval(List>> list) {
        for (Map> element : list) {
            for (String className : element.keySet()) {
                Map userMap = element.get(className);
                for(String userName: userMap.keySet()){
                    Row row = new Row(3);
                    row.setField(0 className);
                    row.setField(1 userName);
                    row.setField(2 userMap.get(userName));
                    collect(row);
                }

            }
        }
    }

    @Override
    public TypeInformation getResultType() {
        return Types.ROW(Types.STRING() Types.STRING() Types.STRING());
    }
}

2. 注冊(cè)函數(shù)

（1） 規(guī)則

在 UFlink SQL 中，注冊(cè)聚合函數(shù)（Aggregate）方式如下：

CREATE TABLE FUNCTION [UDTFName] WITH [UDTF Class Name];

（2） 示例

結(jié)合上述示例，使用該函數(shù)方法如下：

CREATE TABLE FUNCTION ParseArrayUDTF WITH cn.ucloud.sql.ParseArrayUDTF;

3. 使用函數(shù)

結(jié)合上述示例，使用該函數(shù)方法如下：

CREATE TABLE FUNCTION ParseArrayUDTF WITH cn.ucloud.sql.ParseArrayUDTF;

CREATE TABLE tb_class_grade(
    grade_data ARRAY
 )WITH(
    type =kafka11
    bootstrapServers =linux01:9092linux02:9092linux03:9092
    zookeeperQuorum =linux01:2181/kafka
    offsetReset =latest
    topic =flink-sql-test02
    groupId = group1
    parallelism =1
 );


CREATE TABLE tb_class_grade_result(
    class VARCHAR
    student VARCHAR
    grade VARCHAR
)WITH(
    type =mysql
    url = jdbc:mysql://linux01:3306/db_flink_test?useUnicode=true&characterEncoding=utf-8
    userName =root
    password =123456
    tableName =tb_class_grade_result
    parallelism =1
);

INSERT INTO
    tb_class_grade_result
    SELECT
        class
        student
        grade
    FROM
        tb_class_grade LATERAL TABLE(ParseArrayUDTF(grade_data)) as T(class student grade)

6 Watermark 的使用

6.1 Watermark 的使用規(guī)則

在創(chuàng)建表時(shí)，在屬性字段的最后一行，為時(shí)間列創(chuàng)建Watermark，語(yǔ)法如下：

WATERMARK FOR columnName AS WITHOFFSET(columnName delayTime(ms))    

其中 columnName 必須為 LONG ，或BIGINT，或 TimeStamp類型的事件時(shí)間。

隨后，F(xiàn)link 將為該表生成 ROWTIME 字段，用于記錄并表示事件時(shí)間。

6.2 Watermark 的使用示例

舉個(gè)例子來(lái)說(shuō)明 Watermark 的相關(guān)用法，示例 SQL 內(nèi)容如下：

CREATE TABLE ut_kafka_source(
    user_id VARCHAR
    pay_cash FLOAT
    pay_time LONG
    WATERMARK FOR pay_time AS WITHOFFSET(pay_time 2000)
 )WITH(
    type =kafka11
    bootstrapServers =linux01:9092linux02:9092linux03:9092
    topic =flink-test01
    groupId = group1
    parallelism =1
 );

CREATE TABLE ut_kafka_result(
    user_id VARCHAR
    pay_cash FLOAT
 )WITH(
    type =kafka11
    bootstrapServers =linux01:9092linux02:9092linux03:9092
    topic =flink-test02
    parallelism =1
 );

INSERT INTO
    ut_kafka_result
    SELECT
        user_id
        SUM(pay_cash) as pay_cash
    FROM
        ut_kafka_source
    GROUP BY
        TUMBLE (
            ROWTIME
            INTERVAL 2 SECOND
        )
        user_id

其中，TUMBLE 為生成 滾動(dòng)窗口 的函數(shù)，ROWTIME 為當(dāng)前表的事件時(shí)間。
提示：上述操作需要在提交 SQL 任務(wù)時(shí)添加指定參數(shù)：time.characteristic: EventTime

實(shí)時(shí)文檔歡迎https://docs.ucloud.cn/uflink/dev/sql