揭秘数据湖——长文详解Hudi从内核到实战(三)

发布时间:2021年09月07日作者:atguigu浏览次数:902

Hudi实战

Hudi名称概念

  • Time Line

Hudi的核心是维护不同时间对表执行的所有操作的事件表,这有助于提供表的即时视图,同时还有效地支持按到达顺序进行数据检索。Hudi包含以下组件:

(1)Instant action:在表上的操作类型

(2)Instant time:操作开始的一个时间戳,该时间戳会按照开始时间顺序单调递增

(3)state:即时状态

Hudi保证在时间轴上执行的操作都是原先性的,所有执行的操作包括:

(1)commits:原子的写入一张表的操作

(2)cleans:后台消除了表中的旧版本数据,即表中不再需要的数据

(3)delta_commit:增量提交,将一批数据原子写入到MergeOnRead表中,并且只记录到增量日志中

(4)compaction:后台协调Hudi中的差异数据

(5)rollback:回滚,删除在写入过程中的数据

(6)savepoint:将某些文件标记“已保存”,以便清理数据时不会删除它们,一般用于表的还原,可以将数据还原到某个时间点

任何操作都可以处于以下状态:

(1)Requested:表示已安排操作行为,但是尚未开始

(2)Inflight:表示正在执行当前操作

(3)Completed:表示已完成操作

  • File Managerment

Hudi将表组织成DFS上基本路径下的目录结构。表分为几个分区,与hive类似,每个分区均有唯一标示。

在每个分区内,有多个数据组,每个数据组包含几个文件片,其中文件片包含基本文件和日志文件。Hudi采用MVCC设计,其中压缩操作将日志文件和基本数据文件合并成新的文件片,而清除操作则将未使用的文件片去除。

  • 索引

Hudi通过使用索引机制,生成hoodie密钥映射对应文件ID,从而提供高效upsert操作。

  • 表类型

1. Copy on Write:仅使用列式存储,例如parquet。仅更新版本号,通过写入过程中执行同步合并来重写文件。

2. Merge on Read:基于列式存储(parquet)和行式存储(arvo)结合的文件更始进行存储。更新记录到增量文件,压缩同步和异步生成新版本的文件。

以下是对比:

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

  • 查询类型

快照查询(Snapshot Queries):查询操作将查询最新快照的表数据。如果是Merge on Read类型的表,它将动态合并最新文件版本的基本数据和增量数据用于显示查询。如果是Copy On Write类型的表,它直接查询parquet表,同时提供upsert/delete操作。

增量查询(Incremental Queries):查询只能看到写入表的新数据。这有效地提供了changestreams来启用增量数据管道。

优化读查询(Read Optimized Queries):查询将查看给定提交/压缩操作表的最新快照。

以下是对比:

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

Hudi操作

  • pom.xml
<dependencies>
    <dependency>
       <groupId>org.apache.hudi</groupId>
       <artifactId>hudi-client</artifactId>
       <version>0.5.3</version>
    </dependency>
    <dependency>
       <groupId>org.apache.hudi</groupId>
       <artifactId>hudi-hive</artifactId>
       <version>0.5.3</version>
    </dependency>


    <dependency>
       <groupId>org.apache.hudi</groupId>
       <artifactId>hudi-spark-bundle_2.11</artifactId>
       <version>0.5.3</version>
    </dependency>
    <dependency>
       <groupId>org.apache.hudi</groupId>
       <artifactId>hudi-common</artifactId>
       <version>0.5.3</version>
    </dependency>
    <dependency>
       <groupId>org.apache.hudi</groupId>
       <artifactId>hudi-hadoop-mr-bundle</artifactId>
       <version>0.5.3</version>
    </dependency>
    <!-- Spark的依赖引入 -->
    <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-core_2.11</artifactId>
        <!--           <scope>provided</scope>-->
       <version>2.4.5</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
       <artifactId>spark-sql_2.11</artifactId>
        <!--           <scope>provided</scope>-->
       <version>2.4.5</version>
    </dependency>
    <dependency>
       <groupId>org.apache.spark</groupId>
        <artifactId>spark-hive_2.11</artifactId>
        <!--           <scope>provided</scope>-->
       <version>2.4.5</version>
    </dependency>
    <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-avro_2.11</artifactId>
        <version>2.4.5</version>
    </dependency>
    <dependency>
       <groupId>org.scala-lang</groupId>
       <artifactId>scala-library</artifactId>
        <!--           <scope>provided</scope>-->
       <version>${scala.version}</version>
    </dependency>
    <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client</artifactId>
       <version>2.7.2</version>
    </dependency>
    <dependency>
       <groupId>com.alibaba</groupId>
       <artifactId>fastjson</artifactId>
        <version>1.2.47</version>
    </dependency>
    <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-hive_2.11</artifactId>
        <!--           <scope>provided</scope>-->
       <version>2.4.5</version>
    </dependency>
    <dependency>
       <groupId>org.spark-project.hive</groupId>
       <artifactId>hive-jdbc</artifactId>
       <version>1.2.1.spark2</version>
    </dependency>
