使用Apache Hudi + Amazon EMR进行变化数据捕获(CDC)

前一篇文章中我们讨论了如何使用Amazon数据库迁移服务（DMS）无缝地收集CDC数据。

https://towardsdatascience.com/data-lake-change-data-capture-cdc-using-amazon-database-migration-service-part-1-capture-b43c3422aad4

下面将演示如何处理CDC数据，以便在数据湖中近实时表示数据库的变更，我们将使用Apache Hudi和Amazon EMR来完成此操作。Apachehudi是一个开源的数据管理框架，用于简化近实时的增量数据处理。

创建一个新的EMR集群

$ aws emr create-cluster --auto-scaling-role EMR_AutoScaling_DefaultRole --applications Name=Spark Name=Hive --ebs-root-volume-size 10 --ec2-attributes '{"KeyName":"roopikadf","InstanceProfile":"EMR_EC2_DefaultRole","SubnetId":"subnet-097e5d6e","EmrManagedSlaveSecurityGroup":"sg-088d03d676ac73013","EmrManagedMasterSecurityGroup":"sg-062368f478fb07c11"}' --service-role EMR_DefaultRole --release-label emr-6.0.0 --name 'Training' --instance-groups '[{"InstanceCount":3,"EbsConfiguration":{"EbsBlockDeviceConfigs":[{"VolumeSpecification":{"SizeInGB":32,"VolumeType":"gp2"},"VolumesPerInstance":2}]},"InstanceGroupType":"CORE","InstanceType":"m5.xlarge","Name":"Core - 2"},{"InstanceCount":1,"EbsConfiguration":{"EbsBlockDeviceConfigs":[{"VolumeSpecification":{"SizeInGB":32,"VolumeType":"gp2"},"VolumesPerInstance":2}]},"InstanceGroupType":"MASTER","InstanceType":"m5.xlarge","Name":"Master - 1"}]' --scale-down-behavior TERMINATE_AT_TASK_COMPLETION --region us-east-1 --bootstrap-actions Path=s3://aws-analytics-course/job/energy/emr.sh,Name=InstallPythonLibs

在创建EMR集群之后，使用SSH登录到主节点，并运行以下命令，这些命令将把Apache Hudi Jar文件复制到S3。

$ aws s3 cp /usr/lib/hudi/hudi-spark-bundle.jar s3://aws-analytics-course/hudi/jar/   upload: ../../usr/lib/hudi/hudi-spark-bundle.jar to s3://aws-analytics-course/hudi/jar/hudi-spark-bundle.jar
$ aws s3 cp /usr/lib/spark/external/lib/spark-avro.jar s3://aws-analytics-course/hudi/jar/
upload: ../../usr/lib/spark/external/lib/spark-avro.jar to s3://aws-analytics-course/hudi/jar/spark-avro.jar
$ aws s3 ls s3://aws-analytics-course/hudi/jar/
2020-10-21 17:00:41   23214176 hudi-spark-bundle.jar
2020-10-21 17:00:56     101212 spark-avro.jar

接着创建一个新的EMR Notebook，并在以下地址获取上传的NoteBook，上传hudi/hudi.ipynb

$ git clone https://github.com/mkukreja1/blogs.git

使用在上一步上传到S3的Hudi Jar文件创建一个Spark Session

from pyspark.sql import SparkSession
import pyspark
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, array, ArrayType, DateType, DecimalType
from pyspark.sql.functions import *
from pyspark.sql.functions import concat, lit, col
spark = pyspark.sql.SparkSession.builder.appName("Product_Price_Tracking") \
     .config("spark.jars", "s3://aws-analytics-course/hudi/jar/hudi-spark-bundle.jar,s3://aws-analytics-course/hudi/jar/spark-avro.jar") \
     .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
     .config("spark.sql.hive.convertMetastoreParquet", "false") \
     .getOrCreate()

读取CDC文件，我们将从读取全量文件开始

TABLE_NAME = "coal_prod"
S3_RAW_DATA = "s3://aws-analytics-course/raw/dms/fossil/coal_prod/LOAD00000001.csv"
S3_HUDI_DATA = "s3://aws-analytics-course/hudi/data/coal_prod"
coal_prod_schema = StructType([StructField("Mode", StringType()),
                               StructField("Entity", StringType()),
                               StructField("Code", StringType()),
                               StructField("Year", IntegerType()),
                               StructField("Production", DecimalType(10,2)),
                               StructField("Consumption", DecimalType(10,2))
                               ])
