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

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