{"id":10021014,"date":"2025-09-25T18:03:07","date_gmt":"2025-09-25T10:03:07","guid":{"rendered":"https:\/\/itmax.cn\/?p=10021014"},"modified":"2025-09-25T18:03:07","modified_gmt":"2025-09-25T10:03:07","slug":"iceberg%e6%ba%90%e7%a0%81%e9%87%8c%e7%9a%84%e9%82%a3%e4%ba%9b%e8%ae%be%e8%ae%a1","status":"publish","type":"post","link":"https:\/\/itmax.cn\/?p=10021014","title":{"rendered":"Iceberg\u6e90\u7801\u91cc\u7684\u90a3\u4e9b\u8bbe\u8ba1"},"content":{"rendered":"

\u672c\u6587\u5728\u7eff\u6ce1\u6ce1\u201c\u72d7\u54e5\u7410\u8bdd\u201d\u9996\u53d1\u4e8e2025.9.24 <-\u5173\u6ce8\u4e0d\u8d70\u4e22\u3002<\/p><\/blockquote>\n

\u5927\u5bb6\u597d\u8fd9\u91cc\u662f\u72d7\u54e5\u3002\u91d19\u94f610\u9a6c\u4e0a\u8981\u5230\u4e86\u554a\uff0c\u4e00\u4e9b\u673a\u667a\u7684\u540c\u5b66\u5df2\u7ecf\u5f00\u59cb\u63d0\u524d\u5907\u6218\u4e86\uff0c\u8fd1\u671f\u627e\u6211\u8f85\u5bfc\u7684\u540c\u5b66\u4e5f\u5f88\u591a\uff0c\u7ed9\u6211\u7d2f\u9ebb\u4e86\u3002<\/p>\n

\u8bf4\u6b63\u4e8b\u554a\u3002\u867d\u7136\u5f88\u591a\u5927\u6570\u636e\u5f00\u53d1\u540c\u5b66\u4f1a\u5199Java\uff0c\u4f46\u662fCoding\u8bbe\u8ba1\u80fd\u529b\uff08\u8bbe\u8ba1\u6a21\u5f0f\u3001SOLID\u3001\u7b80\u6d01\u67b6\u6784\u3001DDD\u4e4b\u7c7b\u7684\uff09\u662f\u5f31\u4e8e\u4e00\u4e9b\u4f20\u7edfJava\u540e\u53f0\u540c\u5b66\u7684\uff0c\u672c\u8eab\u7684\u5c97\u4f4d\u8981\u6c42\u4ee5\u53ca\u4e1a\u52a1\u7684\u591a\u53d8\u6027\u90fd\u4f1a\u8981\u6c42\u4ed6\u4eec\u8fd9\u4e48\u505a\u3002\u800c\u5927\u6570\u636e\u7684\u540c\u5b66\u5728\u8fd9\u65b9\u9762\u76f8\u5bf9\u4f1a\u504f\u5f31\u4e00\u70b9\uff0c\u633a\u597d\u7406\u89e3\u7684\uff0c\u56e0\u4e3a\u6838\u5fc3\u6280\u80fd\u4e0d\u5728\u8fd9\u91cc\u3002<\/p>\n

\u90a3\u6211\u600e\u4e48\u5e2e\u52a9\u8fd9\u4e9b\u540c\u5b66\u5462\uff1f\u4e4b\u524d\u4e0d\u662f\u5728\u661f\u7403\u91cc\u5256\u6790\u4e86\u4e00\u4e9b\u5f00\u6e90\u8f6f\u4ef6\u7684\u6e90\u7801\u561b\uff0c\u4ecePaimon\u5230AutoMQ\u518d\u5230\u73b0\u5728\u7684Iceberg\u3002\u6211\u60f3\u8fd9\u4e0d\u662f\u4e2a\u597d\u673a\u4f1a\u561b\uff0c\u65e2\u7136\u90fd\u4e86\u89e3\u91cc\u9762\u7684\u903b\u8f91\u4e86\uff0c\u57fa\u4e8e\u7406\u89e3\u5b83\u4eec\u4e1a\u52a1\u4e4b\u4e0a\uff0c\u518d\u8bb2\u8bbe\u8ba1\uff0c\u4e0d\u662f\u7f8e\u6ecb\u6ecb\uff1f<\/p>\n

\u90a3\u90fd\u8bf4\u5230Iceberg\u4e86\uff0c\u6211\u4eec\u4eca\u5929\u5c31\u4ee5SparkSQL\u5199\u5165Iceberg\u7684\u94fe\u8def\uff0c\u770b\u770b\u6709\u54ea\u4e9b\u8bbe\u8ba1\u554a\u3002<\/p>\n

\u9009\u62e9\u4ee3\u7801\u5206\u652f apache-iceberg-1.9.2<\/p><\/blockquote>\n

\u6309\u7167\u56fd\u9645\u60ef\u4f8b\uff0c\u4ee3\u7801\u5730\u56fe\u653e\u4e0a\u3002<\/p>\n

|-- SparkWrite\n  -- createWriter #\u7531Spark\u6846\u67b6\u8c03\u7528\uff0c\u4ea7\u751f\u7684\u5b9e\u4f8b\u4e2d\u7684write\u65b9\u6cd5\u4e5f\u662f\n|-- PartitionedDataWriter\n  -- write\n|-- FanoutDataWriter\n  -- write #\u4e2d\u95f4\u7ed5\u4e86\u4e00\u4e9bfactory\n|-- DataWriter\n  -- write\n|-- FileAppender\n  -- add #\u771f\u6b63\u7684\u6587\u4ef6\u5199\u5165<\/code><\/pre>\n
SparkWrite<\/h2>\n
package org.apache.iceberg.spark.source;\n\n\/\/ignore some import\nabstract class SparkWrite implements Write, RequiresDistributionAndOrdering {\n  private static final Logger LOG = LoggerFactory.getLogger(SparkWrite.class);\n\n  private final JavaSparkContext sparkContext;\n  private final SparkWriteConf writeConf;\n  private final Table table;\n  private final String queryId;\n  private final FileFormat format;\n  private final String applicationId;\n  private final boolean wapEnabled;\n  private final String wapId;\n  private final int outputSpecId;\n  private final String branch;\n  private final long targetFileSize;\n  private final Schema writeSchema;\n  private final StructType dsSchema;\n  private final Map<String, String> extraSnapshotMetadata;\n  private final boolean useFanoutWriter;\n  private final SparkWriteRequirements writeRequirements;\n  private final Map<String, String> writeProperties;\n\n  private boolean cleanupOnAbort = false;\n\n  SparkWrite(\n      SparkSession spark,\n      Table table,\n      SparkWriteConf writeConf,\n      LogicalWriteInfo writeInfo,\n      String applicationId,\n      Schema writeSchema,\n      StructType dsSchema,\n      SparkWriteRequirements writeRequirements) {\n    this.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext());\n    this.table = table;\n    this.writeConf = writeConf;\n    this.queryId = writeInfo.queryId();\n    this.format = writeConf.dataFileFormat();\n    this.applicationId = applicationId;\n    this.wapEnabled = writeConf.wapEnabled();\n    this.wapId = writeConf.wapId();\n    this.branch = writeConf.branch();\n    this.targetFileSize = writeConf.targetDataFileSize();\n    this.writeSchema = writeSchema;\n    this.dsSchema = dsSchema;\n    this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();\n    this.useFanoutWriter = writeConf.useFanoutWriter(writeRequirements);\n    this.writeRequirements = writeRequirements;\n    this.outputSpecId = writeConf.outputSpecId();\n    this.writeProperties = writeConf.writeProperties();\n  }\n\n  @Override\n  public Distribution requiredDistribution() {\n    Distribution distribution = writeRequirements.distribution();\n    LOG.debug("Requesting {} as write distribution for table {}", distribution, table.name());\n    return distribution;\n  }\n\n  @Override\n  public boolean distributionStrictlyRequired() {\n    return false;\n  }\n\n  @Override\n  public SortOrder[] requiredOrdering() {\n    SortOrder[] ordering = writeRequirements.ordering();\n    LOG.debug("Requesting {} as write ordering for table {}", ordering, table.name());\n    return ordering;\n  }\n\n  @Override\n  public long advisoryPartitionSizeInBytes() {\n    long size = writeRequirements.advisoryPartitionSize();\n    LOG.debug("Requesting {} bytes advisory partition size for table {}", size, table.name());\n    return size;\n  }\n\n  BatchWrite asBatchAppend() {\n    return new BatchAppend();\n  }\n\n  BatchWrite asDynamicOverwrite() {\n    return new DynamicOverwrite();\n  }\n\n  BatchWrite asOverwriteByFilter(Expression overwriteExpr) {\n    return new OverwriteByFilter(overwriteExpr);\n  }\n\n  BatchWrite asCopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {\n    return new CopyOnWriteOperation(scan, isolationLevel);\n  }\n\n  BatchWrite asRewrite(String fileSetID) {\n    return new RewriteFiles(fileSetID);\n  }\n\n  StreamingWrite asStreamingAppend() {\n    return new StreamingAppend();\n  }\n\n  StreamingWrite asStreamingOverwrite() {\n    return new StreamingOverwrite();\n  }\n\n  \/\/ the writer factory works for both batch and streaming\n  private WriterFactory createWriterFactory() {\n    \/\/ broadcast the table metadata as the writer factory will be sent to executors\n    Broadcast<Table> tableBroadcast =\n        sparkContext.broadcast(SerializableTableWithSize.copyOf(table));\n    return new WriterFactory(\n        tableBroadcast,\n        queryId,\n        format,\n        outputSpecId,\n        targetFileSize,\n        writeSchema,\n        dsSchema,\n        useFanoutWriter,\n        writeProperties);\n  }\n\n  private void commitOperation(SnapshotUpdate<?> operation, String description) {\n    LOG.info("Committing {} to table {}", description, table);\n    if (applicationId != null) {\n      operation.set("spark.app.id", applicationId);\n    }\n\n    if (!extraSnapshotMetadata.isEmpty()) {\n      extraSnapshotMetadata.forEach(operation::set);\n    }\n\n    if (!CommitMetadata.commitProperties().isEmpty()) {\n      CommitMetadata.commitProperties().forEach(operation::set);\n    }\n\n    if (wapEnabled && wapId != null) {\n      \/\/ write-audit-publish is enabled for this table and job\n      \/\/ stage the changes without changing the current snapshot\n      operation.set(SnapshotSummary.STAGED_WAP_ID_PROP, wapId);\n      operation.stageOnly();\n    }\n\n    if (branch != null) {\n      operation.toBranch(branch);\n    }\n\n    try {\n      long start = System.currentTimeMillis();\n      operation.commit(); \/\/ abort is automatically called if this fails\n      long duration = System.currentTimeMillis() - start;\n      LOG.info("Committed in {} ms", duration);\n    } catch (Exception e) {\n      cleanupOnAbort = e instanceof CleanableFailure;\n      throw e;\n    }\n  }\n\n  private void abort(WriterCommitMessage[] messages) {\n    if (cleanupOnAbort) {\n      SparkCleanupUtil.deleteFiles("job abort", table.io(), files(messages));\n    } else {\n      LOG.warn("Skipping cleanup of written files");\n    }\n  }\n\n  private List<DataFile> files(WriterCommitMessage[] messages) {\n    List<DataFile> files = Lists.newArrayList();\n\n    for (WriterCommitMessage message : messages) {\n      if (message != null) {\n        TaskCommit taskCommit = (TaskCommit) message;\n        files.addAll(Arrays.asList(taskCommit.files()));\n      }\n    }\n\n    return files;\n  }\n\n  @Override\n  public String toString() {\n    return String.format("IcebergWrite(table=%s, format=%s)", table, format);\n  }\n\n  private abstract class BaseBatchWrite implements BatchWrite {\n    @Override\n    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {\n      return createWriterFactory();\n    }\n\n    @Override\n    public boolean useCommitCoordinator() {\n      return false;\n    }\n\n    @Override\n    public void abort(WriterCommitMessage[] messages) {\n      SparkWrite.this.abort(messages);\n    }\n\n    @Override\n    public String toString() {\n      return String.format("IcebergBatchWrite(table=%s, format=%s)", table, format);\n    }\n  }\n\n  private class BatchAppend extends BaseBatchWrite {\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      AppendFiles append = table.newAppend();\n\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        numFiles += 1;\n        append.appendFile(file);\n      }\n\n      commitOperation(append, String.format("append with %d new data files", numFiles));\n    }\n  }\n\n  private class DynamicOverwrite extends BaseBatchWrite {\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      List<DataFile> files = files(messages);\n\n      if (files.isEmpty()) {\n        LOG.info("Dynamic overwrite is empty, skipping commit");\n        return;\n      }\n\n      ReplacePartitions dynamicOverwrite = table.newReplacePartitions();\n      IsolationLevel isolationLevel = writeConf.isolationLevel();\n      Long validateFromSnapshotId = writeConf.validateFromSnapshotId();\n\n      if (isolationLevel != null && validateFromSnapshotId != null) {\n        dynamicOverwrite.validateFromSnapshot(validateFromSnapshotId);\n      }\n\n      if (isolationLevel == SERIALIZABLE) {\n        dynamicOverwrite.validateNoConflictingData();\n        dynamicOverwrite.validateNoConflictingDeletes();\n\n      } else if (isolationLevel == SNAPSHOT) {\n        dynamicOverwrite.validateNoConflictingDeletes();\n      }\n\n      int numFiles = 0;\n      for (DataFile file : files) {\n        numFiles += 1;\n        dynamicOverwrite.addFile(file);\n      }\n\n      commitOperation(\n          dynamicOverwrite,\n          String.format("dynamic partition overwrite with %d new data files", numFiles));\n    }\n  }\n\n  private class OverwriteByFilter extends BaseBatchWrite {\n    private final Expression overwriteExpr;\n\n    private OverwriteByFilter(Expression overwriteExpr) {\n      this.overwriteExpr = overwriteExpr;\n    }\n\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      OverwriteFiles overwriteFiles = table.newOverwrite();\n      overwriteFiles.overwriteByRowFilter(overwriteExpr);\n\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        numFiles += 1;\n        overwriteFiles.addFile(file);\n      }\n\n      IsolationLevel isolationLevel = writeConf.isolationLevel();\n      Long validateFromSnapshotId = writeConf.validateFromSnapshotId();\n\n      if (isolationLevel != null && validateFromSnapshotId != null) {\n        overwriteFiles.validateFromSnapshot(validateFromSnapshotId);\n      }\n\n      if (isolationLevel == SERIALIZABLE) {\n        overwriteFiles.validateNoConflictingDeletes();\n        overwriteFiles.validateNoConflictingData();\n\n      } else if (isolationLevel == SNAPSHOT) {\n        overwriteFiles.validateNoConflictingDeletes();\n      }\n\n      String commitMsg =\n          String.format("overwrite by filter %s with %d new data files", overwriteExpr, numFiles);\n      commitOperation(overwriteFiles, commitMsg);\n    }\n  }\n\n  private class CopyOnWriteOperation extends BaseBatchWrite {\n    private final SparkCopyOnWriteScan scan;\n    private final IsolationLevel isolationLevel;\n\n    private CopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {\n      this.scan = scan;\n      this.isolationLevel = isolationLevel;\n    }\n\n    private List<DataFile> overwrittenFiles() {\n      if (scan == null) {\n        return ImmutableList.of();\n      } else {\n        return scan.tasks().stream().map(FileScanTask::file).collect(Collectors.toList());\n      }\n    }\n\n    private Expression conflictDetectionFilter() {\n      \/\/ the list of filter expressions may be empty but is never null\n      List<Expression> scanFilterExpressions = scan.filterExpressions();\n\n      Expression filter = Expressions.alwaysTrue();\n\n      for (Expression expr : scanFilterExpressions) {\n        filter = Expressions.and(filter, expr);\n      }\n\n      return filter;\n    }\n\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      OverwriteFiles overwriteFiles = table.newOverwrite();\n\n      List<DataFile> overwrittenFiles = overwrittenFiles();\n      int numOverwrittenFiles = overwrittenFiles.size();\n      for (DataFile overwrittenFile : overwrittenFiles) {\n        overwriteFiles.deleteFile(overwrittenFile);\n      }\n\n      int numAddedFiles = 0;\n      for (DataFile file : files(messages)) {\n        numAddedFiles += 1;\n        overwriteFiles.addFile(file);\n      }\n\n      \/\/ the scan may be null if the optimizer replaces it with an empty relation (e.g. false cond)\n      \/\/ no validation is needed in this case as the command does not depend on the table state\n      if (scan != null) {\n        if (isolationLevel == SERIALIZABLE) {\n          commitWithSerializableIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);\n        } else if (isolationLevel == SNAPSHOT) {\n          commitWithSnapshotIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);\n        } else {\n          throw new IllegalArgumentException("Unsupported isolation level: " + isolationLevel);\n        }\n\n      } else {\n        commitOperation(\n            overwriteFiles,\n            String.format("overwrite with %d new data files (no validation)", numAddedFiles));\n      }\n    }\n\n    private void commitWithSerializableIsolation(\n        OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {\n      Long scanSnapshotId = scan.snapshotId();\n      if (scanSnapshotId != null) {\n        overwriteFiles.validateFromSnapshot(scanSnapshotId);\n      }\n\n      Expression conflictDetectionFilter = conflictDetectionFilter();\n      overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);\n      overwriteFiles.validateNoConflictingData();\n      overwriteFiles.validateNoConflictingDeletes();\n\n      String commitMsg =\n          String.format(\n              "overwrite of %d data files with %d new data files, scanSnapshotId: %d, conflictDetectionFilter: %s",\n              numOverwrittenFiles, numAddedFiles, scanSnapshotId, conflictDetectionFilter);\n      commitOperation(overwriteFiles, commitMsg);\n    }\n\n    private void commitWithSnapshotIsolation(\n        OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {\n      Long scanSnapshotId = scan.snapshotId();\n      if (scanSnapshotId != null) {\n        overwriteFiles.validateFromSnapshot(scanSnapshotId);\n      }\n\n      Expression conflictDetectionFilter = conflictDetectionFilter();\n      overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);\n      overwriteFiles.validateNoConflictingDeletes();\n\n      String commitMsg =\n          String.format(\n              "overwrite of %d data files with %d new data files",\n              numOverwrittenFiles, numAddedFiles);\n      commitOperation(overwriteFiles, commitMsg);\n    }\n  }\n\n  private class RewriteFiles extends BaseBatchWrite {\n    private final String fileSetID;\n\n    private RewriteFiles(String fileSetID) {\n      this.fileSetID = fileSetID;\n    }\n\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      FileRewriteCoordinator coordinator = FileRewriteCoordinator.get();\n      coordinator.stageRewrite(table, fileSetID, DataFileSet.of(files(messages)));\n    }\n  }\n\n  private abstract class BaseStreamingWrite implements StreamingWrite {\n    private static final String QUERY_ID_PROPERTY = "spark.sql.streaming.queryId";\n    private static final String EPOCH_ID_PROPERTY = "spark.sql.streaming.epochId";\n\n    protected abstract String mode();\n\n    @Override\n    public StreamingDataWriterFactory createStreamingWriterFactory(PhysicalWriteInfo info) {\n      return createWriterFactory();\n    }\n\n    @Override\n    public boolean useCommitCoordinator() {\n      return false;\n    }\n\n    @Override\n    public final void commit(long epochId, WriterCommitMessage[] messages) {\n      LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, mode());\n\n      table.refresh();\n\n      Long lastCommittedEpochId = findLastCommittedEpochId();\n      if (lastCommittedEpochId != null && epochId <= lastCommittedEpochId) {\n        LOG.info("Skipping epoch {} for query {} as it was already committed", epochId, queryId);\n        return;\n      }\n\n      doCommit(epochId, messages);\n    }\n\n    protected abstract void doCommit(long epochId, WriterCommitMessage[] messages);\n\n    protected <T> void commit(SnapshotUpdate<T> snapshotUpdate, long epochId, String description) {\n      snapshotUpdate.set(QUERY_ID_PROPERTY, queryId);\n      snapshotUpdate.set(EPOCH_ID_PROPERTY, Long.toString(epochId));\n      commitOperation(snapshotUpdate, description);\n    }\n\n    private Long findLastCommittedEpochId() {\n      Snapshot snapshot = table.currentSnapshot();\n      Long lastCommittedEpochId = null;\n      while (snapshot != null) {\n        Map<String, String> summary = snapshot.summary();\n        String snapshotQueryId = summary.get(QUERY_ID_PROPERTY);\n        if (queryId.equals(snapshotQueryId)) {\n          lastCommittedEpochId = Long.valueOf(summary.get(EPOCH_ID_PROPERTY));\n          break;\n        }\n        Long parentSnapshotId = snapshot.parentId();\n        snapshot = parentSnapshotId != null ? table.snapshot(parentSnapshotId) : null;\n      }\n      return lastCommittedEpochId;\n    }\n\n    @Override\n    public void abort(long epochId, WriterCommitMessage[] messages) {\n      SparkWrite.this.abort(messages);\n    }\n\n    @Override\n    public String toString() {\n      return String.format("IcebergStreamingWrite(table=%s, format=%s)", table, format);\n    }\n  }\n\n  private class StreamingAppend extends BaseStreamingWrite {\n    @Override\n    protected String mode() {\n      return "append";\n    }\n\n    @Override\n    protected void doCommit(long epochId, WriterCommitMessage[] messages) {\n      AppendFiles append = table.newFastAppend();\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        append.appendFile(file);\n        numFiles++;\n      }\n      commit(append, epochId, String.format("streaming append with %d new data files", numFiles));\n    }\n  }\n\n  private class StreamingOverwrite extends BaseStreamingWrite {\n    @Override\n    protected String mode() {\n      return "complete";\n    }\n\n    @Override\n    public void doCommit(long epochId, WriterCommitMessage[] messages) {\n      OverwriteFiles overwriteFiles = table.newOverwrite();\n      overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        overwriteFiles.addFile(file);\n        numFiles++;\n      }\n      commit(\n          overwriteFiles,\n          epochId,\n          String.format("streaming complete overwrite with %d new data files", numFiles));\n    }\n  }\n\n  public static class TaskCommit implements WriterCommitMessage {\n    private final DataFile[] taskFiles;\n\n    TaskCommit(DataFile[] taskFiles) {\n      this.taskFiles = taskFiles;\n    }\n\n    \/\/ Reports bytesWritten and recordsWritten to the Spark output metrics.\n    \/\/ Can only be called in executor.\n    void reportOutputMetrics() {\n      long bytesWritten = 0L;\n      long recordsWritten = 0L;\n      for (DataFile dataFile : taskFiles) {\n        bytesWritten += dataFile.fileSizeInBytes();\n        recordsWritten += dataFile.recordCount();\n      }\n\n      TaskContext taskContext = TaskContext$.MODULE$.get();\n      if (taskContext != null) {\n        OutputMetrics outputMetrics = taskContext.taskMetrics().outputMetrics();\n        outputMetrics.setBytesWritten(bytesWritten);\n        outputMetrics.setRecordsWritten(recordsWritten);\n      }\n    }\n\n    DataFile[] files() {\n      return taskFiles;\n    }\n  }\n\n  private static class WriterFactory implements DataWriterFactory, StreamingDataWriterFactory {\n    private final Broadcast<Table> tableBroadcast;\n    private final FileFormat format;\n    private final int outputSpecId;\n    private final long targetFileSize;\n    private final Schema writeSchema;\n    private final StructType dsSchema;\n    private final boolean useFanoutWriter;\n    private final String queryId;\n    private final Map<String, String> writeProperties;\n\n    protected WriterFactory(\n        Broadcast<Table> tableBroadcast,\n        String queryId,\n        FileFormat format,\n        int outputSpecId,\n        long targetFileSize,\n        Schema writeSchema,\n        StructType dsSchema,\n        boolean useFanoutWriter,\n        Map<String, String> writeProperties) {\n      this.tableBroadcast = tableBroadcast;\n      this.format = format;\n      this.outputSpecId = outputSpecId;\n      this.targetFileSize = targetFileSize;\n      this.writeSchema = writeSchema;\n      this.dsSchema = dsSchema;\n      this.useFanoutWriter = useFanoutWriter;\n      this.queryId = queryId;\n      this.writeProperties = writeProperties;\n    }\n\n    @Override\n    public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {\n      return createWriter(partitionId, taskId, 0);\n    }\n\n    @Override\n    public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {\n      Table table = tableBroadcast.value();\n      PartitionSpec spec = table.specs().get(outputSpecId);\n      FileIO io = table.io();\n      String operationId = queryId + "-" + epochId;\n      OutputFileFactory fileFactory =\n          OutputFileFactory.builderFor(table, partitionId, taskId)\n              .format(format)\n              .operationId(operationId)\n              .build();\n      SparkFileWriterFactory writerFactory =\n          SparkFileWriterFactory.builderFor(table)\n              .dataFileFormat(format)\n              .dataSchema(writeSchema)\n              .dataSparkType(dsSchema)\n              .writeProperties(writeProperties)\n              .build();\n\n      if (spec.isUnpartitioned()) {\n        return new UnpartitionedDataWriter(writerFactory, fileFactory, io, spec, targetFileSize);\n\n      } else {\n        return new PartitionedDataWriter(\n            writerFactory,\n            fileFactory,\n            io,\n            spec,\n            writeSchema,\n            dsSchema,\n            targetFileSize,\n            useFanoutWriter);\n      }\n    }\n  }\n\n  private static class UnpartitionedDataWriter implements DataWriter<InternalRow> {\n    private final FileWriter<InternalRow, DataWriteResult> delegate;\n    private final FileIO io;\n\n    private UnpartitionedDataWriter(\n        SparkFileWriterFactory writerFactory,\n        OutputFileFactory fileFactory,\n        FileIO io,\n        PartitionSpec spec,\n        long targetFileSize) {\n      this.delegate =\n          new RollingDataWriter<>(writerFactory, fileFactory, io, targetFileSize, spec, null);\n      this.io = io;\n    }\n\n    @Override\n    public void write(InternalRow record) throws IOException {\n      delegate.write(record);\n    }\n\n    @Override\n    public WriterCommitMessage commit() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));\n      taskCommit.reportOutputMetrics();\n      return taskCommit;\n    }\n\n    @Override\n    public void abort() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());\n    }\n\n    @Override\n    public void close() throws IOException {\n      delegate.close();\n    }\n  }\n\n  private static class PartitionedDataWriter implements DataWriter<InternalRow> {\n    private final PartitioningWriter<InternalRow, DataWriteResult> delegate;\n    private final FileIO io;\n    private final PartitionSpec spec;\n    private final PartitionKey partitionKey;\n    private final InternalRowWrapper internalRowWrapper;\n\n    private PartitionedDataWriter(\n        SparkFileWriterFactory writerFactory,\n        OutputFileFactory fileFactory,\n        FileIO io,\n        PartitionSpec spec,\n        Schema dataSchema,\n        StructType dataSparkType,\n        long targetFileSize,\n        boolean fanoutEnabled) {\n      if (fanoutEnabled) {\n        this.delegate = new FanoutDataWriter<>(writerFactory, fileFactory, io, targetFileSize);\n      } else {\n        this.delegate = new ClusteredDataWriter<>(writerFactory, fileFactory, io, targetFileSize);\n      }\n      this.io = io;\n      this.spec = spec;\n      this.partitionKey = new PartitionKey(spec, dataSchema);\n      this.internalRowWrapper = new InternalRowWrapper(dataSparkType, dataSchema.asStruct());\n    }\n\n    @Override\n    public void write(InternalRow row) throws IOException {\n      partitionKey.partition(internalRowWrapper.wrap(row));\n      delegate.write(row, spec, partitionKey);\n    }\n\n    @Override\n    public WriterCommitMessage commit() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));\n      taskCommit.reportOutputMetrics();\n      return taskCommit;\n    }\n\n    @Override\n    public void abort() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());\n    }\n\n    @Override\n    public void close() throws IOException {\n      delegate.close();\n    }\n  }\n}<\/code><\/pre>\n
SparkWrite\u662fSparkSQL\u5199\u5165\u6570\u636e\u7684\u5165\u53e3\u554a\uff0c\u7136\u540e\u4ece\u4ee3\u7801\u4e0a\uff0c\u6211\u4eec\u4e0d\u96be\u770b\u51fa\u5b83\u662f\u4e00\u4e2a\u62bd\u8c61\u7c7b\u554a\u3002\u90a3\u5b83\u7684\u5b9e\u73b0\u5728\u54ea\u91cc\u5462\uff1f<\/p>\n
  @Override\n  public Write build() {\n    \/\/ Validate\n    Schema writeSchema = validateOrMergeWriteSchema(table, dsSchema, writeConf);\n    SparkUtil.validatePartitionTransforms(table.spec());\n\n    \/\/ Get application id\n    String appId = spark.sparkContext().applicationId();\n\n    return new SparkWrite(\n        spark, table, writeConf, writeInfo, appId, writeSchema, dsSchema, writeRequirements()) {\n\n      @Override\n      public BatchWrite toBatch() {\n        if (rewrittenFileSetId != null) {\n          return asRewrite(rewrittenFileSetId);\n        } else if (overwriteByFilter) {\n          return asOverwriteByFilter(overwriteExpr);\n        } else if (overwriteDynamic) {\n          return asDynamicOverwrite();\n        } else if (overwriteFiles) {\n          return asCopyOnWriteOperation(copyOnWriteScan, copyOnWriteIsolationLevel);\n        } else {\n          return asBatchAppend();\n        }\n      }\n\n      @Override\n      public StreamingWrite toStreaming() {\n        Preconditions.checkState(\n            !overwriteDynamic, "Unsupported streaming operation: dynamic partition overwrite");\n        Preconditions.checkState(\n            !overwriteByFilter || overwriteExpr == Expressions.alwaysTrue(),\n            "Unsupported streaming operation: overwrite by filter: %s",\n            overwriteExpr);\n        Preconditions.checkState(\n            rewrittenFileSetId == null, "Unsupported streaming operation: rewrite");\n\n        if (overwriteByFilter) {\n          return asStreamingOverwrite();\n        } else {\n          return asStreamingAppend();\n        }\n      }\n    };\n  }<\/code><\/pre>\n
