多库多表场景下 CDC 数据实时写入 Redshift 数仓方案

MessageFormat: json-unformatted
IncludePartitionValue: true
PartitionIncludeSchemaTable: true
NoHexPrefix: NoHexPrefix
IncludeNullAndEmpty: true
# endpoint 参数说明看这里 https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-dms-endpoint-kafkasettings.html

# 根据自己 kafka 版本选择
wget https://archive.apache.org/dist/kafka/2.8.2/kafka_2.12-2.8.2.tgz
tar -xvzf /kafka_2.12-2.8.2.tgz && cd kafka_2.12-2.8.2
export bs="xxxxx.kafka.us-east-1.amazonaws.com:9092"
export topic="xxx"
# 从头消费数据
./bin/kafka-console-consumer.sh --bootstrap-server $bs --topic $topic --from-beginning | jq .
# 从头消费数据，key 也输出
./bin/kafka-console-consumer.sh --bootstrap-server $bs --topic $topic --from-beginning --property print.key=true
# cdc 格式样例
https://github.com/yhyyz/kafka-cdc-redshift#mysql-cdc%E6%A0%BC%E5%BC%8F%E6%A0%B7%E4%BE%8B

# 下载依赖的 JAR，上传到 S3
# spark redshift connector，对开源做了点修改，开源默认将 String 类型映射为了 Text，默认长度是 256，调整为了 varchar(65535)。代码地址 https://github.com/yhyyz/emr-spark-redshift。 可以使用如下编译好的 
wget https://dxs9dnjebzm6y.cloudfront.net/tmp/emr-spark-redshift-1.0-SNAPSHOT.jar
# 可以将 offset 通过 listener 提交到 Kafka，方便 Kafka 测监控消费
# glue 3.0
wget https://dxs9dnjebzm6y.cloudfront.net/tmp/spark-sql-kafka-offset-committer-1.0.jar
# glue 4.0
wget https://dxs9dnjebzm6y.cloudfront.net/tmp/spark3.3-sql-kafka-offset-committer-1.0.jar
# clone 代码，cdc_util build 成 whl，方便再在多个环境中使用，直接执行如下命令 build 或者下载 build 好的
# 编译好的
https://dxs9dnjebzm6y.cloudfront.net/tmp/cdc_util_202304251252-1.1-py3-none-any.whl
# 自己编译
git clone https://github.com/yhyyz/kafka-cdc-redshift
python3 -m venv cdc_venv
source ./cdc_venv/bin/activate
python3 setup.py bdist_wheel

# for glue job，请仔细看参数说明
aws_region = us-east-1
s3_endpoint = s3.us-east-1.amazonaws.com
# spark checkpoint 目前作业重启是会自动从 checkpoint 位置消费
checkpoint_location = s3://xxxx/checkpoint/
# 每批次的消费间隔，控制者最小的延迟，建议值 10s~60s 之间，或者配置为 availableNow 表示消费到当下停止（适用定时执行）spark3.3+才支持
checkpoint_interval = 30 seconds
# kafka 地址
kafka_broker = xxxx1.us-east-1.amazonaws.com:9092,xxxx2.us-east-1.amazonaws.com:9092
# 消费 CDC 数据的 Kafka 的 topic，多个以逗号分隔
topic = flink_mysql_cdc
# 启始的消费位置，支持 latest earliest 和指定时间戳，例如 1682682682000。需要注意的是此参数只在第一此启动作业时生效，当作业重启时会从 checkpoint 启动，如果重启时生效，需要删除 checkpoint 目录再启动。Spark 作业如果有 checkpoint 会优先从 checkpoint 启动，如果没有才会从该参数指定的位置启动
startingOffsets = latest
# 并发写的线程数，如果要 50 张表同时写就配置成 50，更多并发所有表的同步延迟就会更短，消耗的资源就会更多，不建议超过 50
thread_max_workers = 30
# 是否启动 Debug 日志，启动 Debug 会影响性能，主要用来排查错误
disable_msg = true
# CDC 的格式，目前支持三个值，FLINK-CDC or DMS-CDC or MSK-DEBEZIUM-CDC
cdc_format = FLINK-CDC
# 每个批次最大从 Kafka 拉取多少条数据，当回溯数据时，防止一个 batch 消费太多数据，资源开的不够，Spark 内存溢出
max_offsets_per_trigger = 1000000
# 消费者组
consumer_group = cdc-redshift-glue-g1
# Redshift 链接信息配置
# 如果用 secret manager 管理链接，填 secret_id 一个就可以
redshift_secret_id =
redshift_host = xxxx.us-east-1.redshift.amazonaws.com
redshift_port = 5439
redshift_username = xxx
redshift_password = xxx
redshift_database = dev
redshift_schema = public
# 数据写 S3 的临时路径
redshift_tmpdir = s3://xxxx/glue-cdc/tmpdir/
# Redshift 关联的 IAM role，需要有 S3 权限
redshift_iam_role = arn:aws:iam::xxxx:role/<your-role-name>

