实战|使用Spark Streaming写入Hudi

1. 项目背景

传统数仓的组织架构是针对离线数据的OLAP（联机事务分析）需求设计的，经常使用的导入数据方式为采用sqoop或spark定时做业逐批将业务库数据导入数仓。随着数据分析对实时性要求的不断提升，按小时、甚至分钟级的数据同步愈来愈广泛。由此展开了基于spark/flink流处理机制的（准）实时同步系统的开发。

然而实时同步数仓从一开始就面临以下几个挑战：

• 小文件问题。不管是spark的microbatch模式，仍是flink的逐条处理模式，每次写入HDFS时都是几M甚至几十KB的文件。长时间下来产生的大量小文件，会对HDFS namenode产生巨大的压力。
• 对update操做的支持。HDFS系统自己不支持数据的修改，没法实现同步过程当中对记录进行修改。
• 事务性。不管是追加数据仍是修改数据，如何保证事务性。即数据只在流处理程序commit操做时一次性写入HDFS，当程序rollback时，已写入或部分写入的数据能随之删除。

Hudi是针对以上问题的解决方案之一。如下是对Hudi的简单介绍，主要内容翻译自官网。

2. Hudi简介

2.1 时间线（Timeline）

Hudi内部按照操做时刻（instant）对表的全部操做维护了一条时间线，由此能够提供表在某一时刻的视图，还可以高效的提取出延后到达的数据。每个时刻包含：

• 时刻行为：对表操做的类型，包含：

commit：提交，将批次的数据原子性的写入表；

clean： 清除，后台做业，不断清除不须要的旧得版本的数据；

delta_commit：delta 提交是将批次记录原子性的写入MergeOnRead表中，数据写入的目的地是delta日志文件；

compacttion：压缩，后台做业，将不一样结构的数据，例如记录更新操做的行式存储的日志文件合并到列式存储的文件中。压缩自己是一个特殊的commit操做；

rollback：回滚，一些不成功时，删除全部部分写入的文件；

savepoint：保存点，标志某些文件组为“保存的“，这样cleaner就不会删除这些文件；

• 时刻时间：操做开始的时间戳；
• 状态：时刻的当前状态，包含：

requested 某个操做被安排执行，但还没有初始化

inflight 某个操做正在执行

completed 某一个操做在时间线上已经完成

Hudi保证按照时间线执行的操做按照时刻时间具备原子性及时间线一致性。

2.2 文件管理

Hudi表存在在DFS系统的 base path（用户写入Hudi时自定义） 目录下，在该目录下被分红不一样的分区。每个分区以 partition path 做为惟一的标识，组织形式与Hive相同。

每个分区内，文件经过惟一的 FileId 文件id 划分到 FileGroup 文件组。每个FileGroup包含多个 FileSlice 文件切片，每个切片包含一个由commit或compaction操做造成的base file 基础文件（parquet文件），以及包含对基础文件进行inserts/update操做的log files 日志文件（log文件)。Hudi采用了MVCC设计，compaction操做会将日志文件和对应的基础文件合并成新的文件切片，clean操做则删除无效的或老版本的文件。

2.3 索引

Hudi经过映射Hoodie键（记录键+ 分区路径）到文件id，提供了高效的upsert操做。当第一个版本的记录写入文件时，这个记录键值和文件的映射关系就不会发生任何改变。换言之，映射的文件组始终包含一组记录的全部版本。

2.4 表类型&查询

Hudi表类型定义了数据是如何被索引、分布到DFS系统，以及以上基本属性和时间线事件如何施加在这个组织上。查询类型定义了底层数据如何暴露给查询。

表类型	支持的查询类型
Copy On Write写时复制	快照查询 + 增量查询
Merge On Read读时合并	快照查询 + 增量查询 + 读取优化

2.4.1 表类型

Copy On Write：仅采用列式存储文件（parquet）存储文件。更新数据时，在写入的同时同步合并文件，仅仅修改文件的版次并重写。

Merge On Read：采用列式存储文件（parquet）+行式存储文件（avro）存储数据。更新数据时，新数据被写入delta文件并随后以异步或同步的方式合并成新版本的列式存储文件。

取舍	CopyOnWrite	MergeOnRead
数据延迟	高	低
Update cost (I/O)更新操做开销（I/O）	高（重写整个parquet）	低（追加到delta记录）
Parquet文件大小	小（高更新（I/O）开销）	大（低更新开销）
写入频率	高	低（取决于合并策略）

2.4.2 查询类型

• 快照查询：查询会看到之后的提交操做和合并操做的最新的表快照。对于merge on read表，会将最新的基础文件和delta文件进行合并，从而会看到近实时的数据（几分钟的延迟）。对于copy on write表，当存在更新/删除操做时或其余写操做时，会直接代替已有的parquet表。
• 增量查询：查询只会看到给定提交/合并操做以后新写入的数据。由此有效的提供了变动流，从而实现了增量数据管道。
• 读优化查询：查询会看到给定提交/合并操做以后表的最新快照。只会查看到最新的文件切片中的基础/列式存储文件，而且保证和非hudi列式存储表相同的查询效率。

取舍	快照	读取优化
数据延迟	低	高
查询延迟	高（合并基础/列式存储文件 + 行式存储delta / 日志 文件）	低（原有的基础/列式存储文件查询性能）

3. Spark结构化流写入Hudi