df_coal_prod = spark.read.csv(S3_RAW_DATA, header=False, schema=coal_prod_schema)
df_coal_prod.show(5)
+----+-----------+----+----+----------+-----------+
|Mode|     Entity|Code|Year|Production|Consumption|
+----+-----------+----+----+----------+-----------+
|   I|Afghanistan| AFG|1949|      0.04|       0.00|
|   I|Afghanistan| AFG|1950|      0.11|       0.00|
|   I|Afghanistan| AFG|1951|      0.12|       0.00|
|   I|Afghanistan| AFG|1952|      0.14|       0.00|
|   I|Afghanistan| AFG|1953|      0.13|       0.00|
+----+-----------+----+----+----------+-----------+
only showing top 5 rows

Apache Hudi需要一个主键来单独标识每个记录。如果是顺序生成的主键最佳，但是我们的表没有主键，为了解决这个问题，我们可以使用Entity和Year列的组合来生成PK，下面的键列将用作主键。

df_coal_prod=df_coal_prod.select("*", concat(col("Entity"),lit(""),col("Year")).alias("key"))
df_coal_prod_f=df_coal_prod.drop(df_coal_prod.Mode)
df_coal_prod_f.show(5)
+-----------+----+----+----------+-----------+---------------+
|     Entity|Code|Year|Production|Consumption|            key|
+-----------+----+----+----------+-----------+---------------+
|Afghanistan| AFG|1949|      0.04|       0.00|Afghanistan1949|
|Afghanistan| AFG|1950|      0.11|       0.00|Afghanistan1950|
|Afghanistan| AFG|1951|      0.12|       0.00|Afghanistan1951|
|Afghanistan| AFG|1952|      0.14|       0.00|Afghanistan1952|
|Afghanistan| AFG|1953|      0.13|       0.00|Afghanistan1953|
+-----------+----+----+----------+-----------+---------------+
only showing top 5 rows

我们现在准备以Hudi格式保存数据，因为这是我们第一次保存这个表，所以我们将使用bulk_insert操作和mode=overwrite。还要注意，我们使用"key"列作为recordkey。

df_coal_prod_f.write.format("org.apache.hudi") \
            .option("hoodie.table.name", TABLE_NAME) \
            .option("hoodie.datasource.write.storage.type", "COPY_ON_WRITE") \
            .option("hoodie.datasource.write.operation", "bulk_insert") \
            .option("hoodie.datasource.write.recordkey.field","key") \
            .option("hoodie.datasource.write.precombine.field", "key") \
            .mode("overwrite") \
            .save(S3_HUDI_DATA)

我们现在可以读取新创建的Hudi表。

df_final = spark.read.format("org.apache.hudi")\
          .load("s3://aws-analytics-course/hudi/data/coal_prod/default/*.parquet")
df_final.registerTempTable("coal_prod")
spark.sql("select count(*) from coal_prod").show(5)
spark.sql("select * from coal_prod where key='India2013'").show(5)
+--------+
|count(1)|
+--------+
|    6282|
+--------+

+-------------------+--------------------+------------------+----------------------+--------------------+------+----+----+----------+-----------+---------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name|Entity|Code|Year|Production|Consumption|      key|
+-------------------+--------------------+------------------+----------------------+--------------------+------+----+----+----------+-----------+---------+
|     20201021215857|20201021215857_54...|         India2013|               default|8fae00ae-34e7-45e...| India| IND|2013|   2841.01|       0.00|India2013|
+-------------------+--------------------+------------------+----------------------+--------------------+------+----+----+----------+-----------+---------+

请注意，对于键India2013，我们有6282行初始化的数据，此键将在下一个操作中更新。现在我们将读取增量数据。

增量数据有4行：插入2行，更新一行，删除一行。我们将首先处理插入和更新的行，注意下面的过滤器（"Mode IN('U'，'I')")。

S3_INCR_RAW_DATA = "s3://aws-analytics-course/raw/dms/fossil/coal_prod/20200808-*.csv"
df_coal_prod_incr = spark.read.csv(S3_INCR_RAW_DATA, header=False, schema=coal_prod_schema)
df_coal_prod_incr_u_i=df_coal_prod_incr.filter("Mode IN ('U', 'I')")
df_coal_prod_incr_u_i=df_coal_prod_incr_u_i.select("*", concat(col("Entity"),lit(""),col("Year")).alias("key"))
df_coal_prod_incr_u_i.show(5)
df_coal_prod_incr_u_i_f=df_coal_prod_incr_u_i.drop(df_coal_prod_incr_u_i.Mode)
df_coal_prod_incr_u_i_f.show()
+----+------+----+----+----------+-----------+---------+
|Mode|Entity|Code|Year|Production|Consumption|      key|
+----+------+----+----+----------+-----------+---------+
|   I| India| IND|2015|   4056.33|       0.00|India2015|
|   I| India| IND|2016|   4890.45|       0.00|India2016|
|   U| India| IND|2013|   2845.66|     145.66|India2013|
+----+------+----+----+----------+-----------+---------+

