使用 Apache SeaTunnel 实现 Kafka Source 解析复杂Json 案例

版本说明：
SeaTunnel：apache-seatunnel-2.3.2-SNAPHOT
引擎说明：
Flink：1.16.2
Zeta：官方自带

前言

近些时间，我们正好接手一个数据集成项目，数据上游方是给我们投递到Kafka，我们一开始的技术选型是SpringBoot+Flink对上游数据进行加工处理（下文简称：方案一），由于测试不到位，后来到线上，发现数据写入效率完全不符合预期。后来将目光转到开源项目SeaTunnel上面，发现Source支持Kafka，于是开始研究测试，开发环境测试了500w+数据，发现效率在10000/s左右。果断放弃方案一，采取SeaTunnel对数据进行集成加工（下文简称：方案二）。在SeaTunnel研究的过程中，总结了两种方法，方法二相较于方法一，可以实现全场景使用，无需担心字段值里面各种意想不到的字符对数据落地造成错位现象的发生。
对比


在方案二的基础上又衍生出两种方法

所以，在经过长时间的探索和我们线上验证得出结论，建议使用方案二的方法二。
好了，我们进入正文，主篇幅主要介绍方案二中的两种方法，让大家主观的感受SeaTunnel的神奇。
方案一 Springboot+Flink实现Kafka 复杂JSON的解析

网上案例很多，在此不做过多介绍。
方案二 SeaTunnel实现Kafka复杂JSON的解析

在开始介绍之前，我们看一下我们上游方投递Kafka的Json样例数据（对部分数据进行了敏感处理），如下：

1.     "magic": "a***G",
2.     "type": "D***",
3.     "headers": null,
4.     "messageSchemaId": null,
5.     "messageSchema": null,
6.     "message": {
7.         "data": {
8.             "LSH": "187eb13****l0214723",
9.             "NSRSBH": "9134****XERG56",
10.             "QMYC": "01*****135468",
11.             "QMZ": "1Zr*****UYGy%2F5bOWtrh",
12.             "QM_SJ": "2023-05-05 16:42:10.000000",
13.             "YX_BZ": "Y",
14.             "ZGHQ_BZ": "Y",
15.             "ZGHQ_SJ": "2023-06-26 16:57:17.000000",
16.             "SKSSQ": 202304,
17.             "SWJG_DM": "",
18.             "SWRY_DM": "00",
19.             "CZSJ": "2023-05-05 16:42:10.000000",
20.             "YNSRSBH": "9134****XERG56",
21.             "SJTBSJ": "2023-06-26 19:29:59.000",
22.             "SJCZBS": "I"
23.         },
24.         "beforeData": null,
25.         "headers": {
26.             "operation": "INSERT",
27.             "changeSequence": "12440977",
28.             "timestamp": "2023-06-26T19:29:59.673000",
29.             "streamPosition": "00****3.16",
30.             "transactionId": "000***0006B0002",
31.             "changeMask": "0FFF***FF",
32.             "columnMask": "0F***FFFF",
33.             "transactionEventCounter": 1,
34.             "transactionLastEvent": false
35.         }
36.     }
37. }

复制代码

方法一、不通过UDF函数实现

存在问题：字段值存在分隔符，例如‘,’  则数据在落地的时候会发生错位现象。
该方法主要使用官网 transform-v2的各种转换插件进行实现，主要用到的插件有 Replace、Split以及Sql实现
ST脚本：（ybjc_qrqm.conf）

