Flink DataStream任务优化为Flink SQL

任务描述：
线上版本背景：之前Flink sql任务不能有多个task，也没有嵌套json解析 udf方法，所以考量用的是 DataStream 实现
优化背景：Flink 目前以实现 一个任务多个task/sink，也有 嵌套json解析 udf方法，为了方便运维，优化成使用 Flink Sql 来实现
用时估计：一星期
实际完成时间：三天（背景：之前已追踪过 Java DataStream 项目代码）

1、Caused by: java.io.IOException: Unknown "type" value "kafka". The Canal JSON message …
‘format’ = ‘canal-json’ 改为 ‘format’ = ‘json’ ，可参考 链接 ,提到了 canal-json 会调用 deserialize 方法，其会查询 key type 是 哪种 DML （update,delete,insert）,其余情况则会Unknown "type" value "kafka"报错。

2、字符串包含判断函数 INTEGER POSITION( x IN y) ,功能：返回目标字符串x在被查询字符串y里第一次出现的位置。如果目标字符串x在被查询字符串y中不存在，返回值为0。

select
POSITION('in' IN 'china') as result
FROM T1; 

3、 如何启用MFA?
有个账号、二维码、密钥。
阿里云APP扫描二维码，就能获取虚拟MFA，用于登录阿里云平台了。
密钥用于windows 的Authy 应用添加账号时使用，就能用windows获取虚拟MFA了。

4、写入多个Sink 的语法，不然会报错。

-- 开始任务
BEGIN STATEMENT SET;      --写入多个Sink时，必填。
...
END;      --写入多个Sink时，必填。

5、source kafka ,sink holo ,发生部分字段值未sink过来，找了半天问题发现是有带大写字母的字段，即获取时，key值是大小写敏感的，传全小写值是获取不到的。
6、 Flink内置函数
7、JSON取值 实现：

--简单 JSON值 获取
select JSON_VALUE('{"aid":"5awplkboi3nr4rpmhkrvex11Bfopln6"}'
, '$.aid') as aid; --right result
--带转义字符的 JSON字符串，JSON_VALUE获取不出来，说明 kafka source Log 读取Flink会自动解析，删除掉转义字符
select JSON_VALUE('{\"aid\":\"5awplkboi3nr4rpmhkrvex11Bfopln6\"}'
, '$.aid') as aid; --wrong result null 
select JSON_VALUE('{"message":"{\"aid\":\"5awplkboi3nr4rpmhkrvex11Bfopln6\"}"}'
, '$.message.aid') as aid; --wrong result null

--嵌套 JSON值 获取
select JSON_VALUE('{"log":{"file":{"path":"/data/statistic/logs/data/visit/user-visit.log"},"offset":5066}}'
, '$.log.file.path') as path; --right
select JSON_VALUE('{"message":{"aid":"5awplkboi3nr4rpmhkrvex11Bfopln6"}}'
, '$.message.aid') as aid; --right

--带转义字符的 嵌套 JSON字符串，JSON_VALUE获取不出来
select JSON_VALUE('{"message":"{\"aid\":\"\",\"bizField\":\"\"}'
, '$.message.aid') as aid; --wrong null

8、特殊嵌套JSON取值 实现：

--目标：获取 字符串值 {"eid":"00000810139911"} 里的eid值
--step1.str1 替换掉{、}、"字符串
select REGEXP_REPLACE('{"eid":"00000810139911"}','\{|}|"', ''); -- eid:00000810139911
--step2.string 转换成map，获取key=eid的value
select STR_TO_MAP(REGEXP_REPLACE('{"eid":"00000810139911"}','\{|}|"', ''), ',', ':')['eid']; --00000810139911  
--step3.TYPEOF 函数 查看数据类型，返回类型是STRING
select TYPEOF(STR_TO_MAP(REGEXP_REPLACE('{"eid":"00000810139911"}','\{|}|"', ''), ',', ':')['eid']); --MAP['key'] 返回的是STRING 类型 

--函数测试过程：
--判断 数据类型和值是否相同
select 1 IS DISTINCT FROM 2; --true A 和 B 的数据类型和值不完全相同返回 TRUE
--将字符串转换为 MAP类型
select STR_TO_MAP('eid:"00000810139911"', ',', ':')['eid']; --"00000810139911"
--注意要转成MAP类型的字符串，key不能再带双引号了
select STR_TO_MAP('"eid":"00000810139911"', ',', ':')['eid']; --null 
--替换字符函数的 str1 没有限制，内含花括号、双引号 是没有问题的
select REGEXP_REPLACE('accc}','a', ''); --ccc}
select REGEXP_REPLACE('{accc}','a', ''); --{ccc}
select REGEXP_REPLACE('{a"ccc}','a', ''); --{"ccc}
select REGEXP_REPLACE('{a"ccc"}','a', ''); --{"ccc"}
--匹配字符串str2 带左花括号，替换无效
select REGEXP_REPLACE('{a"ccc"}','{abc}', ''); --null 
select REGEXP_REPLACE('{a"ccc"}','{|}', ''); --null
--替换掉原字符串str1 的左花括号，需要转义字符
select REGEXP_REPLACE('{a"ccc"}','\{|}', ''); --a"ccc" 
select REGEXP_REPLACE('{"eid":"00000810139911"}','\{|}', ''); --"eid":"00000810139911"
--或符号用于str2 分开不同的替换字符串
select REGEXP_REPLACE('foobar','oo|ar', ''); --fb