+------+----+----+----------+-----------+---------+
|Entity|Code|Year|Production|Consumption|      key|
+------+----+----+----------+-----------+---------+
| India| IND|2015|   4056.33|       0.00|India2015|
| India| IND|2016|   4890.45|       0.00|India2016|
| India| IND|2013|   2845.66|     145.66|India2013|
+------+----+----+----------+-----------+---------+

现在准备好对增量数据执行Hudi Upsert操作，由于这个表已经存在，我们将使用append选项。

df_coal_prod_incr_u_i_f.write.format("org.apache.hudi") \
            .option("hoodie.table.name", TABLE_NAME) \
            .option("hoodie.datasource.write.storage.type", "COPY_ON_WRITE") \
            .option("hoodie.datasource.write.operation", "upsert") \
            .option("hoodie.upsert.shuffle.parallelism", 20) \
            .option("hoodie.datasource.write.recordkey.field","key") \
            .option("hoodie.datasource.write.precombine.field", "key") \
            .mode("append") \
            .save(S3_HUDI_DATA)

检查底层数据，注意，已经添加了两个新行，因此表计数从6282增加到6284，另外主键India2013的行已经更新了Production & Consumption列。

df_final = spark.read.format("org.apache.hudi")\
          .load("s3://aws-analytics-course/hudi/data/coal_prod/default/*.parquet")
df_final.registerTempTable("coal_prod")
spark.sql("select count(*) from coal_prod").show(5)
spark.sql("select * from coal_prod where key='India2013'").show(5)
+--------+
|count(1)|
+--------+
|    6284|
+--------+

+-------------------+--------------------+------------------+----------------------+--------------------+------+----+----+----------+-----------+---------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name|Entity|Code|Year|Production|Consumption|      key|
+-------------------+--------------------+------------------+----------------------+--------------------+------+----+----+----------+-----------+---------+
|     20201021220359|20201021220359_0_...|         India2013|               default|8fae00ae-34e7-45e...| India| IND|2013|   2845.66|     145.66|India2013|
+-------------------+--------------------+------------------+----------------------+--------------------+------+----+----+----------+-----------+---------+

接着处理删除

df_coal_prod_incr_d=df_coal_prod_incr.filter("Mode IN ('D')")
df_coal_prod_incr_d=df_coal_prod_incr_d.select("*", concat(col("Entity"),lit(""),col("Year")).alias("key"))
df_coal_prod_incr_d_f=df_coal_prod_incr_d.drop(df_coal_prod_incr_u_i.Mode)
df_coal_prod_incr_d_f.show()
+------+----+----+----------+-----------+---------+
|Entity|Code|Year|Production|Consumption|      key|
+------+----+----+----------+-----------+---------+
| India| IND|2010|   2710.54|       0.00|India2010|
+------+----+----+----------+-----------+---------+

我们可以使用Hudi Upsert操作来完成此操作，但需要使用和额外的删除选项hudi.datasource.write.payload.class=org.apache.hudi.EmptyHoodieRecordPayload

df_coal_prod_incr_d_f.write.format("org.apache.hudi") \
            .option("hoodie.table.name", TABLE_NAME) \
            .option("hoodie.datasource.write.storage.type", "COPY_ON_WRITE") \
            .option("hoodie.datasource.write.operation", "upsert") \
            .option("hoodie.upsert.shuffle.parallelism", 20) \
            .option("hoodie.datasource.write.recordkey.field","key") \
            .option("hoodie.datasource.write.precombine.field", "key") \
            .option("hoodie.datasource.write.payload.class", "org.apache.hudi.EmptyHoodieRecordPayload") \
            .mode("append") \
            .save(S3_HUDI_DATA)

检查结果删除结果，由于删除了一行，计数从6284下降到6283，另外对已删除行的查询返回为空。

df_final = spark.read.format("org.apache.hudi")\
          .load("s3://aws-analytics-course/hudi/data/coal_prod/default/*.parquet")
df_final.registerTempTable("coal_prod")
spark.sql("select count(*) from coal_prod").show(5)
spark.sql("select * from coal_prod where key='India2010'").show(5)
+--------+
|count(1)|
+--------+
|    6283|
+--------+

+-------------------+--------------------+------------------+----------------------+-----------------+------+----+----+----------+-----------+---+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|_hoodie_file_name|Entity|Code|Year|Production|Consumption|key|
+-------------------+--------------------+------------------+----------------------+-----------------+------+----+----+----------+-----------+---+
+-------------------+--------------------+------------------+----------------------+-----------------+------+----+----+----------+-----------+---+

推荐阅读