Apache Hudi初探(八)(与spark的结合)--非bulk_insert模式

背景

之前讨论的都是’hoodie.datasource.write.operation’:'bulk_insert’的前提下，在这种模式下，是没有json文件的已形成如下的文件：

/dt=1/.hoodie_partition_metadata
/dt=1/2ffe3579-6ddb-4c5f-bf03-5c1b5dfce0a0-0_0-41263-0_20230528233336713.parquet
/dt=1/30b7d5b2-12e8-415a-8ec5-18206fe601c0-0_0-22102-0_20230528231643200.parquet
/dt=1/4abc1c6d-a8aa-4c15-affc-61a35171ce69-0_4-22106-0_20230528231643200.parquet
/dt=1/513dee80-2e8c-4db8-baee-a767b9dba41c-0_2-22104-0_20230528231643200.parquet
/dt=1/57076f86-0a62-4f52-8b50-31a5f769b26a-0_1-22103-0_20230528231643200.parquet
/dt=1/84553727-be9d-4273-bad9-0a38d9240815-0_0-59818-0_20230528233513387.parquet
/dt=1/fecd6a84-9a74-40b1-bfc1-13612a67a785-0_0-26640-0_20230528231723951.parquet

因为是"bulk insert"操作，所以没有去重的需要，所以直接采用spark原生的方式，
以下我们讨论非spark原生的方式，

闲说杂谈