\u5728SparkWriteBuilder\u7684build\u65b9\u6cd5\u4e2d\u6839\u636e\u914d\u7f6e\u6784\u5efa\u51fa\u6765\u7684\u3002<\/p>\n
\u8fd9\u91cc\u4f1a\u6839\u636ebuild\u524d\u4f20\u5165\u7684\u4e0d\u540c\u5c5e\u6027\uff0c\u6784\u5efa\u4e0d\u540c\u7684Write\u3002<\/p>\n
toBatch() \u65b9\u6cd5\u3002\u6839\u636e\u4e0d\u540c\u7684\u5199\u5165\u6a21\u5f0f\u8fd4\u56de\u76f8\u5e94\u7684\u6279\u5904\u7406\u5199\u5165\u5668\uff1a<\/p>\n
    \n
  • \u91cd\u5199\u6a21\u5f0f: asRewrite(rewrittenFileSetId) – \u7528\u4e8e\u91cd\u5199\u73b0\u6709\u6587\u4ef6<\/li>\n
  • \u8fc7\u6ee4\u6761\u4ef6\u8986\u76d6: asOverwriteByFilter(overwriteExpr) – \u6839\u636e\u8fc7\u6ee4\u6761\u4ef6\u8986\u76d6\u6570\u636e<\/li>\n
  • \u52a8\u6001\u5206\u533a\u8986\u76d6: asDynamicOverwrite() – \u52a8\u6001\u8986\u76d6\u5206\u533a\u6570\u636e<\/li>\n
  • \u6587\u4ef6\u7ea7\u522b\u8986\u76d6: asCopyOnWriteOperation(copyOnWriteScan, copyOnWriteIsolationLevel) – Copy-on-write \u64cd\u4f5c<\/li>\n
  • \u8ffd\u52a0\u6a21\u5f0f: asBatchAppend() – \u9ed8\u8ba4\u7684\u8ffd\u52a0\u5199\u5165<\/li>\n<\/ul>\n
    toStreaming() \u65b9\u6cd5<\/p>\n
      \n
    • \u7528\u4e8e\u6d41\u5f0f\u5199\u5165\uff0c\u6839\u636e\u914d\u7f6e\u51b3\u5b9a\u662f\u6d41\u5f0f\u8ffd\u52a0\u8fd8\u662f\u6d41\u5f0f\u8986\u76d6\u3002<\/li>\n<\/ul>\n
      \u8fd9\u91cc\u5176\u5b9e\u662f\u4e00\u4e2a\u5f88\u5178\u578b\u7684Template\u8bbe\u8ba1\u6a21\u5f0f\u7684\u5b9e\u73b0<\/strong>\u554a\u3002SparkWrite\u8fd9\u4e2a\u62bd\u8c61\u7c7b\u5b9a\u4e49\u4e86\u4e00\u4e2a\u7b97\u6cd5\u7684\u9aa8\u67b6\uff08\u6216\u79f0\u201c\u6a21\u677f\u201d\uff09\uff0c\u800c\u5c06\u4e00\u4e9b\u6b65\u9aa4\u5ef6\u8fdf\u5230\u5b50\u7c7b\u4e2d\u5b9e\u73b0\uff08build\u4e2d\u7684\u533f\u540d\u7c7b\uff09\u3002\u6a21\u677f\u65b9\u6cd5\u4f7f\u5f97\u5b50\u7c7b\u53ef\u4ee5\u5728\u4e0d\u6539\u53d8\u7b97\u6cd5\u7ed3\u6784\u7684\u60c5\u51b5\u4e0b\u91cd\u65b0\u5b9a\u4e49\u7b97\u6cd5\u7684\u67d0\u4e9b\u7279\u5b9a\u6b65\u9aa4\u3002<\/p>\n
      createWriter\u4e2dDataWriter\u7684wrapper<\/h2>\n
      \/\/ignore some import\n\nabstract class SparkWrite implements Write, RequiresDistributionAndOrdering {\n  private static final Logger LOG = LoggerFactory.getLogger(SparkWrite.class);\n\n  private final JavaSparkContext sparkContext;\n  private final SparkWriteConf writeConf;\n  private final Table table;\n  private final String queryId;\n  private final FileFormat format;\n  private final String applicationId;\n  private final boolean wapEnabled;\n  private final String wapId;\n  private final int outputSpecId;\n  private final String branch;\n  private final long targetFileSize;\n  private final Schema writeSchema;\n  private final StructType dsSchema;\n  private final Map<String, String> extraSnapshotMetadata;\n  private final boolean useFanoutWriter;\n  private final SparkWriteRequirements writeRequirements;\n  private final Map<String, String> writeProperties;\n\n  private boolean cleanupOnAbort = false;\n\n  SparkWrite(\n      SparkSession spark,\n      Table table,\n      SparkWriteConf writeConf,\n      LogicalWriteInfo writeInfo,\n      String applicationId,\n      Schema writeSchema,\n      StructType dsSchema,\n      SparkWriteRequirements writeRequirements) {\n    this.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext());\n    this.table = table;\n    this.writeConf = writeConf;\n    this.queryId = writeInfo.queryId();\n    this.format = writeConf.dataFileFormat();\n    this.applicationId = applicationId;\n    this.wapEnabled = writeConf.wapEnabled();\n    this.wapId = writeConf.wapId();\n    this.branch = writeConf.branch();\n    this.targetFileSize = writeConf.targetDataFileSize();\n    this.writeSchema = writeSchema;\n    this.dsSchema = dsSchema;\n    this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();\n    this.useFanoutWriter = writeConf.useFanoutWriter(writeRequirements);\n    this.writeRequirements = writeRequirements;\n    this.outputSpecId = writeConf.outputSpecId();\n    this.writeProperties = writeConf.writeProperties();\n  }\n\n  @Override\n  public Distribution requiredDistribution() {\n    Distribution distribution = writeRequirements.distribution();\n    LOG.debug("Requesting {} as write distribution for table {}", distribution, table.name());\n    return distribution;\n  }\n\n  @Override\n  public boolean distributionStrictlyRequired() {\n    return false;\n  }\n\n  @Override\n  public SortOrder[] requiredOrdering() {\n    SortOrder[] ordering = writeRequirements.ordering();\n    LOG.debug("Requesting {} as write ordering for table {}", ordering, table.name());\n    return ordering;\n  }\n\n  @Override\n  public long advisoryPartitionSizeInBytes() {\n    long size = writeRequirements.advisoryPartitionSize();\n    LOG.debug("Requesting {} bytes advisory partition size for table {}", size, table.name());\n    return size;\n  }\n\n  BatchWrite asBatchAppend() {\n    return new BatchAppend();\n  }\n\n  BatchWrite asDynamicOverwrite() {\n    return new DynamicOverwrite();\n  }\n\n  BatchWrite asOverwriteByFilter(Expression overwriteExpr) {\n    return new OverwriteByFilter(overwriteExpr);\n  }\n\n  BatchWrite asCopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {\n    return new CopyOnWriteOperation(scan, isolationLevel);\n  }\n\n  BatchWrite asRewrite(String fileSetID) {\n    return new RewriteFiles(fileSetID);\n  }\n\n  StreamingWrite asStreamingAppend() {\n    return new StreamingAppend();\n  }\n\n  StreamingWrite asStreamingOverwrite() {\n    return new StreamingOverwrite();\n  }\n\n  \/\/ the writer factory works for both batch and streaming\n  private WriterFactory createWriterFactory() {\n    \/\/ broadcast the table metadata as the writer factory will be sent to executors\n    Broadcast<Table> tableBroadcast =\n        sparkContext.broadcast(SerializableTableWithSize.copyOf(table));\n    return new WriterFactory(\n        tableBroadcast,\n        queryId,\n        format,\n        outputSpecId,\n        targetFileSize,\n        writeSchema,\n        dsSchema,\n        useFanoutWriter,\n        writeProperties);\n  }\n\n  private void commitOperation(SnapshotUpdate<?> operation, String description) {\n    LOG.info("Committing {} to table {}", description, table);\n    if (applicationId != null) {\n      operation.set("spark.app.id", applicationId);\n    }\n\n    if (!extraSnapshotMetadata.isEmpty()) {\n      extraSnapshotMetadata.forEach(operation::set);\n    }\n\n    if (!CommitMetadata.commitProperties().isEmpty()) {\n      CommitMetadata.commitProperties().forEach(operation::set);\n    }\n\n    if (wapEnabled && wapId != null) {\n      \/\/ write-audit-publish is enabled for this table and job\n      \/\/ stage the changes without changing the current snapshot\n      operation.set(SnapshotSummary.STAGED_WAP_ID_PROP, wapId);\n      operation.stageOnly();\n    }\n\n    if (branch != null) {\n      operation.toBranch(branch);\n    }\n\n    try {\n      long start = System.currentTimeMillis();\n      operation.commit(); \/\/ abort is automatically called if this fails\n      long duration = System.currentTimeMillis() - start;\n      LOG.info("Committed in {} ms", duration);\n    } catch (Exception e) {\n      cleanupOnAbort = e instanceof CleanableFailure;\n      throw e;\n    }\n  }\n\n  private void abort(WriterCommitMessage[] messages) {\n    if (cleanupOnAbort) {\n      SparkCleanupUtil.deleteFiles("job abort", table.io(), files(messages));\n    } else {\n      LOG.warn("Skipping cleanup of written files");\n    }\n  }\n\n  private List<DataFile> files(WriterCommitMessage[] messages) {\n    List<DataFile> files = Lists.newArrayList();\n\n    for (WriterCommitMessage message : messages) {\n      if (message != null) {\n        TaskCommit taskCommit = (TaskCommit) message;\n        files.addAll(Arrays.asList(taskCommit.files()));\n      }\n    }\n\n    return files;\n  }\n\n  @Override\n  public String toString() {\n    return String.format("IcebergWrite(table=%s, format=%s)", table, format);\n  }\n\n  private abstract class BaseBatchWrite implements BatchWrite {\n    @Override\n    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {\n      return createWriterFactory();\n    }\n\n    @Override\n    public boolean useCommitCoordinator() {\n      return false;\n    }\n\n    @Override\n    public void abort(WriterCommitMessage[] messages) {\n      SparkWrite.this.abort(messages);\n    }\n\n    @Override\n    public String toString() {\n      return String.format("IcebergBatchWrite(table=%s, format=%s)", table, format);\n    }\n  }\n\n  private class BatchAppend extends BaseBatchWrite {\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      AppendFiles append = table.newAppend();\n\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        numFiles += 1;\n        append.appendFile(file);\n      }\n\n      commitOperation(append, String.format("append with %d new data files", numFiles));\n    }\n  }\n\n  private class DynamicOverwrite extends BaseBatchWrite {\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      List<DataFile> files = files(messages);\n\n      if (files.isEmpty()) {\n        LOG.info("Dynamic overwrite is empty, skipping commit");\n        return;\n      }\n\n      ReplacePartitions dynamicOverwrite = table.newReplacePartitions();\n      IsolationLevel isolationLevel = writeConf.isolationLevel();\n      Long validateFromSnapshotId = writeConf.validateFromSnapshotId();\n\n      if (isolationLevel != null && validateFromSnapshotId != null) {\n        dynamicOverwrite.validateFromSnapshot(validateFromSnapshotId);\n      }\n\n      if (isolationLevel == SERIALIZABLE) {\n        dynamicOverwrite.validateNoConflictingData();\n        dynamicOverwrite.validateNoConflictingDeletes();\n\n      } else if (isolationLevel == SNAPSHOT) {\n        dynamicOverwrite.validateNoConflictingDeletes();\n      }\n\n      int numFiles = 0;\n      for (DataFile file : files) {\n        numFiles += 1;\n        dynamicOverwrite.addFile(file);\n      }\n\n      commitOperation(\n          dynamicOverwrite,\n          String.format("dynamic partition overwrite with %d new data files", numFiles));\n    }\n  }\n\n  private class OverwriteByFilter extends BaseBatchWrite {\n    private final Expression overwriteExpr;\n\n    private OverwriteByFilter(Expression overwriteExpr) {\n      this.overwriteExpr = overwriteExpr;\n    }\n\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      OverwriteFiles overwriteFiles = table.newOverwrite();\n      overwriteFiles.overwriteByRowFilter(overwriteExpr);\n\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        numFiles += 1;\n        overwriteFiles.addFile(file);\n      }\n\n      IsolationLevel isolationLevel = writeConf.isolationLevel();\n      Long validateFromSnapshotId = writeConf.validateFromSnapshotId();\n\n      if (isolationLevel != null && validateFromSnapshotId != null) {\n        overwriteFiles.validateFromSnapshot(validateFromSnapshotId);\n      }\n\n      if (isolationLevel == SERIALIZABLE) {\n        overwriteFiles.validateNoConflictingDeletes();\n        overwriteFiles.validateNoConflictingData();\n\n      } else if (isolationLevel == SNAPSHOT) {\n        overwriteFiles.validateNoConflictingDeletes();\n      }\n\n      String commitMsg =\n          String.format("overwrite by filter %s with %d new data files", overwriteExpr, numFiles);\n      commitOperation(overwriteFiles, commitMsg);\n    }\n  }\n\n  private class CopyOnWriteOperation extends BaseBatchWrite {\n    private final SparkCopyOnWriteScan scan;\n    private final IsolationLevel isolationLevel;\n\n    private CopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {\n      this.scan = scan;\n      this.isolationLevel = isolationLevel;\n    }\n\n    private List<DataFile> overwrittenFiles() {\n      if (scan == null) {\n        return ImmutableList.of();\n      } else {\n        return scan.tasks().stream().map(FileScanTask::file).collect(Collectors.toList());\n      }\n    }\n\n    private Expression conflictDetectionFilter() {\n      \/\/ the list of filter expressions may be empty but is never null\n      List<Expression> scanFilterExpressions = scan.filterExpressions();\n\n      Expression filter = Expressions.alwaysTrue();\n\n      for (Expression expr : scanFilterExpressions) {\n        filter = Expressions.and(filter, expr);\n      }\n\n      return filter;\n    }\n\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      OverwriteFiles overwriteFiles = table.newOverwrite();\n\n      List<DataFile> overwrittenFiles = overwrittenFiles();\n      int numOverwrittenFiles = overwrittenFiles.size();\n      for (DataFile overwrittenFile : overwrittenFiles) {\n        overwriteFiles.deleteFile(overwrittenFile);\n      }\n\n      int numAddedFiles = 0;\n      for (DataFile file : files(messages)) {\n        numAddedFiles += 1;\n        overwriteFiles.addFile(file);\n      }\n\n      \/\/ the scan may be null if the optimizer replaces it with an empty relation (e.g. false cond)\n      \/\/ no validation is needed in this case as the command does not depend on the table state\n      if (scan != null) {\n        if (isolationLevel == SERIALIZABLE) {\n          commitWithSerializableIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);\n        } else if (isolationLevel == SNAPSHOT) {\n          commitWithSnapshotIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);\n        } else {\n          throw new IllegalArgumentException("Unsupported isolation level: " + isolationLevel);\n        }\n\n      } else {\n        commitOperation(\n            overwriteFiles,\n            String.format("overwrite with %d new data files (no validation)", numAddedFiles));\n      }\n    }\n\n    private void commitWithSerializableIsolation(\n        OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {\n      Long scanSnapshotId = scan.snapshotId();\n      if (scanSnapshotId != null) {\n        overwriteFiles.validateFromSnapshot(scanSnapshotId);\n      }\n\n      Expression conflictDetectionFilter = conflictDetectionFilter();\n      overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);\n      overwriteFiles.validateNoConflictingData();\n      overwriteFiles.validateNoConflictingDeletes();\n\n      String commitMsg =\n          String.format(\n              "overwrite of %d data files with %d new data files, scanSnapshotId: %d, conflictDetectionFilter: %s",\n              numOverwrittenFiles, numAddedFiles, scanSnapshotId, conflictDetectionFilter);\n      commitOperation(overwriteFiles, commitMsg);\n    }\n\n    private void commitWithSnapshotIsolation(\n        OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {\n      Long scanSnapshotId = scan.snapshotId();\n      if (scanSnapshotId != null) {\n        overwriteFiles.validateFromSnapshot(scanSnapshotId);\n      }\n\n      Expression conflictDetectionFilter = conflictDetectionFilter();\n      overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);\n      overwriteFiles.validateNoConflictingDeletes();\n\n      String commitMsg =\n          String.format(\n              "overwrite of %d data files with %d new data files",\n              numOverwrittenFiles, numAddedFiles);\n      commitOperation(overwriteFiles, commitMsg);\n    }\n  }\n\n  private class RewriteFiles extends BaseBatchWrite {\n    private final String fileSetID;\n\n    private RewriteFiles(String fileSetID) {\n      this.fileSetID = fileSetID;\n    }\n\n    @Override\n    public void commit(WriterCommitMessage[] messages) {\n      FileRewriteCoordinator coordinator = FileRewriteCoordinator.get();\n      coordinator.stageRewrite(table, fileSetID, DataFileSet.of(files(messages)));\n    }\n  }\n\n  private abstract class BaseStreamingWrite implements StreamingWrite {\n    private static final String QUERY_ID_PROPERTY = "spark.sql.streaming.queryId";\n    private static final String EPOCH_ID_PROPERTY = "spark.sql.streaming.epochId";\n\n    protected abstract String mode();\n\n    @Override\n    public StreamingDataWriterFactory createStreamingWriterFactory(PhysicalWriteInfo info) {\n      return createWriterFactory();\n    }\n\n    @Override\n    public boolean useCommitCoordinator() {\n      return false;\n    }\n\n    @Override\n    public final void commit(long epochId, WriterCommitMessage[] messages) {\n      LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, mode());\n\n      table.refresh();\n\n      Long lastCommittedEpochId = findLastCommittedEpochId();\n      if (lastCommittedEpochId != null && epochId <= lastCommittedEpochId) {\n        LOG.info("Skipping epoch {} for query {} as it was already committed", epochId, queryId);\n        return;\n      }\n\n      doCommit(epochId, messages);\n    }\n\n    protected abstract void doCommit(long epochId, WriterCommitMessage[] messages);\n\n    protected <T> void commit(SnapshotUpdate<T> snapshotUpdate, long epochId, String description) {\n      snapshotUpdate.set(QUERY_ID_PROPERTY, queryId);\n      snapshotUpdate.set(EPOCH_ID_PROPERTY, Long.toString(epochId));\n      commitOperation(snapshotUpdate, description);\n    }\n\n    private Long findLastCommittedEpochId() {\n      Snapshot snapshot = table.currentSnapshot();\n      Long lastCommittedEpochId = null;\n      while (snapshot != null) {\n        Map<String, String> summary = snapshot.summary();\n        String snapshotQueryId = summary.get(QUERY_ID_PROPERTY);\n        if (queryId.equals(snapshotQueryId)) {\n          lastCommittedEpochId = Long.valueOf(summary.get(EPOCH_ID_PROPERTY));\n          break;\n        }\n        Long parentSnapshotId = snapshot.parentId();\n        snapshot = parentSnapshotId != null ? table.snapshot(parentSnapshotId) : null;\n      }\n      return lastCommittedEpochId;\n    }\n\n    @Override\n    public void abort(long epochId, WriterCommitMessage[] messages) {\n      SparkWrite.this.abort(messages);\n    }\n\n    @Override\n    public String toString() {\n      return String.format("IcebergStreamingWrite(table=%s, format=%s)", table, format);\n    }\n  }\n\n  private class StreamingAppend extends BaseStreamingWrite {\n    @Override\n    protected String mode() {\n      return "append";\n    }\n\n    @Override\n    protected void doCommit(long epochId, WriterCommitMessage[] messages) {\n      AppendFiles append = table.newFastAppend();\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        append.appendFile(file);\n        numFiles++;\n      }\n      commit(append, epochId, String.format("streaming append with %d new data files", numFiles));\n    }\n  }\n\n  private class StreamingOverwrite extends BaseStreamingWrite {\n    @Override\n    protected String mode() {\n      return "complete";\n    }\n\n    @Override\n    public void doCommit(long epochId, WriterCommitMessage[] messages) {\n      OverwriteFiles overwriteFiles = table.newOverwrite();\n      overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());\n      int numFiles = 0;\n      for (DataFile file : files(messages)) {\n        overwriteFiles.addFile(file);\n        numFiles++;\n      }\n      commit(\n          overwriteFiles,\n          epochId,\n          String.format("streaming complete overwrite with %d new data files", numFiles));\n    }\n  }\n\n  public static class TaskCommit implements WriterCommitMessage {\n    private final DataFile[] taskFiles;\n\n    TaskCommit(DataFile[] taskFiles) {\n      this.taskFiles = taskFiles;\n    }\n\n    \/\/ Reports bytesWritten and recordsWritten to the Spark output metrics.\n    \/\/ Can only be called in executor.\n    void reportOutputMetrics() {\n      long bytesWritten = 0L;\n      long recordsWritten = 0L;\n      for (DataFile dataFile : taskFiles) {\n        bytesWritten += dataFile.fileSizeInBytes();\n        recordsWritten += dataFile.recordCount();\n      }\n\n      TaskContext taskContext = TaskContext$.MODULE$.get();\n      if (taskContext != null) {\n        OutputMetrics outputMetrics = taskContext.taskMetrics().outputMetrics();\n        outputMetrics.setBytesWritten(bytesWritten);\n        outputMetrics.setRecordsWritten(recordsWritten);\n      }\n    }\n\n    DataFile[] files() {\n      return taskFiles;\n    }\n  }\n\n  private static class WriterFactory implements DataWriterFactory, StreamingDataWriterFactory {\n    private final Broadcast<Table> tableBroadcast;\n    private final FileFormat format;\n    private final int outputSpecId;\n    private final long targetFileSize;\n    private final Schema writeSchema;\n    private final StructType dsSchema;\n    private final boolean useFanoutWriter;\n    private final String queryId;\n    private final Map<String, String> writeProperties;\n\n    protected WriterFactory(\n        Broadcast<Table> tableBroadcast,\n        String queryId,\n        FileFormat format,\n        int outputSpecId,\n        long targetFileSize,\n        Schema writeSchema,\n        StructType dsSchema,\n        boolean useFanoutWriter,\n        Map<String, String> writeProperties) {\n      this.tableBroadcast = tableBroadcast;\n      this.format = format;\n      this.outputSpecId = outputSpecId;\n      this.targetFileSize = targetFileSize;\n      this.writeSchema = writeSchema;\n      this.dsSchema = dsSchema;\n      this.useFanoutWriter = useFanoutWriter;\n      this.queryId = queryId;\n      this.writeProperties = writeProperties;\n    }\n\n    @Override\n    public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {\n      return createWriter(partitionId, taskId, 0);\n    }\n\n    @Override\n    public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {\n      Table table = tableBroadcast.value();\n      PartitionSpec spec = table.specs().get(outputSpecId);\n      FileIO io = table.io();\n      String operationId = queryId + "-" + epochId;\n      OutputFileFactory fileFactory =\n          OutputFileFactory.builderFor(table, partitionId, taskId)\n              .format(format)\n              .operationId(operationId)\n              .build();\n      SparkFileWriterFactory writerFactory =\n          SparkFileWriterFactory.builderFor(table)\n              .dataFileFormat(format)\n              .dataSchema(writeSchema)\n              .dataSparkType(dsSchema)\n              .writeProperties(writeProperties)\n              .build();\n\n      if (spec.isUnpartitioned()) {\n        return new UnpartitionedDataWriter(writerFactory, fileFactory, io, spec, targetFileSize);\n\n      } else {\n        return new PartitionedDataWriter(\n            writerFactory,\n            fileFactory,\n            io,\n            spec,\n            writeSchema,\n            dsSchema,\n            targetFileSize,\n            useFanoutWriter);\n      }\n    }\n  }\n\n  private static class UnpartitionedDataWriter implements DataWriter<InternalRow> {\n    private final FileWriter<InternalRow, DataWriteResult> delegate;\n    private final FileIO io;\n\n    private UnpartitionedDataWriter(\n        SparkFileWriterFactory writerFactory,\n        OutputFileFactory fileFactory,\n        FileIO io,\n        PartitionSpec spec,\n        long targetFileSize) {\n      this.delegate =\n          new RollingDataWriter<>(writerFactory, fileFactory, io, targetFileSize, spec, null);\n      this.io = io;\n    }\n\n    @Override\n    public void write(InternalRow record) throws IOException {\n      delegate.write(record);\n    }\n\n    @Override\n    public WriterCommitMessage commit() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));\n      taskCommit.reportOutputMetrics();\n      return taskCommit;\n    }\n\n    @Override\n    public void abort() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());\n    }\n\n    @Override\n    public void close() throws IOException {\n      delegate.close();\n    }\n  }\n\n  private static class PartitionedDataWriter implements DataWriter<InternalRow> {\n    private final PartitioningWriter<InternalRow, DataWriteResult> delegate;\n    private final FileIO io;\n    private final PartitionSpec spec;\n    private final PartitionKey partitionKey;\n    private final InternalRowWrapper internalRowWrapper;\n\n    private PartitionedDataWriter(\n        SparkFileWriterFactory writerFactory,\n        OutputFileFactory fileFactory,\n        FileIO io,\n        PartitionSpec spec,\n        Schema dataSchema,\n        StructType dataSparkType,\n        long targetFileSize,\n        boolean fanoutEnabled) {\n      if (fanoutEnabled) {\n        this.delegate = new FanoutDataWriter<>(writerFactory, fileFactory, io, targetFileSize);\n      } else {\n        this.delegate = new ClusteredDataWriter<>(writerFactory, fileFactory, io, targetFileSize);\n      }\n      this.io = io;\n      this.spec = spec;\n      this.partitionKey = new PartitionKey(spec, dataSchema);\n      this.internalRowWrapper = new InternalRowWrapper(dataSparkType, dataSchema.asStruct());\n    }\n\n    @Override\n    public void write(InternalRow row) throws IOException {\n      partitionKey.partition(internalRowWrapper.wrap(row));\n      delegate.write(row, spec, partitionKey);\n    }\n\n    @Override\n    public WriterCommitMessage commit() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));\n      taskCommit.reportOutputMetrics();\n      return taskCommit;\n    }\n\n    @Override\n    public void abort() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());\n    }\n\n    @Override\n    public void close() throws IOException {\n      delegate.close();\n    }\n  }\n}<\/code><\/pre>\n
      \u8fd9\u91cc\u4f1a\u6839\u636e\u662f\u5426\u5206\u533a\uff0c\u5bf9new\u51fa\u6765\u7684writer\u4f20\u5165\u4e0a\u6e38\u7ed9\u7684\u53c2\u6570\u3002<\/p>\n
      \u8fd9\u4e24\u4e2a\u7c7b\uff0c\u90fd\u5b9e\u73b0\u4e86DataWriter\u7c7b\uff0c\u9488\u5bf9\u5199\u5165\u7684\u7c7b\u578b\u662fInternalRow\uff0c\u8fd9\u662fSparkSQL\u4e2d\u7528\u4e8e\u505a\u6570\u636e\u6d41\u52a8\u7684\u7c7b\u3002\u800c\u4e24\u4e2a\u5bf9\u5e94\u7684\u5b9e\u73b0\u7c7b\uff0c\u5176\u5b9e\u5b9e\u73b0\u90fd\u5f88\u8f7b\u8584\uff1a<\/p>\n
        private static class UnpartitionedDataWriter implements DataWriter<InternalRow> {\n    private final FileWriter<InternalRow, DataWriteResult> delegate;\n    private final FileIO io;\n\n    private UnpartitionedDataWriter(\n        SparkFileWriterFactory writerFactory,\n        OutputFileFactory fileFactory,\n        FileIO io,\n        PartitionSpec spec,\n        long targetFileSize) {\n      this.delegate =\n          new RollingDataWriter<>(writerFactory, fileFactory, io, targetFileSize, spec, null);\n      this.io = io;\n    }\n\n    @Override\n    public void write(InternalRow record) throws IOException {\n      delegate.write(record);\n    }\n\n    @Override\n    public WriterCommitMessage commit() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));\n      taskCommit.reportOutputMetrics();\n      return taskCommit;\n    }\n\n    @Override\n    public void abort() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());\n    }\n\n    @Override\n    public void close() throws IOException {\n      delegate.close();\n    }\n  }\n\n  private static class PartitionedDataWriter implements DataWriter<InternalRow> {\n    private final PartitioningWriter<InternalRow, DataWriteResult> delegate;\n    private final FileIO io;\n    private final PartitionSpec spec;\n    private final PartitionKey partitionKey;\n    private final InternalRowWrapper internalRowWrapper;\n\n    private PartitionedDataWriter(\n        SparkFileWriterFactory writerFactory,\n        OutputFileFactory fileFactory,\n        FileIO io,\n        PartitionSpec spec,\n        Schema dataSchema,\n        StructType dataSparkType,\n        long targetFileSize,\n        boolean fanoutEnabled) {\n      if (fanoutEnabled) {\n        this.delegate = new FanoutDataWriter<>(writerFactory, fileFactory, io, targetFileSize);\n      } else {\n        this.delegate = new ClusteredDataWriter<>(writerFactory, fileFactory, io, targetFileSize);\n      }\n      this.io = io;\n      this.spec = spec;\n      this.partitionKey = new PartitionKey(spec, dataSchema);\n      this.internalRowWrapper = new InternalRowWrapper(dataSparkType, dataSchema.asStruct());\n    }\n\n    @Override\n    public void write(InternalRow row) throws IOException {\n      partitionKey.partition(internalRowWrapper.wrap(row));\n      delegate.write(row, spec, partitionKey);\n    }\n\n    @Override\n    public WriterCommitMessage commit() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));\n      taskCommit.reportOutputMetrics();\n      return taskCommit;\n    }\n\n    @Override\n    public void abort() throws IOException {\n      close();\n\n      DataWriteResult result = delegate.result();\n      SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());\n    }\n\n    @Override\n    public void close() throws IOException {\n      delegate.close();\n    }\n  }<\/code><\/pre>\n
      \u4e00\u770b\u5230delegate\uff0c\u6709\u70b9\u4ee3\u7801\u7ecf\u9a8c\u7684\u540c\u5b66\u90fd\u77e5\u9053\u4e86\u554a\u2014\u2014\u5b83\u624d\u662f\u771f\u6b63\u5e72\u6d3b\u7684\u4eba\uff0c\u50cf\u6781\u4e86\u4f60\u7684\u8001\u677f\u5bf9\u5916\u75af\u72c2\u63fd\u6d3b\u800c\u5728\u5185\u90e8\u75af\u72c2\u6d88\u5316\u9700\u6c42\u7684\u4f60\u3002<\/p>\n
      \u4e24\u8005\u5185\u90e8\u5bf9\u5e94\u7684delegate\u5b9e\u73b0\u7c7b\u4e3a\uff1a<\/p>\n