实时数据入湖的选型和实践

1
2
3
4
5

EMR 6.9.0 
Hudi 0.12.1
Spark 3.3.0 
Flink 1.15.2  
MySQL 8.0.22

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

-- create databases
create database if not exists test_db default character set utf8mb4 collate utf8mb4_general_ci;
use test_db;

-- create  user table
drop table if exists user;
create table if not exists user
(
    id           int auto_increment primary key,
    name         varchar(155)                        null,
    device_model varchar(155)                        null,
    email        varchar(50)                         null,
    phone        varchar(50)                         null,
    create_time  timestamp default CURRENT_TIMESTAMP not null,
    modify_time  timestamp default CURRENT_TIMESTAMP null on update CURRENT_TIMESTAMP
)charset = utf8mb4;

-- insert data
insert into user(name,device_model,email,phone) values
('customer-01','dm-01','abc01@email.com','188776xxxxx'),
('customer-02','dm-02','abc02@email.com','166776xxxxx');

-- create product table
drop table if exists product;
create table if not exists product
(
    pid          int not null primary key,
    pname        varchar(155)                        null,
    pprice       decimal(10,2)                           ,
    create_time  timestamp default CURRENT_TIMESTAMP not null,
    modify_time  timestamp default CURRENT_TIMESTAMP null on update CURRENT_TIMESTAMP
)charset = utf8mb4;

-- insert data
insert into product(pid,pname,pprice) values
('1','prodcut-001',125.12),
('2','prodcut-002',225.31);

-- create order table
drop table if exists user_order;
create table if not exists user_order
(
    id           int auto_increment primary key,
    oid          varchar(155)                        not null,
    uid          int                                         ,
    pid          int                                         ,
    onum         int                                         ,
    create_time  timestamp default CURRENT_TIMESTAMP not null,
    modify_time  timestamp default CURRENT_TIMESTAMP null on update CURRENT_TIMESTAMP
)charset = utf8mb4;