# 同步表的配置，db 和 table 都支持正则表达式适用于分库分表的合并，target_tabel 指定合并后的表名
sync_table_list = [\
{"db": "test_db", "table": "product", "primary_key": "pid"},\
{"db": "test_db", "table": "user", "primary_key": "id"},\
{"db": "test_db", "table": "product_02", "primary_key": "pid"},\
{"db": "test_db", "table": "product_multiple_key_01", "primary_key": "pid,pname"},\
{"db": "cdc_db_02", "table": "sbtest1", "primary_key": "id"},\
{"db": "cdc_db_02", "table": "sbtest2", "primary_key": "id"},\
{"db": "cdc_db_02", "table": "sbtest3", "primary_key": "id"},\
....
{"db": "cdc_db_02", "table": "sbtest100", "primary_key": "id"}
]

# 其它同步方式的配置例子
# save_delete 设置为 true，表示同步原表同时，将delete数据单独写一张表
# only_save_delete 设置为 true，表示只同步 delete 数据，不同步原表数据
sync_table_list = [\
{"db": "test_db", "table": "product", "primary_key": "pid","ignore_ddl":"true","save_delete":"true"},\
{"db": "test_db", "table": "user", "primary_key": "id","only_save_delete":"true"}\
]

# ignore_ddl，忽略 ddl 变更，表需要用户自己创建，创建表名如果不配置，请用源端的表名字创建 Redshift 表
# target_table 配置在 Redshift 创建表名称
sync_table_list = [\
{"db": "test_db", "table": "product", "primary_key": "pid","ignore_ddl":"true"},\
{"db": "test_db", "table": "product", "primary_key": "pid","ignore_ddl":"true","target_table":"t_product"},\
{"db": "test_db", "table": "user", "primary_key": "id"}\
]

--extra-jars s3://xxxx/jars/emr-spark-redshift-1.0-SNAPSHOT.jar,s3://xxxxx/tmp/spark-sql-kafka-offset-committer-1.0.jar
--additional-python-modules  redshift_connector,jproperties,s3://xxxx/tmp/cdc_util-1.1-py3-none-any.whl
--aws_region us-east-1
# 注意这个参数 --conf 直接写后边内容，spark.executor.cores 调成了 8，表示一个 worker 可以同时运行的 task 是 8
# --conf spark.sql.shuffle.partitions=1  --conf spark.default.parallelism=1 设置为 1，这是为了降低并行度，保证当多个线程同时写多张表时，都尽可能有资源执行，设置为 1 时，最终生产的数据文件也是 1 个，如果数据量很大，生产的一个文件可能会比较大，比如 500MB，这样 Redshift copy 花费的时间就会长一些。如果想要加速，就把这两个值调大一些，比如 4，这样就会生产 4 个 125M 的文件，Redshift 并行 copy 就会快一些，但 Glue 作业的资源对应就要设置多一些，可以观察执行速度评估
--conf spark.sql.streaming.streamingQueryListeners=net.heartsavior.spark.KafkaOffsetCommitterListener  --conf spark.executor.cores=8 --conf spark.sql.shuffle.partitions=1  --conf spark.default.parallelism=1 --conf spark.speculation=false --conf spark.dynamicAllocation.enabled=false
--config_s3_path  s3://xxxx/kafka-cdc-redshift/job-4x.properties
# 关闭 spark event log, streaming 模式 event log 增长会将 driver 端的磁盘打满
--enable-spark-ui false
# Glue 选择 3.x，作业类型选择 Spark Streaming 作业，worker 个数根据同步表的数量和大小选择，Number of retries 在 Streaming 作业下可以设置大些，比如 100。失败自动重启，且会从 checkpoint 自动重启 