</dependencies>
  • case class
packagecom.atguigu.bean
case class DwsMember(
                      uid: Int,
                      ad_id: Int,
                      var fullname:String,
                      iconurl: String,
                      dt: String,
                      dn: String
                    )
  • 配置文件

将集群配置文件复制到,项目resource源码包下,使本地环境可以访问hadoop集群。

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

  • Hudi写入Hdfs
packagecom.atguigu.hudi.test


import com.atguigu.bean.DwsMember
import com.atguigu.hudi.util.ParseJsonData
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}


object HoodieDataSourceExample {
  def main(args: Array[String]): Unit = {
   System.setProperty("HADOOP_USER_NAME", "root")
    val sparkConf = newSparkConf().setAppName("dwd_member_import")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .setMaster("local[*]")
    val sparkSession =SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
    val ssc = sparkSession.sparkContext
   ssc.hadoopConfiguration.set("fs.defaultFS","hdfs://mycluster")
    ssc.hadoopConfiguration.set("dfs.nameservices","mycluster")
//    insertData(sparkSession)
        queryData(sparkSession)
  }


  /**
    * 读取hdfs日志文件通过hudi写入hdfs
    *
    * @param sparkSession
    */
  def insertData(sparkSession:SparkSession) = {
    import org.apache.spark.sql.functions._
    import sparkSession.implicits._
    val commitTime =System.currentTimeMillis().toString //生成提交时间
    val df =sparkSession.read.text("/user/atguigu/ods/member.log")
      .mapPartitions(partitions => {
        partitions.map(item => {
          val jsonObject =ParseJsonData.getJsonData(item.getString(0))
         DwsMember(jsonObject.getIntValue("uid"),jsonObject.getIntValue("ad_id")
            ,jsonObject.getString("fullname"),jsonObject.getString("iconurl")
            , jsonObject.getString("dt"),jsonObject.getString("dn"))
        })
      })
    val result =df.withColumn("ts", lit(commitTime)) //添加ts 时间戳列
      .withColumn("uuid",col("uid")) //添加uuid列如果数据中uuid相同hudi会进行去重
     .withColumn("hudipartition", concat_ws("/",col("dt"), col("dn"))) //增加hudi分区列
   result.write.format("org.apache.hudi")
      //     .options(org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs)
     .option("hoodie.insert.shuffle.parallelism", 12)
      .option("hoodie.upsert.shuffle.parallelism",12)
     .option("PRECOMBINE_FIELD_OPT_KEY", "ts") //指定提交时间列
     .option("RECORDKEY_FIELD_OPT_KEY", "uuid") //指定uuid唯一标示列
     .option("hoodie.table.name", "testTable")
      //     .option(DataSourceWriteOptions.DEFAULT_PARTITIONPATH_FIELD_OPT_VAL,"dt") //  发现api方式不起作用分区列
     .option("hoodie.datasource.write.partitionpath.field","hudipartition") //分区列
      .mode(SaveMode.Overwrite)
     .save("/user/atguigu/hudi")
  }

测试发现,Hudi不能指定多分区列,所以代码上分区列采用两列拼接成一列的方式,提交操作时必须指定ts和uuid。写入成功后查看hadoop路径上的文件。

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

  • 查询Hdfs上Hudi数据
/**
  * 查询hdfs上的hudi数据
  *
  * @param sparkSession
  */
def queryData(sparkSession: SparkSession) = {
  val df =sparkSession.read.format("org.apache.hudi")
   .load("/user/atguigu/hudi/*/*")
  df.show()
}

显示结果为:

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

  • 修改Hdfs上Hudi数据
def updateData(sparkSession: SparkSession) = {
  import org.apache.spark.sql.functions._
  import sparkSession.implicits._
  val commitTime =System.currentTimeMillis().toString //生成提交时间
  val df =sparkSession.read.text("/user/atguigu/ods/member.log")
    .mapPartitions(partitions => {
      partitions.map(item => {
        val jsonObject =ParseJsonData.getJsonData(item.getString(0))
       DwsMember(jsonObject.getIntValue("uid"),jsonObject.getIntValue("ad_id")
          , jsonObject.getString("fullname"),jsonObject.getString("iconurl")
          ,jsonObject.getString("dt"), jsonObject.getString("dn"))
      })
    })
  val result =df.withColumn("ts", lit(commitTime)) //添加ts 时间戳列
    .withColumn("uuid",col("uid")) //添加uuid列如果数据中uuid相同hudi会进行去重
   .withColumn("hudipartition", concat_ws("/",col("dt"), col("dn"))) //增加hudi分区列
 result.write.format("org.apache.hudi")
    //     .options(org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs)
    .option("hoodie.insert.shuffle.parallelism",12)
   .option("hoodie.upsert.shuffle.parallelism", 12)
   .option("PRECOMBINE_FIELD_OPT_KEY", "ts") //指定提交时间列
   .option("RECORDKEY_FIELD_OPT_KEY", "uuid") //指定uuid唯一标示列
   .option("hoodie.table.name", "testTable")
    //      .option(DataSourceWriteOptions.DEFAULT_PARTITIONPATH_FIELD_OPT_VAL,"dt") //  发现api方式不起作用分区列
   .option("hoodie.datasource.write.partitionpath.field","hudipartition") //分区列
    .mode(SaveMode.Append)
    .save("/user/atguigu/hudi")
}

虽然代码操作和新增一样只是修改了插入模式为append,但是hudi会根据uid判断进行更新数据,操作完毕后,生成一份最新的修改后的数据。同时hdfs路径上写入一份数据。

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

提交时间发生了变化

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

数据条数为94175

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

