Apache Hudi 入门学习总结

jar=$1

sudo rm -r tez_temp
mkdir tez_temp
cd tez_temp
hadoop fs -get /hdp/apps/3.1.0.0-78/tez/tez.tar.gz
mkdir tez
tar -zxvf tez.tar.gz -C tez
mkdir gz
sudo rm -r tez/lib/hudi-hadoop-mr*
cp $jar tez/lib/
cd tez
tar -zcvf  ../gz/tez.tar.gz ./*
hadoop fs -rm -r  /hdp/apps/3.1.0.0-78/tez/tez.tar.gz.back
hadoop fs -mv  /hdp/apps/3.1.0.0-78/tez/tez.tar.gz /hdp/apps/3.1.0.0-78/tez/tez.tar.gz.back
cd ../gz/
hadoop fs -put tez.tar.gz /hdp/apps/3.1.0.0-78/tez/
su - hdfs <<EOF
kinit -kt /etc/security/keytabs/hdfs.headless.keytab hdfs-cluster1@INDATA.COM
hadoop fs -chown hdfs:haoop /hdp/apps/3.1.0.0-78/tez/tez.tar.gz
hadoop fs -chmod 444  /hdp/apps/3.1.0.0-78/tez/tez.tar.gz
hadoop fs -ls /hdp/apps/3.1.0.0-78/tez/
exit
EOF

import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.config.HoodieWriteConfig.TBL_NAME
import org.apache.hudi.hive.MultiPartKeysValueExtractor
import org.apache.hudi.keygen.ComplexKeyGenerator
import org.apache.spark.sql.SaveMode.{Append, Overwrite}
import org.apache.spark.sql.hudi.command.UuidKeyGenerator
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}


 val spark = SparkSession.builder().
      master("local[*]").
      appName("SparkHudiDemo").
      config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").
      // 扩展Spark SQL，使Spark SQL支持Hudi
      config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension").
      // 支持Hive，本地测试时，注释掉
      //      enableHiveSupport().
      getOrCreate()

    val spark = SparkSession.builder().
      master("local[*]").
      appName("SparkHudiDemo").
      config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").
      // 扩展Spark SQL，使Spark SQL支持Hudi
      config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension").
      // 支持Hive，本地测试时，注释掉
      //      enableHiveSupport().
      getOrCreate()

    import spark.implicits._
    val df = Seq((1, "a1", 10, 1000, "2022-05-12")).toDF("id", "name", "value", "ts", "dt")

    val databaseName = "default"
    val tableName1 = "test_hudi_table_1"
    val primaryKey = "id"
    val preCombineField = "ts"
    val partitionField = "dt"
    val tablePath1 = "/tmp/test_hudi_table_1"
    save2HudiSyncHiveWithPrimaryKey(df, databaseName, tableName1, primaryKey, preCombineField, partitionField,
      UPSERT_OPERATION_OPT_VAL, tablePath1, Overwrite)
    spark.read.format("hudi").load(tablePath1).show(false)
    // 删除表
    save2HudiSyncHiveWithPrimaryKey(df, databaseName, tableName1, primaryKey, preCombineField, partitionField,
      DELETE_OPERATION_OPT_VAL, tablePath1, Append)
    spark.read.format("hudi").load(tablePath1).show(false)


  /**
   * 写hudi并同步到hive，有主键，分区字段dt
   *
   */
  def save2HudiSyncHiveWithPrimaryKey(df: DataFrame, databaseName: String, tableName: String, primaryKey: String, preCombineField: String,
                                      partitionField: String, operation: String, tablePath: String, mode: SaveMode): Unit = {
    df.
      write.format("hudi").
      option(RECORDKEY_FIELD.key, primaryKey). // 主键字段
      option(PRECOMBINE_FIELD.key, preCombineField). // 预合并字段
      option(PARTITIONPATH_FIELD.key, partitionField).
      option(TBL_NAME.key, tableName).
      option(KEYGENERATOR_CLASS_NAME.key(), classOf[ComplexKeyGenerator].getName).
      option(OPERATION.key(), operation).
      // 下面的参数和同步hive元数据，查询hive有关
      option(META_SYNC_ENABLED.key, true).
      option(HIVE_USE_JDBC.key, false).
      option(HIVE_DATABASE.key, databaseName).
      option(HIVE_AUTO_CREATE_DATABASE.key, true).
      // 内部表，这里非必须，但是在用saveAsTable时则必须，因为0.9.0有bug，默认外部表
      option(HIVE_CREATE_MANAGED_TABLE.key, true).
      option(HIVE_TABLE.key, tableName).
      option(HIVE_CREATE_MANAGED_TABLE.key, true).
      option(HIVE_STYLE_PARTITIONING.key, true).
      option(HIVE_PARTITION_FIELDS.key, partitionField).
      option(HIVE_PARTITION_EXTRACTOR_CLASS.key, classOf[MultiPartKeysValueExtractor].getName).
      // 为了SparkSQL更新用，0.9.0版本有bug，需要设置这个参数，最新版本已经修复，可以不设置这个参数
      // 详情查看PR：https://github.com/apache/hudi/pull/3745
      option(DataSourceWriteOptions.HIVE_TABLE_SERDE_PROPERTIES.key, s"primaryKey=$primaryKey").
      mode(mode)
      .save(tablePath)
  }