-- insert data
insert into user_order(oid,uid,pid,onum) values 
('o10001',1,1,100),
('o10002',1,2,30),
('o10001',2,1,22),
('o10002',2,2,16);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val params = Config.parseConfig(MySQLCDC, args)
    env.enableCheckpointing(params.checkpointInterval.toInt ` 1000)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    val rocksBackend: StateBackend = new RocksDBStateBackend(params.checkpointDir)
    env.setStateBackend(rocksBackend)

    env.fromSource(createCDCSource(params), WatermarkStrategy.noWatermarks(), "mysql cdc source")
      .addSink(createKafkaSink(params)).name("cdc sink kafka")
      .setParallelism(params.parallel.toInt)

    env.execute("MySQL Binlog CDC")
  }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 创建topic, bs为 kafka broker地址
kafka-topics.sh --create ----bootstrap-server ${bs}  --replication-factor 2 --partitions 8  --topic cdc_topic
# 编译打包
mvn clean package  -Dscope.type=provided  -DskipTests
# disalbe check-leaked-classloader
sudo sed -i -e '$a\classloader.check-leaked-classloader: false' /etc/flink/conf/flink-conf.yaml
# 启动flink cdc 发送数据到Kafka
sudo flink run -m yarn-cluster \
-yjm 1024 -ytm 2048 -d \
-ys 4 -p 8 \
-c  com.kezaihui.MySQLCDC  \
/home/hadoop/emr-flink-cdc-1.0-SNAPSHOT.jar \
-b xxxxx.amazonaws.com:9092 \
-t cdc_topic \
-c s3://xxxxx/flink/checkpoint/ \
-l 30 -h xxxxx.rds.amazonaws.com:3306 -u admin \
-P admin123456 \
-d test_db -T test_db.` \
-p 4 \
-e 5400-5408

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

    val ds = df.selectExpr("CAST(value AS STRING)").as[String]
    val query = ds
      .writeStream
      .queryName("debezium2hudi")
      .option("checkpointLocation", params.checkpointDir)
      // if set 0, as fast as possible
      .trigger(Trigger.ProcessingTime(params.trigger + " seconds"))
      .foreachBatch { (batchDF: Dataset[String], batchId: Long) =>
        log.warn("current batch: " + batchId.toString)
        val pool = Executors.newFixedThreadPool(50)
        implicit val xc: ExecutionContextExecutor = ExecutionContext.fromExecutor(pool)

        val newsDF = batchDF.map(cdc => DebeziumParser.apply().debezium2Hudi(cdc))
          .filter(_ != null)
        if (!newsDF.isEmpty) {
          val tasks = Seq[Future[Unit]]()
          for (tableInfo <- tableInfoList.tableInfo) {
            val insertORUpsertDF = newsDF
              .filter($"database" === tableInfo.database && $"table" === tableInfo.table)
              .filter($"operationType" === HudiOP.UPSERT || $"operationType" === HudiOP.INSERT)
              .select($"data".as("jsonData"))
            if (!insertORUpsertDF.isEmpty) {
              val json_schema = ss.read.json(insertORUpsertDF.select("jsonData").as[String]).schema
              val cdcDF = insertORUpsertDF.select(from_json($"jsonData", json_schema).as("cdc_data"))
              val cdcPartitionDF = cdcDF.select($"cdc_data.`")
                .withColumn(tableInfo.hudiPartitionField, sqlPartitionFunc(col(tableInfo.partitionTimeColumn)))
              params.concurrent match {
                case "true" => {
                  val runTask = HudiWriteTask.run(cdcPartitionDF, params, tableInfo)(xc)
                  tasks :+ runTask
                }
                case _ => HudiWriteTask.runSerial(cdcPartitionDF, params, tableInfo)
              }
            }
          }
          if (params.concurrent == "true" && tasks.nonEmpty) {
            Await.result(Future.sequence(tasks), Duration(60, MINUTES))
            pool.shutdown()
            //            ()
          }

        }
      }.start()
    query.awaitTermination()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 编译打包
mvn clean package  -Dscope.type=provided  -DskipTests
# bs 为 kafka broker 地址
sudo spark-submit --master yarn \
--deploy-mode client \
--driver-memory 1g \
--executor-memory 1g \
--executor-cores 2 \
--num-executors  2 \
--conf spark.dynamicAllocation.enabled=false \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
--conf spark.sql.parquet.binaryAsString=true \
--jars  /usr/lib/hudi/hudi-spark-bundle.jar \
--class com.kezaihui.Debezium2Hudi /home/zaihui/cdc-jar/emr-hudi-spark-1.0-SNAPSHOT-jar-with-dependencies.jar \
-e dev -b $bs \
-t cdc_t -p emr-cdc-group-0 -s true \
-o earliest \
-i 60 -y cow -p 10 \
-c s3://xxxx/spark-checkpoint/emr-cdc-grpup-0/ \
-g s3://xxx/emr-cdc-grpup-0/ \
-w upsert  \
-s hms \
-z thrift://xxx:9083 \
--concurrent true \
-m "{\"tableInfo\":[{\"database\":\"test_db\",\"table\":\"user\",\"hudiTable\":\"user\",\"hiveDatabase\":\"ods\",\"recordKey\":\"id\",\"precombineKey\":\"modify_time\",\"partitionTimeColumn\":\"create_time\",\"hudiPartitionField\":\"year_month\"},
{\"database\":\"test_db\",\"table\":\"user_order\",\"hudiTable\":\"user_order\",\"hiveDatabase\":\"ods\",\"recordKey\":\"id\",\"precombineKey\":\"modify_time\",\"partitionTimeColumn\":\"create_time\",\"hudiPartitionField\":\"year_month\"},{\"database\":\"test_db\",\"table\":\"product\",\"hudiTable\":\"product\",\"hiveDatabase\":\"ods\",\"recordKey\":\"pid\",\"precombineKey\":\"modify_time\",\"partitionTimeColumn\":\"create_time\",\"hudiPartitionField\":\"year_month\"}]}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# checkpoint 地址
checkpoints=s3://xxxxx/flink/checkpoints/datagen/
# 创建 yarn-session,后续的 flink 任务将会提交到这个 cluster 执行
flink-yarn-session -jm 1024 -tm 4096 -s 2  \
-D state.backend=rocksdb \
-D state.checkpoint-storage=filesystem \
-D state.checkpoints.dir=${checkpoints} \
-D execution.checkpointing.interval=5000 \
-D state.checkpoints.num-retained=5 \
-D execution.checkpointing.mode=EXACTLY_ONCE \
-D execution.checkpointing.externalized-checkpoint-retention=RETAIN_ON_CANCELLATION \
-D state.backend.incremental=true \
-D execution.checkpointing.max-concurrent-checkpoints=1 \
-D rest.flamegraph.enabled=true \
-d \

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

CREATE TABLE `user`(
   id string,
   name STRING,
   device_model STRING,
   email STRING,
   phone STRING,
   age string,
   create_time STRING,
   modify_time STRING,
   year_month STRING
)
PARTITIONED BY (`year_month`)
WITH (
  'connector' = 'hudi',
  'path' = 's3://xxx/emr-cdc-grpup-0/test_db/user/',
  'hoodie.datasource.write.recordkey.field' = 'id',
  'table.type' = 'COPY_ON_WRITE',
  'index.bootstrap.enabled' = 'true',
  'read.streaming.enabled' = 'true',
  'read.start-commit' = '20230223014223',
  'changelog.enabled' = 'false',
  'read.streaming.check-interval' = '1'
);

CREATE TABLE `user_order`(
   id string,
   oid string,
   uid STRING,
   pid STRING,
   onum STRING,
   remark STRING,
   create_time STRING,
   modify_time STRING,
   year_month STRING
)
    PARTITIONED BY (`year_month`)
WITH (
  'connector' = 'hudi',
  'path' = 's3://xxx/emr-cdc-grpup-0/test_db/user_order/',
  'hoodie.datasource.write.recordkey.field' = 'id',
  'table.type' = 'COPY_ON_WRITE',
  'index.bootstrap.enabled' = 'true',
  'read.streaming.enabled' = 'true',
  'read.start-commit' = '20230223014223',
  'changelog.enabled' = 'false',
  'read.streaming.check-interval' = '1'
);

-- Flink 聚合操作 Sink 到 Hudi 表
-- batch
CREATE TABLE  user_order_agg(
  oid STRING,
  uid INT,
  pid INT,
  name STRING,
  phone STRING,
  onum BIGINT,
  remark STRING,
  year_month STRING
)   PARTITIONED BY (`year_month`)
WITH(
 'connector' = 'hudi',
 'path' = 's3://xxx/emr-cdc-grpup-0/dws/user_order_agg',
 'table.type' = 'COPY_ON_WRITE',
 'write.precombine.field' = 'oid',
 'write.operation' = 'upsert',
 'hoodie.datasource.write.recordkey.field' = 'oid',
 'hive_sync.enable' = 'true',
 'hive_sync.db' = 'dws',
 'hive_sync.mode' = 'hms',
 'hive_sync.metastore.uris' = 'thrift://xxx:9083'
 );

Insert Into user_order_agg select o.oid, cast(o.uid as int), cast(o.pid as int), u.name, u.phone, cast(o.onum as bigint), o.remark, o.year_month from user_order as o inner join `user` as u on o.uid=u.id;