如下是整合spark结构化流+hudi的示意代码，因为Hudi OutputFormat目前只支持在spark rdd对象中调用，所以写入HDFS操做采用了spark structured streaming的forEachBatch算子。具体说明见注释。

package pers.machi.sparkhudi
    
import org.apache.log4j.Logger
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.{DataFrame, Row, SaveMode}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}

object SparkHudi {
  val logger = Logger.getLogger(SparkHudi.getClass)

  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder
      .appName("SparkHudi")
      //.master("local[*]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.default.parallelism", 9)
      .config("spark.sql.shuffle.partitions", 9)
      .enableHiveSupport()
      .getOrCreate()

    // 添加监听器，每一批次处理完成，将该批次的相关信息，如起始offset，抓取记录数量，处理时间打印到控制台
	spark.streams.addListener(new StreamingQueryListener() {
    	override def onQueryStarted(queryStarted: QueryStartedEvent): Unit = {
        	println("Query started: " + queryStarted.id)
    	}
    	override def onQueryTerminated(queryTerminated: QueryTerminatedEvent): Unit = {
       		println("Query terminated: " + queryTerminated.id)
    	}
    	override def onQueryProgress(queryProgress: QueryProgressEvent): Unit = {
        	println("Query made progress: " + queryProgress.progress)
    	}
	})

    // 定义kafka流
    val dataStreamReader = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "testTopic")
      .option("startingOffsets", "latest")
      .option("maxOffsetsPerTrigger", 100000)
      .option("failOnDataLoss", false)

    // 加载流数据，这里由于只是测试使用，直接读取kafka消息而不作其余处理，是spark结构化流会自动生成每一套消息对应的kafka元数据，如消息所在主题，分区，消息对应offset等。
    val df = dataStreamReader.load()
      .selectExpr(
        "topic as kafka_topic"
        "CAST(partition AS STRING) kafka_partition",
        "cast(timestamp as String) kafka_timestamp",
        "CAST(offset AS STRING) kafka_offset",
        "CAST(key AS STRING) kafka_key",
        "CAST(value AS STRING) kafka_value",
        "current_timestamp() current_time",
)
       .selectExpr(
        "kafka_topic"
 		"concat(kafka_partition,'-',kafka_offset) kafka_partition_offset",
        "kafka_offset",
        "kafka_timestamp",
        "kafka_key",
        "kafka_value",
    	"substr(current_time,1,10) partition_date")

	// 建立并启动query
    val query = df
      .writeStream
      .queryName("demo").
      .foreachBatch { (batchDF: DataFrame, _: Long) => {
        batchDF.persist()
            
        println(LocalDateTime.now() + "start writing cow table")
		batchDF.write.format("org.apache.hudi")
        	.option(TABLE_TYPE_OPT_KEY, "COPY_ON_WRITE")
			.option(PRECOMBINE_FIELD_OPT_KEY, "kafka_timestamp")
            // 以kafka分区和偏移量做为组合主键
            .option(RECORDKEY_FIELD_OPT_KEY, "kafka_partition_offset")
            // 以当前日期做为分区
            .option(PARTITIONPATH_FIELD_OPT_KEY, "partition_date")
            .option(TABLE_NAME, "copy_on_write_table")
            .option(HIVE_STYLE_PARTITIONING_OPT_KEY, true)
            .mode(SaveMode.Append)
            .save("/tmp/sparkHudi/COPY_ON_WRITE")

        println(LocalDateTime.now() + "start writing mor table")
        batchDF.write.format("org.apache.hudi")
            .option(TABLE_TYPE_OPT_KEY, "MERGE_ON_READ")
        	.option(TABLE_TYPE_OPT_KEY, "COPY_ON_WRITE")
			.option(PRECOMBINE_FIELD_OPT_KEY, "kafka_timestamp")
            .option(RECORDKEY_FIELD_OPT_KEY, "kafka_partition_offset")
            .option(PARTITIONPATH_FIELD_OPT_KEY, "partition_date")
            .option(TABLE_NAME, "merge_on_read_table")
            .option(HIVE_STYLE_PARTITIONING_OPT_KEY, true)
            .mode(SaveMode.Append)
            .save("/tmp/sparkHudi/MERGE_ON_READ")
            
        println(LocalDateTime.now() + "finish")
        batchDF.unpersist()
      }
      }
      .option("checkpointLocation", "/tmp/sparkHudi/checkpoint/")
      .start()

    query.awaitTermination()
  }
}

4. 测试结果

受限于测试条件，此次测试没有考虑update操做，而仅仅是测试hudi对追加新数据的性能。

数据程序一共运行5天，期间未发生报错致使程序退出。

kafka天天读取数据约1500万条，被消费的topic共有9个分区。

几点说明以下

1 是否有数据丢失及重复

因为每条记录的分区+偏移量具备惟一性，经过检查同一分区下是否有偏移量重复及不连续的状况，能够判定数据不存丢失及重复消费的状况。

2 最小可支持的单日写入数据条数

数据写入效率，对于cow及mor表，不存在更新操做时，写入速率接近。这本次测试中，spark每秒处理约170条记录。单日可处理1500万条记录。

3 cow和mor表文件大小对比

每十分钟读取两种表同一分区小文件大小，单位M。结果以下图，mor表文件大小增长较大，占用磁盘资源较多。不存在更新操做时，尽量使用cow表。