继续Apache Hudi初探(二)(与spark的结合)
剩下的代码：

 val reconcileSchema = parameters(DataSourceWriteOptions.RECONCILE_SCHEMA.key()).toBooleanval (writeResult, writeClient: SparkRDDWriteClient[HoodieRecordPayload[Nothing]]) =...case _ => { // any other operation// register classes & schemasval (structName, nameSpace) = AvroConversionUtils.getAvroRecordNameAndNamespace(tblName)sparkContext.getConf.registerKryoClasses(Array(classOf[org.apache.avro.generic.GenericData],classOf[org.apache.avro.Schema]))// TODO(HUDI-4472) revisit and simplify schema handlingval sourceSchema = AvroConversionUtils.convertStructTypeToAvroSchema(df.schema, structName, nameSpace)val latestTableSchema = getLatestTableSchema(sqlContext.sparkSession, tableMetaClient).getOrElse(sourceSchema)val schemaEvolutionEnabled = parameters.getOrDefault(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key(), "false").toBooleanvar internalSchemaOpt = getLatestTableInternalSchema(hoodieConfig, tableMetaClient)val writerSchema: Schema =if (reconcileSchema) {// In case we need to reconcile the schema and schema evolution is enabled,// we will force-apply schema evolution to the writer's schemaif (schemaEvolutionEnabled && internalSchemaOpt.isEmpty) {internalSchemaOpt = Some(AvroInternalSchemaConverter.convert(sourceSchema))}if (internalSchemaOpt.isDefined) {...// Convert to RDD[HoodieRecord]val genericRecords: RDD[GenericRecord] = HoodieSparkUtils.createRdd(df, structName, nameSpace, reconcileSchema,org.apache.hudi.common.util.Option.of(writerSchema))val shouldCombine = parameters(INSERT_DROP_DUPS.key()).toBoolean ||operation.equals(WriteOperationType.UPSERT) ||parameters.getOrElse(HoodieWriteConfig.COMBINE_BEFORE_INSERT.key(),HoodieWriteConfig.COMBINE_BEFORE_INSERT.defaultValue()).toBooleanval hoodieAllIncomingRecords = genericRecords.map(gr => {val processedRecord = getProcessedRecord(partitionColumns, gr, dropPartitionColumns)val hoodieRecord = if (shouldCombine) {val orderingVal = HoodieAvroUtils.getNestedFieldVal(gr, hoodieConfig.getString(PRECOMBINE_FIELD), false, parameters.getOrElse(DataSourceWriteOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.key(),DataSourceWriteOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.defaultValue()).toBoolean).asInstanceOf[Comparable[_]]DataSourceUtils.createHoodieRecord(processedRecord,orderingVal,keyGenerator.getKey(gr),hoodieConfig.getString(PAYLOAD_CLASS_NAME))} else {DataSourceUtils.createHoodieRecord(processedRecord, keyGenerator.getKey(gr), hoodieConfig.getString(PAYLOAD_CLASS_NAME))}hoodieRecord}).toJavaRDD()val writerDataSchema = if (dropPartitionColumns) generateSchemaWithoutPartitionColumns(partitionColumns, writerSchema) else writerSchema// Create a HoodieWriteClient & issue the write.val client = hoodieWriteClient.getOrElse(DataSourceUtils.createHoodieClient(jsc, writerDataSchema.toString, path,tblName, mapAsJavaMap(addSchemaEvolutionParameters(parameters, internalSchemaOpt) - HoodieWriteConfig.AUTO_COMMIT_ENABLE.key))).asInstanceOf[SparkRDDWriteClient[HoodieRecordPayload[Nothing]]]if (isAsyncCompactionEnabled(client, tableConfig, parameters, jsc.hadoopConfiguration())) {asyncCompactionTriggerFn.get.apply(client)}if (isAsyncClusteringEnabled(client, parameters)) {asyncClusteringTriggerFn.get.apply(client)}val hoodieRecords =if (hoodieConfig.getBoolean(INSERT_DROP_DUPS)) {DataSourceUtils.dropDuplicates(jsc, hoodieAllIncomingRecords, mapAsJavaMap(parameters))} else {hoodieAllIncomingRecords}client.startCommitWithTime(instantTime, commitActionType)val writeResult = DataSourceUtils.doWriteOperation(client, hoodieRecords, instantTime, operation)(writeResult, client)}

• 如果开启了Schema Evolution，也就是hoodie.datasource.write.reconcile.schema是true,默认是false，就会进行schema的合并
  convertStructTypeToAvroSchema 把df的schema转换成avro的schema
  并且从*.hoodie/20230530073115535.deltacommit* 获取internalSchemaOpt，具体的合并就是把即将写入的schema和internalSchemaOpt进行合并
  最后赋值给writerSchema，有可能还需要hoodie.schema.on.read.enable，默认是false

• HoodieSparkUtils.createRdd 创建RDD
  把df转换为了RDD[GenericRecord]类型，赋值给genericRecords

• val hoodieAllIncomingRecords = genericRecords.map(gr => {

  • 首先如果是hoodie.datasource.write.drop.partition.columns为true(默认是false)，则会从schema中删除hoodie.datasource.write.
    partitionpath.field字段
  • 如果hoodie.datasource.write.insert.drop.duplicates为true(默认是false)或者hoodie.datasource.write.operation是upsert（默认
    是upsert），或者hoodie.combine.before.insert为true（默认是false），
    则会创建HoodieAvroRecord<>(hKey, payload)类型的实例，其中HoodieKey以recordkey和partitionpath组成,playload为OverwriteWithLatestAvroPayload实例，
  • hoodieAllIncomingRecords就变成了RDD[HoodieAvroRecord]

• writerDataSchema= client 这些就是创建SparkRDDWriteClient 客户端

• isAsyncCompactionEnabled
  默认asyncCompactionTriggerFnDefined是没有的，所以不会开启异步的Compaction，isAsyncClusteringEnabled同理也是

• val hoodieRecords =
  如果配置了hoodie.datasource.write.insert.drop.duplicates为true（默认是false），则会进行去重处理，具体是调用DataSourceUtils.dropDuplicates方法：

  SparkRDDReadClient client = new SparkRDDReadClient<>(new HoodieSparkEngineContext(jssc), writeConfig);return client.tagLocation(incomingHoodieRecords).filter(r -> !((HoodieRecord<HoodieRecordPayload>) r).isCurrentLocationKnown());
  

  • SparkRDDReadClient client 在创建Client的时候，会进行索引的创建this.index = SparkHoodieIndexFactory.createIndex(clientConfig);
    如果有hoodie.index.class设置，则实例化对象，否则根据hoodie.index.type的值来建立索引（默认是HoodieSimpleIndex,适合做测试用）
  • client.tagLocation(incomingHoodieRecords)…
    从要插入的记录中过滤出在index中不存在的记录，最终调用的是index.tagLocation方法
    如果hoodie.datasource.write.insert.drop.duplicates为false，则保留所有的记录

• client.startCommitWithTime 开始写操作，这涉及到回滚的操作

  • 会先过滤出需要回滚的的的写失败的文件，如果hoodie.cleaner.policy.failed.writes是EAGER（默认是EAGER），就会在这次提交中回滚失败的文件
  • 然后创建一个后缀为deltacommit.requested的文件，此时没有真正的写

• val writeResult = DataSourceUtils.doWriteOperation
  真正的写操作