通过标准化的 flink sql 语言，在线开发、测试、运维流式任务，不依赖jar包，方便更新和维护。

适用场景

1. 对流式数据通过sql的方式做聚合加工，产出实时指标。
2. 对结构比较清晰的数据做清洗过滤的流式ETL需求。
3. 构建流式数据仓库。

数据源的使用

通过DDL语句实现对数据源的使用，目前提供以下几类语法：

1. create table：用于定义表结构，包含有定义源表（source）、结果表（sink）
2. create view：用于定义中间结果视图
3. create function：用于自定义udf的声明
4. insert into XXX select：用于表示真正需要执行的sql语句，并将执行结果insert到对应的结果表中

新建任务

功能入口：

1. 在任务开发首页，点击“新建节点”；
2. 在项目下，左侧目录结构中，右键新建任务；
3. 任务编辑页面，点击“新建”。
   新建任务类型选择“流式sql任务“
   

SQL 编辑

通过DDL语法声明使用数据源，Flink SQL 加工数据，DataLeap 版本的DDL语法与开源版有少量不同，具体看下方的说明与DDL参数表格。
一个完整的任务示例:

1.	-- 1.创建 kafka source table
2.	create table click_data(
3.	    user_id bigint,
4.	    article_id bigint,
5.	    click_time timestamp,
6.	    watermark for click_time as withoffset(click_time, 3000)
7.	) with (
8.	    'connector.type' = 'kafka',
9.	    'connector.version' = '0.10',
10.	    'connector.topic' = 'test_topic',
11.	    'connector.cluster' = 'test_cluster',
12.	    'connector.group.id' = 'flink_1.9_sql_test',
13.	    'update-mode' = 'append',
14.	    'format.type' = 'json',
15.	);
16.	
17.	-- 2.创建 Mysql sink table
18.	create table article_pv(
19.	    article_id bigint,
20.	    PV bigint,
21.	    s_time timestamp
22.	) with (
23.	    'connector.type' = 'jdbc',
24.	    'connector.table' = 'test_table',
25.	    'connector.dbname' = 'testdb',
26.	    'connector.url' = 'mysql.testdb_write',
27.	    'connector.username' = 'sink.test',
28.	    'connector.password' = 'abcabc',
29.	    'connector.write.flush.max-rows' = '5'
30.	 );
31.	-- 3. 计算逻辑
32.	insert into article_pv 
33.	select article_id, count(*) as PV, TUMBLE_START(click_time, INTERVAL '1' HOUR) as s_time 
34.	from click_data 
35.	group by article_id, TUMBLE(click_time, INTERVAL '1' HOUR)

PB 格式定义
使用PB格式的数据源，需要在当前任务上传PB类的定义文件，或手工输入PB类，json格式无需设置。
一次只支持一个PB类的定义，例：

1.	syntax = "proto2";
2.	package abase_test;
3.	message AbaseTest {
4.	required int64 first_id = 1;
5.	required int64 latest_id = 2;
6.	}

说明

1. 如果kafka topic是binlog数据，在flink sql source中指定下'format.type' = 'pb_binlog'即可，数据类型选择'json'方式
2. 如果Kafka topic是自定义pb数据
3. 在flink sql source中指定下'format.pb-class' = '{package的路径}.{java_outer_classname的类名}${入口message名称}'.
4. 譬如,
   a)   `'format.pb-class' = 'parser.proto.ProtoParser$Instance'`
   数据源格式选择"pb","Pb类定义"中开头设置 package 和 option java_outer_classname，"入口message名称" 指定入口message名称
   举例：
   "Pb类定义"开头：
   package parser.proto;
   option java_outer_classname = "ProtoParser";
   "入口message名称":Instance

任务调试

流式任务支持在线调试，通过2种方式获取测试数据：
1.   在当前任务消费的 topic 中抽取最近100条数据
2.   手工上传json格式的文件
测试过程中，SQL 在独立环境运行，所有的输出直接写到本地浏览器，不会对线上生产的数据存储系统造成影响。

参数设置

基础信息：任务名称、描述、责任人、队列等基础信息按需填写。

资源设置

获取推荐配置：根据任务的历史24小时运行情况，给出推荐配置
Flink 运行参数：Flink 相关的动态参数和执行参数，具体设置详见 Flink 官方文档
用户自定义参数：kv格式，按需设置。

提交上线

任务测试通过后，先“保存”任务生成一个草稿版本。然后点击“提交上线”，即按最新的版本启动一个 Flink 实例，后续的运行情况可在“任务运维”中查看。
运维列表的“重启”功能以线上配置为准，不应用最新版本草稿。

历史版本

每次上线会生成一个版本，在“历史版本”功能中可以查看、对比和回滚，回滚的效果是恢复至草稿，需要重新点击上线。

DDL参数设置

DataLeap的流式SQL在DDL部分与开源版本有少量不同，具体参考以下内容。

Kafka DDL 参数

Mysql DDL 参数

Kafka json Format

使用 json format 需在 create table 语句中通过'format.type'='json' 指定。
注意事项：
json类型根据用户在create table中声明的column name和column type做解析。SQL任务目前仅做部分自动适配，例如声明为int类型，但实际json里面是int的string表示，SQL也能自动识别并转换回int。但是如果例如声明为int但实际为long或者声明为int但实际是带有字母的string等，SQL无法直接转换，会报错。
json类型支持嵌套json。例如嵌套json "{"a": "a", "b": {"c": 1, "d": "s"}}", 声明嵌套结构时，需要声明为b Row<c int, d varchar> 即可表明字段c、d是嵌套于字段b中。json也支持array形式。
1.   如果是简单的array，例如"a": [1,2,3,4]，则可声明为a Array，即表明a是一个int类型的array，注意，select 的时候，下标是从 1 开始的。例如"a": [6,7,8,9], select a[1] 会返回 6。
2.   如果是嵌套json的array，例如"a": [{"b": 3}, {"b": 6}]，则可声明为a Array<Row>, 即表明a是一个object类型array，a的结构中，有一个类型为int的b列。获取的方式如 select a[1].b 会返回3.
参数