# S3 存储桶，例如：s3://<my-bucket>
export APP_S3_BUCKET='s3://<my-bucket>'
export AWS_REGION="us-east-1"
# EMR Serveless application id
export EMR_SERVERLESS_APP_ID='00fcvtfb3pisut09'
# execution role 需要有对应的 S3 桶权限
export EMR_SERVERLESS_EXECUTION_ROLE_ARN='arn:aws:iam::xxxxx:role/<your-role-name>'

export APP_NAME='emr-serverless-cdc-redshift'
export APP_LOCAL_HOME="/home/ec2-user/$APP_NAME"

export CDC_UTIL_WHL_NAME="cdc_util_202310201517-1.1-py3-none-any.whl"
export CDC_UTIL_WHL_LOCATION="https://dxs9dnjebzm6y.cloudfront.net/tmp/$CDC_UTIL_WHL_NAME"

export APP_S3_HOME="$APP_S3_BUCKET/$APP_NAME"
export MAIN_SCRIPT_PATH="$APP_S3_HOME/script/cdc_redshift.py"
export JOB_CONFIG_PATH="$APP_S3_HOME/script/job-ec2-canal.properties"
export PYTHON_VENV_PATH="$APP_S3_HOME/venv/cdc_venv.tar.gz"

# python 3.7+
mkdir -p $APP_LOCAL_HOME/venv/
cd $APP_LOCAL_HOME/venv/
deactivate
rm -rf ./cdc_venv
rm -rf ./cdc_venv.tar.gz

python3 -m venv cdc_venv
source cdc_venv/bin/activate
pip3 install --upgrade pip
pip3 install redshift_connector jproperties 
# cdc_util 是封装好的 Spark CDC Redshift 的包，源代码在 cdc_util中
wget $CDC_UTIL_WHL_LOCATION
pip3 install $CDC_UTIL_WHL_NAME

pip3 install venv-pack
venv-pack -f -o cdc_venv.tar.gz
# 上传到 S3
aws s3 cp $APP_LOCAL_HOME/venv/cdc_venv.tar.gz $APP_S3_HOME/venv/

