问题需要再AWS的数据湖里面对数据进行CRUD这里就需要使用Apache Iceberg来管理数据。这了我们使用AWS DMS服务来构建CDC将MySQL中的数据搬到s3然后使用Lambda来触发CDC更新数据到AthenaApache Iceberg中使用。一图胜千言修改MySQL参数binlog_format:ROWbinlog_row_image:Full这是MySQL审计日志最全的方式。创建mysql用户# CDC权限CREATEUSERdms_user%IDENTIFIEDBYdms_User349;GRANTREPLICATIONCLIENT,REPLICATIONSLAVEON*.*TOdms_user%;# 业务库和业务表grantselectonmydb.*todms_user;# 使用 MySQL 特定的迁移前评估grantselectonmysql.usertodms_user;grantselectonmysql.dbtodms_user;grantselectonmysql.tables_privtodms_user;grantselectonmysql.role_edgestodms_user;#only for MySQL version 8.0.11 and highergrantselectonperformance_schema.replication_connection_statustodms_user;#Required for primary instance validation - MySQL version 5.7 and higher only# RDS 运行 MySQL 特定的迁移前评估grantselectonmysql.rds_configurationtodms_user;#Required for binary log retention check# 如果参数BatchEnable是必需的true则需要授予权限grantcreatetemporarytableson*.*todms_user;FLUSHPRIVILEGES;Amazon Secrets Manager我这里的mysql是使用了AWS Secrets Manager进行用户名和密码托管轮转的。这里就不重点介绍了我们主要关注AWS DMS数据迁移任务怎么创建。AWS DMS创建源端点或者如下方式创建注意这里连接mysql启用了必须使用SSL方式才能连接需要手动将Aurora MySQL中的证书手动下载添加到DMS中。在如下页面中找到AWS Aurora MySQL中的证书如下页面https://docs.amazonaws.cn/AmazonRDS/latest/AuroraUserGuide/UsingWithRDS.SSL.html#UsingWithRDS.SSL.RegionCertificates-BJS运行如下命令验证一下下载的SSL证书keytool-printcert-v-fileglobal-bundle.pem我们在AWS DMS证书管理页面导入上面MySQL的证书给DMS如下图证书配置成功后下面重新创建源端点创建目标端点s3创建内网端点这里的路由表你如果不知道怎么设置或者后续无法写数据到s3的话你可以一股脑全部把路由表都够选上也可以。Athena的vpc 端点创建Athena vpc端点安全组创建这里的athena vpc 端点的安全组中的出入站规则都是指定了dms的安全组。下面就可以开始创建athena的vpc内网端点了如下图接下来创建s3目标端点注意这里使用到的s3桶名最好以aws-glue-开头方便以后glue中任务读取。如下图参数配置如下{DataFormat:parquet,ParquetVersion:PARQUET_2_0,CompressionType:GZIP,IncludeOpForFullLoad:true,GlueCatalogGeneration:true,TimestampColumnName:last_updated_ts,UseTaskStartTimeForFullLoadTimestamp:true,CdcMaxBatchInterval:3600,CdcMinFileSize:64000,DatePartitionEnabled:true,DatePartitionSequence:YYYYMMDD,DatePartitionDelimiter:SLASH,DatePartitionTimezone:Asia/Shanghai}源和目标端点都创建之后如下图这个目标S3端点会创建一个角色这个角色缺失s3:DeleteObject,权限需要我们自己手动加一下这个权限。还得添加s3解密glue和athena相关权限配置{Version:2012-10-17,Statement:[{Sid:AllObjectActions,Effect:Allow,Action:[s3:PutObject,s3:GetObject,s3:DeleteObject,s3:GetObjectVersion],Resource:[arn:aws-cn:s3:::aws-glue-dev-bronze/*],Condition:{StringEquals:{aws:ResourceAccount:111112234434}}},{Sid:ListBucketActions,Effect:Allow,Action:[s3:ListBucket],Resource:*,Condition:{StringEquals:{aws:ResourceAccount:111112234434}}},{Sid:GetBucketActions,Effect:Allow,Action:[s3:GetBucketLocation,s3:GetBucketVersioning],Resource:[arn:aws-cn:s3:::aws-glue-dev-bronze],Condition:{StringEquals:{aws:ResourceAccount:111112234434}}},{Sid:EnableBucketHTTPSOnly,Action:s3:*,Effect:Deny,Resource:[arn:aws-cn:s3:::aws-glue-dev-bronze/*,arn:aws-cn:s3:::aws-glue-dev-bronze],Condition:{Bool:{aws:SecureTransport:false}}},{Sid:AllowUseOfTheKey,Effect:Allow,Action:[kms:Encrypt,kms:Decrypt,kms:ReEncrypt*,kms:GenerateDataKey*,kms:DescribeKey],Resource:*},{Effect:Allow,Action:[glue:CreateDatabase,glue:GetDatabase,glue:GetDatabases,glue:CreateTable,glue:DeleteTable,glue:UpdateTable,glue:GetTable,glue:GetTables,glue:BatchCreatePartition,glue:CreatePartition,glue:UpdatePartition,glue:GetPartition,glue:GetPartitions,glue:BatchGetPartition],Resource:[arn:aws-cn:glue:*:111112234434:catalog,arn:aws-cn:glue:*:111112234434:database/*,arn:aws-cn:glue:*:111112234434:table/*]},{Effect:Allow,Action:[athena:StartQueryExecution,athena:GetQueryExecution,athena:CreateWorkGroup],Resource:arn:aws-cn:athena:*:111112234434:workgroup/*}]}就可以开始CDC任务创建了。开始CDC创建安全组创建一个安全组给CDC任务使用开始创建复制如下图注意这里的前缀不要使用-字符因为AWS DMS在Athena中创建数据库使用连字符会失败应该改成下划线_需要注意一下。创建完成手动启动如下图启动之前先让AWS云先评估一下如果没问题再真正开始CDC任务如下图看到如下AWS DMS任务表示我们的CDC任务运行成功了到此CDC任务运行成功。到这里只是将mysql里面的binlog数据全部搬到数据湖里面的铜牌层还需要进一步清理数据到银牌层。银牌层创建Silver S3桶Iceberg数据库创建CREATEDATABASEdev_silver_db;直接在glue创建银牌层桶。创建Iceberg表根据需要处理的原始表结构来写下面的DDL语句CREATETABLEmy_iceberg(last_updated_tsTIMESTAMP,idBIGINT,name STRING)LOCATIONs3://aws-glue-dev-silver/dbname/my_iceberg/TBLPROPERTIES(table_typeICEBERG,formatparquet)AWS Glue合并任务AWSGlueServiceRoleDev角色创建这个角色必须以AWSGlueServiceRole开头不然AWS Glue任务无法选择这个角色。创建Glue任务如下其中配置参数为--datalake-formats iceberg--confspark.sql.extensionsorg.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions--confspark.sql.catalog.glue_catalog.warehouses3://aws-glue-dev-silver/dbname/my_iceberg/--confspark.sql.catalog.glue_catalogorg.apache.iceberg.spark.SparkCatalog--confspark.sql.catalog.glue_catalog.catalog-implorg.apache.iceberg.aws.glue.GlueCatalog--confspark.sql.catalog.glue_catalog.io-implorg.apache.iceberg.aws.s3.S3FileIO--confspark.sql.catalog.glue_catalog.glue.lakeformation-enabledtrue--confspark.sql.iceberg.handle-timestamp-without-timezonetrue任务源代码如下importsysfromawsglue.transformsimport*fromawsglue.utilsimportgetResolvedOptionsfrompyspark.contextimportSparkContextfromawsglue.contextimportGlueContextfromawsglue.jobimportJobfrompyspark.sql.typesimportTimestampType## params: [JOB_NAME]argsgetResolvedOptions(sys.argv,[JOB_NAME])scSparkContext()glueContextGlueContext(sc)sparkglueContext.spark_session jobJob(glueContext)job.init(args[JOB_NAME],args)# # 第一步从 Bronze 表读取增量数据# # 注意transformation_ctx 对 Glue 书签机制至关重要请保持唯一且不变[reference:0]myIcebergIncrementalInputDFglueContext.create_dynamic_frame.from_catalog(databasedev_bronze,# 修改为你的 Glue 数据库名table_namemy_iceberg,# 修改为你的 Bronze 表名transformation_ctxmyIcebergIncrementalInputDF).toDF()# # 第二步去重 - 按 id 分组保留 last_updated_ts 最新的记录[reference:1]# windowSpecWindow.partitionBy(myIcebergIncrementalInputDF.id).orderBy(col(last_updated_ts).desc())rankedDFmyIcebergIncrementalInputDF.withColumn(row_num,row_number().over(windowSpec))deduplicatedDFrankedDF.filter(col(row_num)1).drop(row_num)# # 第三步清洗 - 数据类型转换、字段拆分等[reference:2]# deduplicatedDFdeduplicatedDF.withColumn(last_updated_ts,col(last_updated_ts).cast(TimestampType()))# 注册为临时视图供 MERGE 语句使用[reference:3]deduplicatedDF.createOrReplaceTempView(deduplicated_view)# # 第四步执行 MERGE 到 Silver 层 Iceberg 表[reference:4]# merge_sql MERGE INTO glue_catalog.dev_silver_db.my_iceberg AS target USING deduplicated_view AS source ON target.id source.id WHEN MATCHED AND source.op U THEN UPDATE SET target.last_updated_ts source.last_updated_ts, target.name source.name WHEN MATCHED AND source.op D THEN DELETE WHEN NOT MATCHED THEN INSERT ( last_updated_ts, id, name ) VALUES ( source.last_updated_ts, source.id, source.name ) spark.sql(merge_sql)job.commit()lambda触发函数创建一个AWS Lambda函数源代码如下importjsonimportboto3deflambda_handler(event,context):glue_clientboto3.client(glue)glue_job_namedev-b2s-xxxxtry:# Start the Glue jobresponseglue_client.start_job_run(JobNameglue_job_name)print(fGlue job started:{response[JobRunId]})exceptExceptionase:print(fError starting Glue job:{str(e)})raiseereturn{statusCode:200,body:json.dumps(Hello from Lambda!)}注意这里需要给aws lambda的角色添加启动glue任务的权限类似如下{Version:2012-10-17,Statement:[{Effect:Allow,Action:[glue:StartJobRun],Resource:arn:aws-cn:glue:cn-north-1:111112234434:job/*}]}然后我们回到bronze层的s3桶创建一个当文件创建的事件来激活调用这个lambda函数如下图总结没有搞过大数据的朋友需要注意一下如下两个概念Iceberg是表格式Parquet是文件格式这就是AWS里面的将mysql里面数据搬到数据湖的方式。AWS DMSAWS LambdaAWS Glue。这里主要就是借助iceberg表来进行增量更新。参考How to Merge your Relational Database into a Data Lake on AWSReal-Time CDC Pipeline: MySQL to Apache Iceberg Using AWS DMS Serverless, S3, and Lambda.Step 5: Configure an Amazon DMS Target EndpointUsing Amazon S3 as a target for AWS Database Migration ServiceHow to MERGE your Database into a Data Lake on AWS | Change Data Capture | Apache IcebergApache Iceberg vs Parquet – File Formats vs Table FormatsChange data capture使用与 MySQL 兼容的数据库作为源 Amazon DMSCREATE TABLE ASHow to MERGE your Database into a Data Lake on AWS | Change Data Capture | Apache IcebergAWS Aurora MySQL证书地址为 MySQL-compatible、PostgreSQL 或 SQL Server 端点启用 SSLHow can I use AWS DMS to migrate data to Amazon S3 in Parquet format?Athena JDBC 3.x 驱动程序Create an AWS Glue Data Catalog with AWS DMSConfiguring VPC endpoints for AWS DMSMigrating MSSQL to AWS using Data Migration Service with CDC Migration在 Amazon Glue 中使用 Iceberg 框架