1. env {
2.   execution.parallelism = 100
3.   job.mode = "STREAMING"
4.   job.name = "kafka2mysql_ybjc"
5.   execution.checkpoint.interval = 60000
6. }
8. source {
9.   Kafka {
10.     result_table_name = "DZFP_***_QRQM1"
11.     topic = "DZFP_***_QRQM"
12.     bootstrap.servers = "centos1:19092,centos2:19092,centos3:19092"
13.     schema = {
14.       fields {
15.         message =  {
16.             data = {
17.                 LSH = "string",
18.                 NSRSBH =  "string",
19.                 QMYC =  "string",
20.                 QMZ =  "string",
21.                 QM_SJ =  "string",
22.                 YX_BZ =  "string",
23.                 ZGHQ_BZ =  "string",
24.                 ZGHQ_SJ =  "string",
25.                 SKSSQ =  "string",
26.                 SWJG_DM = "string",
27.                 SWRY_DM = "string",
28.                 CZSJ = "string",
29.                 YNSRSBH = "string",
30.                 SJTBSJ = "string",
31.                 SJCZBS = "string"
32.             }
33.         }
34.       }  
35.     }
36.           start_mode = "earliest"
37.     #start_mode.offsets = {
38.     #     0 = 0
39.     #     1 = 0
40.     #     2 = 0
41.     #}       
42.     kafka.config = {
43.       auto.offset.reset = "earliest"
44.       enable.auto.commit = "true"
45.             # max.poll.interval.ms = 30000000
46.             max.partition.fetch.bytes = "5242880"
47.             session.timeout.ms = "30000"
48.       max.poll.records = "100000"
49.     }
50.   }
51. }
53. transform {
54. Replace {
55.     source_table_name = "DZFP_***_QRQM1"
56.     result_table_name = "DZFP_***_QRQM2"
57.     replace_field = "message"
58.     pattern = "[["
59.     replacement = ""
60.     #is_regex = true
61.           #replace_first = true
62.   }
63.   Replace {
64.     source_table_name = "DZFP_***_QRQM2"
65.     result_table_name = "DZFP_***_QRQM3"
66.     replace_field = "message"
67.     pattern = "]]"
68.     replacement = ""
69.     #is_regex = true
70.           #replace_first = true
71.   }
72.   
73.   Split {
74.     source_table_name = "DZFP_***_QRQM3"
75.     result_table_name = "DZFP_***_QRQM4"
76.     # 存在问题： 如果字段值存在分隔符 separator，则数据会错位
77.     separator = ","
78.     split_field = "message"
79.     # 你的第一个字段包含在zwf5里面，，前五个占位符是固定的。
80.     output_fields = [zwf1,zwf2,zwf3,zwf4,zwf5,nsrsbh,qmyc,qmz,qm_sj,yx_bz,zghq_bz,zghq_sj,skssq,swjg_dm ,swry_dm ,czsj ,ynsrsbh ,sjtbsj ,sjczbs]
81.   }
82.   
83.   sql{
84.    source_table_name = "DZFP_***_QRQM4"
85.    query = "select replace(zwf5 ,'fields=[','') as lsh,nsrsbh,trim(qmyc) as qmyc,qmz,qm_sj,yx_bz, zghq_bz,zghq_sj,skssq,swjg_dm ,swry_dm ,czsj ,ynsrsbh ,sjtbsj ,replace(sjczbs,']}]}','') as sjczbs  from DZFP_DZDZ_QRPT_YWRZ_QRQM4 where skssq <> ' null'"
86.    result_table_name = "DZFP_***_QRQM5"
87.   }
88.   
89. }
91. sink {
92.         Console {
93.                 source_table_name = "DZFP_***_QRQM5"
94.         }
95.        
96.     jdbc {
97.             source_table_name = "DZFP_***_QRQM5"
98.         url = "jdbc:mysql://localhost:3306/dbname?serverTimezone=GMT%2b8"
99.         driver = "com.mysql.cj.jdbc.Driver"
100.         user = "user"
101.         password = "pwd"
102.         batch_size = 200000
103.         database = "dbname"
104.         table = "tablename"
105.                                 generate_sink_sql = true
106.         primary_keys = ["nsrsbh","skssq"]
107.     }
108. }

复制代码

正常写入数据是可以写入了。
写入成功如下：
● kafka源数据：

● tidb目标数据：

现在我们模拟给kafka发送一条数据，其中，SJTBSJ字段我在中间设置一个, 是逗号。
原始值：2023-06-26 19:29:59.000 更改之后的值2023-06-26 19:29:59.0,00

1. 往topic生产一条数据命令
2. kafka-console-producer.sh --topic DZFP_***_QRQM --broker-list centos1:19092,centos2:19092,centos3:19092

复制代码

发送如下：

1.     "magic": "a***G",
2.     "type": "D***",
3.     "headers": null,
4.     "messageSchemaId": null,
5.     "messageSchema": null,
6.     "message": {
7.         "data": {
8.             "LSH": "187eb13****l0214723",
9.             "NSRSBH": "9134****XERG56",
10.             "QMYC": "01*****135468",
11.             "QMZ": "1Zr*****UYGy%2F5bOWtrh",
12.             "QM_SJ": "2023-05-05 16:42:10.000000",
13.             "YX_BZ": "Y",
14.             "ZGHQ_BZ": "Y",
15.             "ZGHQ_SJ": "2023-06-26 16:57:17.000000",
16.             "SKSSQ": 202304,
17.             "SWJG_DM": "",
18.             "SWRY_DM": "00",
19.             "CZSJ": "2023-05-05 16:42:10.000000",
20.             "YNSRSBH": "9134****XERG56",
21.             "SJTBSJ": "2023-06-26 19:29:59.0,00",
22.             "SJCZBS": "I"
23.         },
24.         "beforeData": null,
25.         "headers": {
26.             "operation": "INSERT",
27.             "changeSequence": "12440977",
28.             "timestamp": "2023-06-26T19:29:59.673000",
29.             "streamPosition": "00****3.16",
30.             "transactionId": "000***0006B0002",
31.             "changeMask": "0FFF***FF",
32.             "columnMask": "0F***FFFF",
33.             "transactionEventCounter": 1,
34.             "transactionLastEvent": false
35.         }
36.     }
37. }

复制代码

写入之后，发现数据错位了。

结论：其实这个问题线上还是能遇到的，比如地址字段里面含有逗号，备注信息里面含有逗号等等，这种现象是不可避免的，所以此种方案直接pass。对数据危害性极大！可以处理简单的数据，当做一种思路。
方法二：通过UDF函数实现

该方法通过UDF函数扩展（https://seatunnel.apache.org/docs/2.3.2/transform-v2/sql-udf）的方式，实现嵌套kafka source json源数据的解析。可以大大简化ST脚本的配置
ST脚本：（ybjc_qrqm_yh.conf）