spark.read.format("hudi").load(tablePath1).show(false)

+-------------------+--------------------+------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|_hoodie_file_name                                                       |id |name|value|ts  |dt        |
+-------------------+--------------------+------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+
|20220512101542     |20220512101542_0_1  |id:1              |2022-05-12            |38c1ff87-8bc9-404c-8d2c-84f720e8133c-0_0-20-12004_20220512101542.parquet|1  |a1  |10   |1000|2022-05-12|
+-------------------+--------------------+------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+

option(KEYGENERATOR_CLASS_NAME.key(), classOf[ComplexKeyGenerator].getCanonicalName).

show create table test_hudi_table_1;

+----------------------------------------------------+
|                   createtab_stmt                   |
+----------------------------------------------------+
| CREATE TABLE `test_hudi_table_1`(                  |
|   `_hoodie_commit_time` string,                    |
|   `_hoodie_commit_seqno` string,                   |
|   `_hoodie_record_key` string,                     |
|   `_hoodie_partition_path` string,                 |
|   `_hoodie_file_name` string,                      |
|   `id` int,                                        |
|   `name` string,                                   |
|   `value` int,                                     |
|   `ts` int)                                        |
| PARTITIONED BY (                                   |
|   `dt` string)                                     |
| ROW FORMAT SERDE                                   |
|   'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'  |
| WITH SERDEPROPERTIES (                             |
|   'hoodie.query.as.ro.table'='false',              |
|   'path'='/tmp/test_hudi_table_1',                  |
|   'primaryKey'='id')                               |
| STORED AS INPUTFORMAT                              |
|   'org.apache.hudi.hadoop.HoodieParquetInputFormat'  |
| OUTPUTFORMAT                                       |
|   'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' |
| LOCATION                                           |
|   'hdfs://cluster1/tmp/test_hudi_table_1'           |
| TBLPROPERTIES (                                    |
|   'last_commit_time_sync'='20220512101500',        |
|   'spark.sql.sources.provider'='hudi',             |
|   'spark.sql.sources.schema.numPartCols'='1',      |
|   'spark.sql.sources.schema.numParts'='1',         |
|   'spark.sql.sources.schema.part.0'='{"type":"struct","fields":[{"name":"_hoodie_commit_time","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_commit_seqno","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_record_key","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_partition_path","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_file_name","type":"string","nullable":true,"metadata":{}},{"name":"id","type":"integer","nullable":false,"metadata":{}},{"name":"name","type":"string","nullable":true,"metadata":{}},{"name":"value","type":"integer","nullable":false,"metadata":{}},{"name":"ts","type":"integer","nullable":false,"metadata":{}},{"name":"dt","type":"string","nullable":true,"metadata":{}}]}',  |
|   'spark.sql.sources.schema.partCol.0'='dt',       |
|   'transient_lastDdlTime'='1652320902')            |
+----------------------------------------------------+

select * from test_hudi_table_1;

+----------------------+-----------------------+---------------------+-------------------------+---------------------------------------------------------------------------+-----+-------+--------+-------+-------------+--+
| _hoodie_commit_time  | _hoodie_commit_seqno  | _hoodie_record_key  | _hoodie_partition_path  |                             _hoodie_file_name                             | id  | name  | value  |  ts   |     dt      |
+----------------------+-----------------------+---------------------+-------------------------+---------------------------------------------------------------------------+-----+-------+--------+-------+-------------+--+
| 20220513150854       | 20220513150854_0_1    | id:1                | dt=2022-05-12           | dd4ef080-97b6-4046-a337-abb01e26943e-0_0-21-12005_20220513150854.parquet  | 1   | a1    | 10     | 1000  | 2022-05-12  |
+----------------------+-----------------------+---------------------+-------------------------+---------------------------------------------------------------------------+-----+-------+--------+-------+-------------+--+

