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

/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

   val writeResult = DataSourceUtils.doWriteOperation(client, hoodieRecords, instantTime, operation)
   val (writeSuccessful, compactionInstant, clusteringInstant) =
        commitAndPerformPostOperations(sqlContext.sparkSession, df.schema,
          writeResult, parameters, writeClient, tableConfig, jsc,
          TableInstantInfo(basePath, instantTime, commitActionType, operation))

 public JavaRDD<WriteStatus> upsert(JavaRDD<HoodieRecord<T>> records, String instantTime) {
  HoodieTable<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>> table =
      initTable(WriteOperationType.UPSERT, Option.ofNullable(instantTime));
  table.validateUpsertSchema();
  preWrite(instantTime, WriteOperationType.UPSERT, table.getMetaClient());
  HoodieWriteMetadata<HoodieData<WriteStatus>> result = table.upsert(context, instantTime, HoodieJavaRDD.of(records));
  HoodieWriteMetadata<JavaRDD<WriteStatus>> resultRDD = result.clone(HoodieJavaRDD.getJavaRDD(result.getWriteStatuses()));
  if (result.getIndexLookupDuration().isPresent()) {
    metrics.updateIndexMetrics(LOOKUP_STR, result.getIndexLookupDuration().get().toMillis());
  }
  return postWrite(resultRDD, instantTime, table);
}

  public HoodieWriteMetadata<O> write(String instantTime,
                                  I inputRecords,
                                  HoodieEngineContext context,
                                  HoodieTable<T, I, K, O> table,
                                  boolean shouldCombine,
                                  int shuffleParallelism,
                                  BaseCommitActionExecutor<T, I, K, O, R> executor,
                                  WriteOperationType operationType) {
    try {
        // De-dupe/merge if needed
        I dedupedRecords =
            combineOnCondition(shouldCombine, inputRecords, shuffleParallelism, table);
        Instant lookupBegin = Instant.now();
        I taggedRecords = dedupedRecords;
        if (table.getIndex().requiresTagging(operationType)) {
          // perform index loop up to get existing location of records
          context.setJobStatus(this.getClass().getSimpleName(), "Tagging: " + table.getConfig().getTableName());
          taggedRecords = tag(dedupedRecords, context, table);
        }
        Duration indexLookupDuration = Duration.between(lookupBegin, Instant.now());
        HoodieWriteMetadata<O> result = executor.execute(taggedRecords);
        result.setIndexLookupDuration(indexLookupDuration);
        return result;
      } catch (Throwable e) {
        if (e instanceof HoodieUpsertException) {
          throw (HoodieUpsertException) e;
        }
        throw new HoodieUpsertException("Failed to upsert for commit time " + instantTime, e);
      }
    }

public HoodieWriteMetadata<HoodieData<WriteStatus>> execute(HoodieData<HoodieRecord<T>> inputRecords) {
   // Cache the tagged records, so we don't end up computing both
   // TODO: Consistent contract in HoodieWriteClient regarding preppedRecord storage level handling
   JavaRDD<HoodieRecord<T>> inputRDD = HoodieJavaRDD.getJavaRDD(inputRecords);
   if (inputRDD.getStorageLevel() == StorageLevel.NONE()) {
     inputRDD.persist(StorageLevel.MEMORY_AND_DISK_SER());
   } else {
     LOG.info("RDD PreppedRecords was persisted at: " + inputRDD.getStorageLevel());
   }
   WorkloadProfile workloadProfile = null;
   if (isWorkloadProfileNeeded()) {
     context.setJobStatus(this.getClass().getSimpleName(), "Building workload profile: " + config.getTableName());
     workloadProfile = new WorkloadProfile(buildProfile(inputRecords), operationType, table.getIndex().canIndexLogFiles());
     LOG.info("Input workload profile :" + workloadProfile);
   }
   // partition using the insert partitioner
   final Partitioner partitioner = getPartitioner(workloadProfile);
   if (isWorkloadProfileNeeded()) {
     saveWorkloadProfileMetadataToInflight(workloadProfile, instantTime);
   }
   // handle records update with clustering
   Set<HoodieFileGroupId> fileGroupsInPendingClustering =
       table.getFileSystemView().getFileGroupsInPendingClustering().map(Pair::getKey).collect(Collectors.toSet());
   HoodieData<HoodieRecord<T>> inputRecordsWithClusteringUpdate = fileGroupsInPendingClustering.isEmpty() ? inputRecords : clusteringHandleUpdate(inputRecords, fileGroupsInPendingClustering);
   context.setJobStatus(this.getClass().getSimpleName(), "Doing partition and writing data: " + config.getTableName());
   HoodieData<WriteStatus> writeStatuses = mapPartitionsAsRDD(inputRecordsWithClusteringUpdate, partitioner);
   HoodieWriteMetadata<HoodieData<WriteStatus>> result = new HoodieWriteMetadata<>();
   updateIndexAndCommitIfNeeded(writeStatuses, result);
   return result;
 }