如何构建智能湖仓架构？亚马逊工程师的代码实践

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"
aws emr create-cluster \--termination-protected \--region ap-southeast-1 \--applications Name=Hadoop Name=Hive Name=Flink Name=Tez Name=SparkName=JupyterEnterpriseGateway Name=Presto Name=HCatalog \--scale-down-behavior TERMINATE_AT_TASK_COMPLETION \--release-label emr-6.4.0 \--ebs-root-volume-size 50 \--service-role EMR_DefaultRole \--enable-debugging \--instance-groups InstanceGroupType=MASTER,InstanceCount=1,InstanceType=m5.xlargeInstanceGroupType=CORE,InstanceCount=2,InstanceType=m5.xlarge \--managed-scaling-policyComputeLimits='{MinimumCapacityUnits=2,MaximumCapacityUnits=5,MaximumOnDemandCapacityUnits=2,MaximumCoreCapacityUnits=2,UnitType=Instances}' \--name "${cluster_name}" \--log-uri "${log_uri}" \--ec2-attributes '{"KeyName":"'${key_name}'","SubnetId":"subnet-0f79e4471cfa74ced","InstanceProfile":"EMR_EC2_DefaultRole"}' \--configurations '[{"Classification": "hive-site","Properties":{"javax.jdo.option.ConnectionURL": "'${jdbc}'","javax.jdo.option.ConnectionDriverName":"org.mariadb.jdbc.Driver","javax.jdo.option.ConnectionUserName":"admin","javax.jdo.option.ConnectionPassword": "xxxxxx"}}]'

# 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;

# sysbench 写⼊mysql数据# 下载sysbenchcurl -s https://packagecloud.io/install/repositories/akopytov/sysbench/script.rpm.sh | sudobashsudo yum -y install sysbench# 注意当前使用的“lua”并未提供构建，请根据自身情况定义,上述⽤到表结构如下CREATE TABLE if not exists `test_tb_01` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(155) DEFAULT NULL,`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,`modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATECURRENT_TIMESTAMP,PRIMARY KEY (`id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;# 创建表sysbench creates.lua --mysql-user=admin --mysql-password=admin123456 --mysql?host=****.rds.amazonaws.com --mysql-db=cdc_test_db --report-interval=1 --events=1 run# 插⼊数据sysbench insert.lua --mysql-user=admin --mysql-password=admin123456 --mysql?host=****.rds.amazonaws.com --mysql-db=cdc_test_db --report-interval=1 --events=500 --time=0 --threads=1 --skip_trx=true run# 更新数据sysbench update.lua --mysql-user=admin --mysql-password=admin123456 --mysql?host=****.rds.amazonaws.com --mysql-db=cdc_test_db --report-interval=1 --events=1000 --time=0 --threads=10 --skip_trx=true --update_id_min=3 --update_id_max=500 run# 删除表sysbench drop.lua --mysql-user=admin --mysql-password=admin123456 --mysql?host=****.rds.amazonaws.com --mysql-db=cdc_test_db --report-interval=1 --events=1 run