class SqlKeyGenerator(props: TypedProperties) extends ComplexKeyGenerator(props)

hoodie.table.precombine.field=ts
hoodie.table.partition.fields=dt
hoodie.table.type=COPY_ON_WRITE
hoodie.archivelog.folder=archived
hoodie.populate.meta.fields=true
hoodie.timeline.layout.version=1
hoodie.table.version=2
hoodie.table.recordkey.fields=id
hoodie.table.base.file.format=PARQUET
hoodie.table.keygenerator.class=org.apache.hudi.keygen.ComplexKeyGenerator
hoodie.table.name=test_hudi_table_1

    val tableName2 = "test_hudi_table_2"
    val tablePath2 = "/tmp/test_hudi_table_2"
    save2HudiWithNoPrimaryKey(df, tableName2, tablePath2)
    spark.read.format("hudi").load(tablePath2).show(false)

  /**
   * 非主键表，非分区表
   */
  def save2HudiWithNoPrimaryKey(df: DataFrame, tableName: String, tablePath: String): Unit = {
    df.
      write.format("hudi").
      option(KEYGENERATOR_CLASS_NAME.key, classOf[UuidKeyGenerator].getName).
      option(RECORDKEY_FIELD.key, "").
      option(PARTITIONPATH_FIELD.key, "").
      option(TBL_NAME.key, tableName).
      option(OPERATION.key(), INSERT_OPERATION_OPT_VAL).
      mode(Overwrite).
      save(tablePath)
  }

+-------------------+--------------------+------------------------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key                  |_hoodie_partition_path|_hoodie_file_name                                                       |id |name|value|ts  |dt        |
+-------------------+--------------------+------------------------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+
|20220512145525     |20220512145525_0_1  |7263eac1-51f6-42eb-834d-bb5dfe13708e|                      |4fe619f1-58b1-4f58-94e6-002f9f5f5155-0_0-20-12004_20220512145525.parquet|1  |a1  |10   |1000|2022-05-12|
+-------------------+--------------------+------------------------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+

hoodie.merge.allow.duplicate.on.inserts = true

    val tableName3 = "test_hudi_table_3"
    save2HudiWithSaveAsTable(df, databaseName, tableName3, primaryKey)
    spark.table(tableName3).show()

  def save2HudiWithSaveAsTable(df: DataFrame, databaseName: String, tableName: String, primaryKey: String): Unit = {
    df.
      write.format("hudi").
      option(RECORDKEY_FIELD.key(), primaryKey).
      // 不需要预合并，所以设置为primaryKey
      // 当insert/bulk_insert等操作，并且关闭了相关参数,则不需要设置
      // SparkSQL中如果没有显示配置预合并字段，则默认将预合并字段设置为schema的最后一个字段
      // 如果为默认值的话，则可能会报null异常，所以设置为主键
      // `PRECOMBINE_FIELD.key -> tableSchema.fields.last.name`
      // 相关issue：https://github.com/apache/hudi/issues/4131
      option(PRECOMBINE_FIELD.key(), primaryKey).
      option(DataSourceWriteOptions.HIVE_TABLE_SERDE_PROPERTIES.key, s"primaryKey=$primaryKey").
      option(TBL_NAME.key(), tableName).
      option(HIVE_CREATE_MANAGED_TABLE.key, true).
      // 关闭预合并，虽然默认值为false，但是0.9.0版本SparkSQL，当有主键时，设置为了true
      option(HoodieWriteConfig.COMBINE_BEFORE_INSERT.key, false).
      // 使用bulk_insert
      option(DataSourceWriteOptions.SQL_ENABLE_BULK_INSERT.key, true).
      // 这里虽然为Overwrite，但是Hudi CTAS要求目录必须为空，否则会报验证错误
      mode(Overwrite).
      saveAsTable(s"$databaseName.$tableName")
  }

+-------------------+--------------------+------------------+----------------------+--------------------+---+----+-----+----+----------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name| id|name|value|  ts|        dt|
+-------------------+--------------------+------------------+----------------------+--------------------+---+----+-----+----+----------+
|     20220512154039|  20220512154039_0_1|              id:1|                      |de3c99a2-3858-462...|  1|  a1|   10|1000|2022-05-12|
+-------------------+--------------------+------------------+----------------------+--------------------+---+----+-----+----+----------+