1. env {
2.     execution.parallelism = 5
3.     job.mode = "STREAMING"
4.     job.name = "kafka2mysql_ybjc_yh"
5.     execution.checkpoint.interval = 60000
6. }
8. source {
9.     Kafka {
10.         result_table_name = "DZFP_***_QRQM1"
11.         topic = "DZFP_***_QRQM"
12.         bootstrap.servers = "centos1:19092,centos2:19092,centos3:19092"
13.         schema = {
14.         fields {
15.              message =  {
16.                 data = "map<string,string>"
17.              }
18.         }  
19.         }
20.         start_mode = "earliest"
21.         #start_mode.offsets = {
22.         #     0 = 0
23.         #     1 = 0
24.         #     2 = 0
25.         #}       
26.         kafka.config = {
27.         auto.offset.reset = "earliest"
28.         enable.auto.commit = "true"
29.         # max.poll.interval.ms = 30000000
30.         max.partition.fetch.bytes = "5242880"
31.         session.timeout.ms = "30000"
32.         max.poll.records = "100000"
33.         }
34.     }
35. }
36. transform {
37.     sql{
38.         source_table_name = "DZFP_***_QRQM1"
39.         result_table_name = "DZFP_***_QRQM2"
40.         # 这里的qdmx就是我自定义的UDF函数，具体实现下文详细讲解。。。
41.         query = "select qdmx(message,'lsh') as lsh,qdmx(message,'nsrsbh') as nsrsbh,qdmx(message,'qmyc') as qmyc,qdmx(message,'qmz') as qmz,qdmx(message,'qm_sj') as qm_sj,qdmx(message,'yx_bz') as yx_bz,qdmx(message,'zghq_bz') as zghq_bz,qdmx(message,'zghq_sj') as zghq_sj,qdmx(message,'skssq') as skssq,qdmx(message,'swjg_dm') as swjg_dm,qdmx(message,'swry_dm') as swry_dm,qdmx(message,'czsj') as czsj,qdmx(message,'ynsrsbh') as ynsrsbh, qdmx(message,'sjtbsj') as sjtbsj,qdmx(message,'sjczbs') as sjczbs  from  DZFP_DZDZ_QRPT_YWRZ_QRQM1"
42.     }
43. }
45.    
46. sink {
47.     Console {
48.         source_table_name = "DZFP_***_QRQM2"
49.     }
50.     jdbc {
51.                   source_table_name = "DZFP_***_QRQM2"
52.                   url = "jdbc:mysql://localhost:3306/dbname?serverTimezone=GMT%2b8"
53.                   driver = "com.mysql.cj.jdbc.Driver"
54.                   user = "user"
55.                   password = "pwd"
56.                   batch_size = 200000
57.                   database = "dbname"
58.                   table = "tablename"
59.                   generate_sink_sql = true
60.                   primary_keys = ["nsrsbh","skssq"]
61.         }
62. }

复制代码

执行脚本：查看结果，发现并没有错位，还在原来的字段（sjtbsj）上面。
这种方法，是通过key获取value值。不会出现方法一中的按照逗号分割出现数据错位现象。

具体UDF函数编写如下。
maven引入如下：

1. <dependencies>
2.   <dependency>
3.     <groupId>org.apache.seatunnel</groupId>
4.     <artifactId>seatunnel-transforms-v2</artifactId>
5.     <version>2.3.2</version>
6.     <scope>provided</scope>
7.   </dependency>
9.   <dependency>
10.     <groupId>org.apache.seatunnel</groupId>
11.     <artifactId>seatunnel-api</artifactId>
12.     <version>2.3.2</version>
13.   </dependency>
15.   <dependency>
16.     <groupId>cn.hutool</groupId>
17.     <artifactId>hutool-all</artifactId>
18.     <version>5.8.20</version>
19.   </dependency>
21.   <dependency>
22.     <groupId>com.google.auto.service</groupId>
23.     <artifactId>auto-service-annotations</artifactId>
24.     <version>1.1.1</version>
25.     <optional>true</optional>
26.     <scope>compile</scope>
27.   </dependency>
29.   <dependency>
30.     <groupId>com.google.auto.service</groupId>
31.     <artifactId>auto-service</artifactId>
32.     <version>1.1.1</version>
33.     <optional>true</optional>
34.     <scope>compile</scope>
35.   </dependency>
36. </dependencies>  

复制代码

UDF具体实现java代码如下：
[code]package org.seatunnel.sqlUDF;import cn.hutool.core.util.StrUtil;import cn.hutool.json.JSONObject;import cn.hutool.json.JSONUtil;import com.google.auto.service.AutoService;import org.apache.seatunnel.api.table.type.BasicType;import org.apache.seatunnel.api.table.type.SeaTunnelDataType;import org.apache.seatunnel.transform.sql.zeta.ZetaUDF;import java.util.HashMap;import java.util.List;import java.util.Map;@AutoService(ZetaUDF.class)public class QdmxUDF implements ZetaUDF {    @Override    public String functionName() {        return "QDMX";    }    @Override    public SeaTunnelDataType resultType(List