mkdir $APP_LOCAL_HOME/jars/
rm -rf $APP_LOCAL_HOME/jars/*
cd $APP_LOCAL_HOME/
aws s3 rm --recursive $APP_S3_HOME/jars/

wget -P ./jars  https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.12/3.3.2/spark-sql-kafka-0-10_2.12-3.3.2.jar
wget -P ./jars  https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/2.8.2/kafka-clients-2.8.2.jar
wget -P ./jars  https://repo1.maven.org/maven2/org/apache/spark/spark-token-provider-kafka-0-10_2.12/3.3.2/spark-token-provider-kafka-0-10_2.12-3.3.2.jar
wget -P ./jars  https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/2.11.1/commons-pool2-2.11.1.jar

# emr 6.10 之后 emr or emr serverless 默认会在 spark.executor/driver.extraClassPath 中配置/usr/share/aws/redshift/spark-redshift/lib/* 加入自带的 redshift connector。所以下面的 emr-spark-redshift-1.3-SNAPSHOT.jar 在 emr serverless 中并不生效，可以不使用，如果使用自定义的在 emr serverless 中要通过自定义 image 实现。emr on ec2 中，需要在/etc/spark/conf/spark-defaults.conf 中去掉依赖
# wget -P  ./jars https://dxs9dnjebzm6y.cloudfront.net/tmp/emr-spark-redshift-1.3-SNAPSHOT.jar

wget -P  ./jars https://dxs9dnjebzm6y.cloudfront.net/tmp/spark3.3-sql-kafka-offset-committer-1.0.jar

aws s3 sync $APP_LOCAL_HOME/jars ${APP_S3_HOME}/jars/

cd $APP_LOCAL_HOME/
mkdir $APP_LOCAL_HOME/script/
rm -rf $APP_LOCAL_HOME/script/*
wget -P  ./script https://raw.githubusercontent.com/yhyyz/kafka-cdc-redshift/main/emr_ec2/cdc_redshift.py
aws s3 cp ./script/cdc_redshift.py $APP_S3_HOME/script/

wget -P ./script https://raw.githubusercontent.com/yhyyz/kafka-cdc-redshift/main/config/job-ec2-canal.properties
# 注意修改 job-ec2-canal.properties 配置， 这里使用的是 canal cdc 格式，也支持 Flink CDC,Debezium CDC，DMS CDC 格式
aws s3 cp ./script/job-ec2-canal.properties $APP_S3_HOME/script/

cd $APP_LOCAL_HOME/
cat << EOF > $APP_LOCAL_HOME/start-job-run.json
{
    "name":"$APP_NAME",
    "applicationId":"$EMR_SERVERLESS_APP_ID",
    "executionRoleArn":"$EMR_SERVERLESS_EXECUTION_ROLE_ARN",
    "jobDriver":{
        "sparkSubmit":{
        "entryPoint":"$MAIN_SCRIPT_PATH",
        "entryPointArguments":[
        
  
  "$AWS_REGION",
        
  
  "$JOB_CONFIG_PATH"
        ],
         "sparkSubmitParameters":"--conf spark.sql.streaming.streamingQueryListeners=net.heartsavior.spark.KafkaOffsetCommitterListener --conf spark.executor.cores=16 --conf spark.executor.memory=16g --conf spark.driver.cores=8 --conf spark.driver.memory=16g --conf spark.executor.instances=10 --conf spark.sql.shuffle.partitions=2  --conf spark.default.parallelism=2 --conf spark.dynamicAllocation.enabled=false --conf spark.emr-serverless.driver.disk=150G --conf spark.emr-serverless.executor.disk=150G --conf spark.jars=$(aws s3 ls $APP_S3_HOME/jars/ | grep -o '\S*\.jar$'| awk '{print "'"$APP_S3_HOME/jars/"'"$1","}' | tr -d '\n' | sed 's/,$//'),/usr/share/aws/redshift/jdbc/RedshiftJDBC.jar,/usr/share/aws/redshift/spark-redshift/lib/spark-avro.jar --conf spark.archives=${PYTHON_VENV_PATH}#environment --conf spark.emr-serverless.driverEnv.PYSPARK_DRIVER_PYTHON=./environment/bin/python --conf spark.emr-serverless.driverEnv.PYSPARK_PYTHON=./environment/bin/python --conf spark.executorEnv.PYSPARK_PYTHON=./environment/bin/python --conf spark.hadoop.hive.metastore.client.factory.class=com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
        }
   },
   "configurationOverrides":{
        "monitoringConfiguration":{
            "s3MonitoringConfiguration":{
                "logUri":"$APP_S3_HOME/logs"
            }
        }
   }
}
EOF
jq . $APP_LOCAL_HOME/start-job-run.json

export EMR_SERVERLESS_JOB_RUN_ID=$(aws emr-serverless start-job-run \
    --no-paginate --no-cli-pager --output text \
    --region $AWS_REGION \
    --name $APP_NAME \
    --application-id $EMR_SERVERLESS_APP_ID \
    --execution-role-arn $EMR_SERVERLESS_EXECUTION_ROLE_ARN \
    --execution-timeout-minutes 0 \
    --cli-input-json file://$APP_LOCAL_HOME/start-job-run.json \
    --query jobRunId)

now=$(date +%s)sec
while true; do
    jobStatus=$(aws emr-serverless get-job-run \
                    --no-paginate --no-cli-pager --output text \
                    --application-id $EMR_SERVERLESS_APP_ID \
                    --job-run-id $EMR_SERVERLESS_JOB_RUN_ID \
                    --query jobRun.state)
    if [ "$jobStatus" = "PENDING" ] || [ "$jobStatus" = "SCHEDULED" ] || [ "$jobStatus" = "RUNNING" ]; then
        for i in {0..5}; do
            echo -ne "\E[33;5m>>> The job [ $EMR_SERVERLESS_JOB_RUN_ID ] state is [ $jobStatus ], duration [ $(date -u --date now-$now +%H:%M:%S) ] ....\r\E[0m"
            sleep 1
        done
    else
        echo -ne "The job [ $EMR_SERVERLESS_JOB_RUN_ID ] is [ $jobStatus ]\n\n"
        break
    fi
done

JOB_LOG_HOME=$APP_LOCAL_HOME/log/$EMR_SERVERLESS_JOB_RUN_ID
rm -rf $JOB_LOG_HOME && mkdir -p $JOB_LOG_HOME
aws s3 cp --recursive $APP_S3_HOME/logs/applications/$EMR_SERVERLESS_APP_ID/jobs/$EMR_SERVERLESS_JOB_RUN_ID/ $JOB_LOG_HOME >& /dev/null
gzip -d -r -f $JOB_LOG_HOME >& /dev/null
grep --color=always -r -i -E 'error|failed|exception' $JOB_LOG_HOME

# 对于 Full load，ongoing replication 类型作业，使用 API 可以从上次停止的的位点恢复，console 上的 resume 也可以， 但此类型作业不可以从任意指定的 checkpoint 点或者 native LSN 恢复作业。因此对于此类型的作业，如果想要用新的 Task 启动作业，可以新启动一个 cdc 类型作业
# resume-processing 和 console 上 resume 等同。reload-target 和 console 上的 restart 等同。此类型作业 start-replication 只能首次启动作业才可以用
aws dms start-replication-task \
    --replication-task-arn arn:aws:dms:ap-southeast-1:xxx:task:xxx \
    --start-replication-task-type resume-processing \
    
# 指定位点，from native LSN 如何果获取这个位点，在下方查看 mysql binlog 位点说明
aws dms start-replication-task \
    --replication-task-arn arn:aws:dms:ap-southeast-1:xxx:task:xxx \
    --start-replication-task-type start-replication  \
    --cdc-start-position mysql-bin-changelog.038818:568
    
# 指定 checkpoint from checkpoint 
aws dms start-replication-task \
    --replication-task-arn arn:aws:dms:ap-southeast-1:xxx:task:xxx \
    --start-replication-task-type start-replication  \
    --cdc-start-position checkpoint:V1#264#mysql-bin-changelog.039022:2167:-1:2198:167598213826489:mysql-bin-changelog.039022:2052#0#0#*#0#825

# 从上一次失败恢复
aws dms start-replication-task \
    --replication-task-arn arn:aws:dms:ap-southeast-1:xxx:task:xxx \
   --start-replication-task-type resume-processing

# 根据情况，binlog 保留时间，防止 binlog 删除，cdc 恢复数据可能找不到 binlog 位点
call mysql.rds_show_configuration;
call mysql.rds_set_configuration('binlog retention hours', 72);
# cdc 需要 mysql 设置 binlog_format 为 ROW

# 查看 mysql binlog 位点
show master status;
show binary logs;
show binlog events in 'mysql-bin-changelog.xxxx';

# cdc 阶段
BatchApplyEnabled=true 这个默认就是 True
BatchSplitSize=0 0 表示没有限制，设为 0 即可
BatchApplyTimeoutMax=30 等待的最小值，默认是 1s
BatchApplyTimeoutMin=60 等待的最大值，默认是 30s
ParallelApplyThreads=32 默认值是 0，调整为 32，表中有 Lob 和 Text 列，不建议调整，保持默认值
ParallelApplyBufferSize=1000 每个 buffer queue 中最大记录数默认是 100，调整为 1000
BatchApplyMemoryLimit=1500 默认值 500MB，MaxFileSize（默认 32MB）*ParallelApplyThreads=32*32=1024MB，调成 1500
MemoryLimitTotal=6144 默认值 1024MB，事务在内存中超过这个值开始写磁盘
MemoryKeepTime=180 默认值 60s，事务在内存中超过这个时间没有写到 target 开始写磁盘
# full load 阶段
MaxFullLoadSubTasks=8 默认值 8，最大 49。表示全量加载阶段最多同时并行同步多少张表
ParallelLoadThreads=16 表示对于一张表，可以启动多少个线程来同步，增大该值，对大表的全量数据同步会更快，划分表中数据的方式可通过指定范围字段或者分区方式，具体可以参考 https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Tablesettings.html#CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Tablesettings.ParallelLoad