图中标号 1：日志数据和业务数据发送⾄MSK(Kafka)，通过 Flink(TableAPI) 建立Kafka 表，消费 Kafka 数据，Hive Metastore 存储 Schema；

图中标号 2：RDS(MySQL) 中的数据通过 Flink CDC(flink-cdc-connector) 直接消费 Binlog 数据，⽆需搭建其他消费 Binlog 的服务 (⽐如 Canal,Debezium)。注意使⽤flink-cdc-connector 的 2.x 版本，⽀持parallel reading,lock-free and checkpoint feature；

图中标号 3：使用Flink Hudi Connector, 将数据写⼊Hudi(S3) 表, 对于⽆需 Update 的数据使⽤Insert 模式写⼊，对于需要 Update 的 数据 (业务数据和 CDC 数据) 使用Upsert 模式写⼊；

图中标号 4：使用Presto 作为查询引擎，对外提供查询服务。此条数据链路的延迟取决于入Hudi 的延迟及 Presto 查询的延迟，总体在分钟级别；

图中标号 5：对于需要秒级别延迟的指标，直接在 Flink 引擎中做计算，计算结果输出到 RDS 或者 KV 数据库，对外提供 API 查询服务；

图中标号 6：使用QuickSight 做数据可视化，支持多种数据源接入。

log_uri="s3://*****/emr/log/"key_name="****"jdbc="jdbc:mysql:\/\/*****.ap-southeast-1.rds.amazonaws.com:3306\/hive_metadata_01?createDatabaseIfNotExist=true"cluster_name="tech-talk-001"

# MSK集群创建可以通过CLI, 也可以通过Console创建# 下载kafka,创建topic写⼊数据wget https://dlcdn.apache.org/kafka/2.6.2/kafka_2.12-2.6.2.tgz# msk zk地址，broker 地址zk_servers=*****.c3.kafka.ap-southeast-1.amazonaws.com:2181bootstrap_server=******.5ybaio.c3.kafka.ap-southeast-1.amazonaws.com:9092topic=tech-talk-001# 创建tech-talk-001 topic./bin/kafka-topics.sh --create --zookeeper ${zk_servers} --replication-factor 2 --partitions 4--topic ${topic}# 写⼊消息./bin/kafka-console-producer.sh --bootstrap-server ${bootstrap_server} --topic ${topic}{"id":"1","name":"customer"}{"id":"2","name":"aws"}# 消费消息./bin/kafka-console-consumer.sh --bootstrap-server ${bootstrap_server} --topic ${topic}

# 启动flink on yarn session cluster# 下载kafka connectorsudo wget -P /usr/lib/flink/lib/ https://repo1.maven.org/maven2/org/apache/flink/flink-sql?connector-kafka_2.12/1.13.1/flink-sql-connector-kafka_2.12-1.13.1.jar && sudo chown flink:flink/usr/lib/flink/lib/flink-sql-connector-kafka_2.12-1.13.1.jar# hudi-flink-bundle 0.10.0sudo wget -P /usr/lib/flink/lib/ https://dxs9dnjebzm6y.cloudfront.net/tmp/hudi-flink?bundle_2.12-0.10.0-SNAPSHOT.jar && sudo chown flink:flink /usr/lib/flink/lib/hudi-flink?bundle_2.12-0.10.0-SNAPSHOT.jar# 下载 cdc connectorsudo wget -P /usr/lib/flink/lib/ https://repo1.maven.org/maven2/com/ververica/flink-sql?connector-mysql-cdc/2.0.0/flink-sql-connector-mysql-cdc-2.0.0.jar && sudo chown flink:flink/usr/lib/flink/lib/flink-sql-connector-mysql-cdc-2.0.0.jar# flink sessionflink-yarn-session -jm 1024 -tm 4096 -s 2 \-D state.checkpoints.dir=s3://*****/flink/checkpoints \-D state.backend=rocksdb \-D state.checkpoint-storage=filesystem \-D execution.checkpointing.interval=60000 \-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

# 这是使⽤flink sql client写SQL提交作业# 启动client/usr/lib/flink/bin/sql-client.sh -s application_*****# result-modeset sql-client.execution.result-mode=tableau;# set default parallesimset 'parallelism.default' = '1';

# 创建kafka表CREATE TABLE kafka_tb_001 (id string,name string,`ts` TIMESTAMP(3) METADATA FROM 'timestamp') WITH ('connector' = 'kafka','topic' = 'tech-talk-001','properties.bootstrap.servers' = '****:9092','properties.group.id' = 'test-group-001','scan.startup.mode' = 'latest-offset','format' = 'json','json.ignore-parse-errors' = 'true','json.fail-on-missing-field' = 'false','sink.parallelism' = '2');# 创建flink hudi表CREATE TABLE flink_hudi_tb_106(uuid string,name string,ts TIMESTAMP(3),logday VARCHAR(255),hh VARCHAR(255))PARTITIONED BY (`logday`,`hh`)WITH ('connector' = 'hudi','path' = 's3://*****/teck-talk/flink_hudi_tb_106/','table.type' = 'COPY_ON_WRITE','write.precombine.field' = 'ts','write.operation' = 'upsert','hoodie.datasource.write.recordkey.field' = 'uuid','hive_sync.enable' = 'true','hive_sync.metastore.uris' = 'thrift://******:9083','hive_sync.table' = 'flink_hudi_tb_106','hive_sync.mode' = 'HMS','hive_sync.username' = 'hadoop','hive_sync.partition_fields' = 'logday,hh','hive_sync.partition_extractor_class' = 'org.apache.hudi.hive.MultiPartKeysValueExtractor');# 插⼊数据insert into flink_hudi_tb_106 select id as uuid,name,ts,DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy?MM-dd') as logday, DATE_FORMAT(CURRENT_TIMESTAMP, 'hh') as hh from kafka_tb_001;# 除了在创建表是指定同步数据的⽅式，也可以通过cli同步hudi表元数据到hive,但要注意分区格式./run_sync_tool.sh --jdbc-url jdbc:hive2:\/\/*****:10000 --user hadop --pass hadoop --partitioned-by logday --base-path s3://****/ --database default --table *****# presto 查询数据presto-cli --server *****:8889 --catalog hive --schema default

# 创建mysql CDC表CREATE TABLE mysql_cdc_002 (id INT NOT NULL,name STRING,create_time TIMESTAMP(3),modify_time TIMESTAMP(3),PRIMARY KEY(id) NOT ENFORCED) WITH ('connector' = 'mysql-cdc','hostname' = '*******','port' = '3306','username' = 'admin','password' = '*****','database-name' = 'cdc_test_db','table-name' = 'test_tb_01','scan.startup.mode' = 'initial');# 创建hudi表CREATE TABLE hudi_cdc_002 (id INT ,name STRING,create_time TIMESTAMP(3),modify_time TIMESTAMP(3)) WITH ('connector' = 'hudi','path' = 's3://******/hudi_cdc_002/','table.type' = 'COPY_ON_WRITE','write.precombine.field' = 'modify_time','hoodie.datasource.write.recordkey.field' = 'id','write.operation' = 'upsert','write.tasks' = '2','hive_sync.enable' = 'true','hive_sync.metastore.uris' = 'thrift://*******:9083','hive_sync.table' = 'hudi_cdc_002','hive_sync.db' = 'default','hive_sync.mode' = 'HMS','hive_sync.username' = 'hadoop');# 写⼊数据insert into hudi_cdc_002 select * from mysql_cdc_002;