  • 增量查询
def incrementalQuery(sparkSession: SparkSession) = {
  val beginTime = 20200703130000l
  val df =sparkSession.read.format("org.apache.hudi")
    .option(DataSourceReadOptions.QUERY_TYPE_OPT_KEY,DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL) //指定模式为增量查询
   .option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY, beginTime) //设置开始查询的时间戳  不需要设置结束时间戳
    .load("/user/atguigu/hudi")
  df.show()
  println(df.count())
}
揭秘数据湖——长文详解Hudi从内核到实战(三)

 

根据haoodie_commit_time,时间进行查询,查询增量修改数据,注意参数begintime是和hadoop_commit_time对比而不是跟ts对比。如果beginitime填了比haoodie_commit_time大则会过滤所有数据。

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

  • 指定特定时间查询
def updateData(sparkSession: SparkSession) = {
  import org.apache.spark.sql.functions._
  import sparkSession.implicits._
  val commitTime =System.currentTimeMillis().toString //生成提交时间
  val df =sparkSession.read.text("/user/atguigu/ods/member.log")
    .mapPartitions(partitions => {
      partitions.map(item => {
        val jsonObject =ParseJsonData.getJsonData(item.getString(0))
       DwsMember(jsonObject.getIntValue("uid"),jsonObject.getIntValue("ad_id")
          ,jsonObject.getString("fullname"),jsonObject.getString("iconurl")
          , jsonObject.getString("dt"),jsonObject.getString("dn"))
      })
    }).limit(100)
  val result =df.withColumn("ts", lit(commitTime)) //添加ts 时间戳列
    .withColumn("uuid",col("uid")) //添加uuid列如果数据中uuid相同hudi会进行去重
   .withColumn("hudipartition", concat_ws("/",col("dt"), col("dn"))) //增加hudi分区列
 result.write.format("org.apache.hudi")
    //     .options(org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs)
   .option("hoodie.insert.shuffle.parallelism", 12)
   .option("hoodie.upsert.shuffle.parallelism", 12)
    .option("PRECOMBINE_FIELD_OPT_KEY","ts") //指定提交时间列
   .option("RECORDKEY_FIELD_OPT_KEY", "uuid") //指定uuid唯一标示列
   .option("hoodie.table.name", "testTable")
    //     .option(DataSourceWriteOptions.DEFAULT_PARTITIONPATH_FIELD_OPT_VAL,"dt") //  发现api方式不起作用分区列
    .option("hoodie.datasource.write.partitionpath.field","hudipartition") //分区列
    .mode(SaveMode.Append)
    .save("/user/atguigu/hudi")
}
def pointInTimeQuery(sparkSession: SparkSession) = {
  val beginTime = 20200703150000l
  val endTime = 20200703160000l
  val df =sparkSession.read.format("org.apache.hudi")
   .option(DataSourceReadOptions.QUERY_TYPE_OPT_KEY,DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL) //指定模式为增量查询
   .option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY, beginTime) //设置开始查询的时间戳
   .option(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, endTime)
    .load("/user/atguigu/hudi")
  df.show()
  println(df.count())


}

演示,update时limit只修改100条数据,然后根据时间戳进行查询,只会查询出进行修改的100条符合时间的数据.

揭秘数据湖——长文详解Hudi从内核到实战(三)

 

揭秘数据湖——长文详解Hudi从内核到实战(三)
 
想要了解跟多关于大数据培训课程内容欢迎关注尚硅谷大数据培训,尚硅谷除了这些技术文章外还有免费的高质量大数据培训课程视频供广大学员下载学习。

上一篇:
下一篇:
相关课程

java培训 大数据培训 前端培训 UI/UE设计培训

关于尚硅谷
教育理念
名师团队
学员心声
资源下载
视频下载
资料下载
工具下载
加入我们
招聘岗位
岗位介绍
招贤纳师
联系我们
全国统一咨询电话:010-56253825
地址:北京市昌平区宏福科技园2号楼3层(北京校区)

深圳市宝安区西部硅谷大厦B座C区一层(深圳校区)

上海市松江区谷阳北路166号大江商厦6层(上海校区)

武汉市东湖高新开发区东湖网谷(武汉校区)

西安市雁塔区和发智能大厦B座3层(西安校区)