IT极限技术分享汇本文在绿泡泡“狗哥琐话”首发于2025.9.24 <-关注不走丢。
大家好这里是狗哥。金9银10马上要到了啊，一些机智的同学已经开始提前备战了，近期找我辅导的同学也很多，给我累麻了。
说正事啊。虽然很多大数据开发同学会写Java，但是Coding设计能力（设计模式、SOLID、简洁架构、DDD之类的）是弱于一些传统Java后台同学的，本身的岗位要求以及业务的多变性都会要求他们这么做。而大数据的同学在这方面相对会偏弱一点，挺好理解的，因为核心技能不在这里。
那我怎么帮助这些同学呢？之前不是在星球里剖析了一些开源软件的源码嘛，从Paimon到AutoMQ再到现在的Iceberg。我想这不是个好机会嘛，既然都了解里面的逻辑了，基于理解它们业务之上，再讲设计，不是美滋滋？
那都说到Iceberg了，我们今天就以SparkSQL写入Iceberg的链路，看看有哪些设计啊。
选择代码分支 apache-iceberg-1.9.2
按照国际惯例，代码地图放上。
|-- SparkWrite
-- createWriter #由Spark框架调用，产生的实例中的write方法也是
|-- PartitionedDataWriter
-- write
|-- FanoutDataWriter
-- write #中间绕了一些factory
|-- DataWriter
-- write
|-- FileAppender
-- add #真正的文件写入SparkWrite
package org.apache.iceberg.spark.source;
//ignore some import
abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
private static final Logger LOG = LoggerFactory.getLogger(SparkWrite.class);
private final JavaSparkContext sparkContext;
private final SparkWriteConf writeConf;
private final Table table;
private final String queryId;
private final FileFormat format;
private final String applicationId;
private final boolean wapEnabled;
private final String wapId;
private final int outputSpecId;
private final String branch;
private final long targetFileSize;
private final Schema writeSchema;
private final StructType dsSchema;
private final Map<String, String> extraSnapshotMetadata;
private final boolean useFanoutWriter;
private final SparkWriteRequirements writeRequirements;
private final Map<String, String> writeProperties;
private boolean cleanupOnAbort = false;
SparkWrite(
SparkSession spark,
Table table,
SparkWriteConf writeConf,
LogicalWriteInfo writeInfo,
String applicationId,
Schema writeSchema,
StructType dsSchema,
SparkWriteRequirements writeRequirements) {
this.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext());
this.table = table;
this.writeConf = writeConf;
this.queryId = writeInfo.queryId();
this.format = writeConf.dataFileFormat();
this.applicationId = applicationId;
this.wapEnabled = writeConf.wapEnabled();
this.wapId = writeConf.wapId();
this.branch = writeConf.branch();
this.targetFileSize = writeConf.targetDataFileSize();
this.writeSchema = writeSchema;
this.dsSchema = dsSchema;
this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();
this.useFanoutWriter = writeConf.useFanoutWriter(writeRequirements);
this.writeRequirements = writeRequirements;
this.outputSpecId = writeConf.outputSpecId();
this.writeProperties = writeConf.writeProperties();
}
@Override
public Distribution requiredDistribution() {
Distribution distribution = writeRequirements.distribution();
LOG.debug("Requesting {} as write distribution for table {}", distribution, table.name());
return distribution;
}
@Override
public boolean distributionStrictlyRequired() {
return false;
}
@Override
public SortOrder[] requiredOrdering() {
SortOrder[] ordering = writeRequirements.ordering();
LOG.debug("Requesting {} as write ordering for table {}", ordering, table.name());
return ordering;
}
@Override
public long advisoryPartitionSizeInBytes() {
long size = writeRequirements.advisoryPartitionSize();
LOG.debug("Requesting {} bytes advisory partition size for table {}", size, table.name());
return size;
}
BatchWrite asBatchAppend() {
return new BatchAppend();
}
BatchWrite asDynamicOverwrite() {
return new DynamicOverwrite();
}
BatchWrite asOverwriteByFilter(Expression overwriteExpr) {
return new OverwriteByFilter(overwriteExpr);
}
BatchWrite asCopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {
return new CopyOnWriteOperation(scan, isolationLevel);
}
BatchWrite asRewrite(String fileSetID) {
return new RewriteFiles(fileSetID);
}
StreamingWrite asStreamingAppend() {
return new StreamingAppend();
}
StreamingWrite asStreamingOverwrite() {
return new StreamingOverwrite();
}
// the writer factory works for both batch and streaming
private WriterFactory createWriterFactory() {
// broadcast the table metadata as the writer factory will be sent to executors
Broadcast<Table> tableBroadcast =
sparkContext.broadcast(SerializableTableWithSize.copyOf(table));
return new WriterFactory(
tableBroadcast,
queryId,
format,
outputSpecId,
targetFileSize,
writeSchema,
dsSchema,
useFanoutWriter,
writeProperties);
}
private void commitOperation(SnapshotUpdate<?> operation, String description) {
LOG.info("Committing {} to table {}", description, table);
if (applicationId != null) {
operation.set("spark.app.id", applicationId);
}
if (!extraSnapshotMetadata.isEmpty()) {
extraSnapshotMetadata.forEach(operation::set);
}
if (!CommitMetadata.commitProperties().isEmpty()) {
CommitMetadata.commitProperties().forEach(operation::set);
}
if (wapEnabled && wapId != null) {
// write-audit-publish is enabled for this table and job
// stage the changes without changing the current snapshot
operation.set(SnapshotSummary.STAGED_WAP_ID_PROP, wapId);
operation.stageOnly();
}
if (branch != null) {
operation.toBranch(branch);
}
try {
long start = System.currentTimeMillis();
operation.commit(); // abort is automatically called if this fails
long duration = System.currentTimeMillis() - start;
LOG.info("Committed in {} ms", duration);
} catch (Exception e) {
cleanupOnAbort = e instanceof CleanableFailure;
throw e;
}
}
private void abort(WriterCommitMessage[] messages) {
if (cleanupOnAbort) {
SparkCleanupUtil.deleteFiles("job abort", table.io(), files(messages));
} else {
LOG.warn("Skipping cleanup of written files");
}
}
private List<DataFile> files(WriterCommitMessage[] messages) {
List<DataFile> files = Lists.newArrayList();
for (WriterCommitMessage message : messages) {
if (message != null) {
TaskCommit taskCommit = (TaskCommit) message;
files.addAll(Arrays.asList(taskCommit.files()));
}
}
return files;
}
@Override
public String toString() {
return String.format("IcebergWrite(table=%s, format=%s)", table, format);
}
private abstract class BaseBatchWrite implements BatchWrite {
@Override
public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
return createWriterFactory();
}
@Override
public boolean useCommitCoordinator() {
return false;
}
@Override
public void abort(WriterCommitMessage[] messages) {
SparkWrite.this.abort(messages);
}
@Override
public String toString() {
return String.format("IcebergBatchWrite(table=%s, format=%s)", table, format);
}
}
private class BatchAppend extends BaseBatchWrite {
@Override
public void commit(WriterCommitMessage[] messages) {
AppendFiles append = table.newAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
numFiles += 1;
append.appendFile(file);
}
commitOperation(append, String.format("append with %d new data files", numFiles));
}
}
private class DynamicOverwrite extends BaseBatchWrite {
@Override
public void commit(WriterCommitMessage[] messages) {
List<DataFile> files = files(messages);
if (files.isEmpty()) {
LOG.info("Dynamic overwrite is empty, skipping commit");
return;
}
ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
IsolationLevel isolationLevel = writeConf.isolationLevel();
Long validateFromSnapshotId = writeConf.validateFromSnapshotId();
if (isolationLevel != null && validateFromSnapshotId != null) {
dynamicOverwrite.validateFromSnapshot(validateFromSnapshotId);
}
if (isolationLevel == SERIALIZABLE) {
dynamicOverwrite.validateNoConflictingData();
dynamicOverwrite.validateNoConflictingDeletes();
} else if (isolationLevel == SNAPSHOT) {
dynamicOverwrite.validateNoConflictingDeletes();
}
int numFiles = 0;
for (DataFile file : files) {
numFiles += 1;
dynamicOverwrite.addFile(file);
}
commitOperation(
dynamicOverwrite,
String.format("dynamic partition overwrite with %d new data files", numFiles));
}
}
private class OverwriteByFilter extends BaseBatchWrite {
private final Expression overwriteExpr;
private OverwriteByFilter(Expression overwriteExpr) {
this.overwriteExpr = overwriteExpr;
}
@Override
public void commit(WriterCommitMessage[] messages) {
OverwriteFiles overwriteFiles = table.newOverwrite();
overwriteFiles.overwriteByRowFilter(overwriteExpr);
int numFiles = 0;
for (DataFile file : files(messages)) {
numFiles += 1;
overwriteFiles.addFile(file);
}
IsolationLevel isolationLevel = writeConf.isolationLevel();
Long validateFromSnapshotId = writeConf.validateFromSnapshotId();
if (isolationLevel != null && validateFromSnapshotId != null) {
overwriteFiles.validateFromSnapshot(validateFromSnapshotId);
}
if (isolationLevel == SERIALIZABLE) {
overwriteFiles.validateNoConflictingDeletes();
overwriteFiles.validateNoConflictingData();
} else if (isolationLevel == SNAPSHOT) {
overwriteFiles.validateNoConflictingDeletes();
}
String commitMsg =
String.format("overwrite by filter %s with %d new data files", overwriteExpr, numFiles);
commitOperation(overwriteFiles, commitMsg);
}
}
private class CopyOnWriteOperation extends BaseBatchWrite {
private final SparkCopyOnWriteScan scan;
private final IsolationLevel isolationLevel;
private CopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {
this.scan = scan;
this.isolationLevel = isolationLevel;
}
private List<DataFile> overwrittenFiles() {
if (scan == null) {
return ImmutableList.of();
} else {
return scan.tasks().stream().map(FileScanTask::file).collect(Collectors.toList());
}
}
private Expression conflictDetectionFilter() {
// the list of filter expressions may be empty but is never null
List<Expression> scanFilterExpressions = scan.filterExpressions();
Expression filter = Expressions.alwaysTrue();
for (Expression expr : scanFilterExpressions) {
filter = Expressions.and(filter, expr);
}
return filter;
}
@Override
public void commit(WriterCommitMessage[] messages) {
OverwriteFiles overwriteFiles = table.newOverwrite();
List<DataFile> overwrittenFiles = overwrittenFiles();
int numOverwrittenFiles = overwrittenFiles.size();
for (DataFile overwrittenFile : overwrittenFiles) {
overwriteFiles.deleteFile(overwrittenFile);
}
int numAddedFiles = 0;
for (DataFile file : files(messages)) {
numAddedFiles += 1;
overwriteFiles.addFile(file);
}
// the scan may be null if the optimizer replaces it with an empty relation (e.g. false cond)
// no validation is needed in this case as the command does not depend on the table state
if (scan != null) {
if (isolationLevel == SERIALIZABLE) {
commitWithSerializableIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
} else if (isolationLevel == SNAPSHOT) {
commitWithSnapshotIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
} else {
throw new IllegalArgumentException("Unsupported isolation level: " + isolationLevel);
}
} else {
commitOperation(
overwriteFiles,
String.format("overwrite with %d new data files (no validation)", numAddedFiles));
}
}
private void commitWithSerializableIsolation(
OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {
Long scanSnapshotId = scan.snapshotId();
if (scanSnapshotId != null) {
overwriteFiles.validateFromSnapshot(scanSnapshotId);
}
Expression conflictDetectionFilter = conflictDetectionFilter();
overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);
overwriteFiles.validateNoConflictingData();
overwriteFiles.validateNoConflictingDeletes();
String commitMsg =
String.format(
"overwrite of %d data files with %d new data files, scanSnapshotId: %d, conflictDetectionFilter: %s",
numOverwrittenFiles, numAddedFiles, scanSnapshotId, conflictDetectionFilter);
commitOperation(overwriteFiles, commitMsg);
}
private void commitWithSnapshotIsolation(
OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {
Long scanSnapshotId = scan.snapshotId();
if (scanSnapshotId != null) {
overwriteFiles.validateFromSnapshot(scanSnapshotId);
}
Expression conflictDetectionFilter = conflictDetectionFilter();
overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);
overwriteFiles.validateNoConflictingDeletes();
String commitMsg =
String.format(
"overwrite of %d data files with %d new data files",
numOverwrittenFiles, numAddedFiles);
commitOperation(overwriteFiles, commitMsg);
}
}
private class RewriteFiles extends BaseBatchWrite {
private final String fileSetID;
private RewriteFiles(String fileSetID) {
this.fileSetID = fileSetID;
}
@Override
public void commit(WriterCommitMessage[] messages) {
FileRewriteCoordinator coordinator = FileRewriteCoordinator.get();
coordinator.stageRewrite(table, fileSetID, DataFileSet.of(files(messages)));
}
}
private abstract class BaseStreamingWrite implements StreamingWrite {
private static final String QUERY_ID_PROPERTY = "spark.sql.streaming.queryId";
private static final String EPOCH_ID_PROPERTY = "spark.sql.streaming.epochId";
protected abstract String mode();
@Override
public StreamingDataWriterFactory createStreamingWriterFactory(PhysicalWriteInfo info) {
return createWriterFactory();
}
@Override
public boolean useCommitCoordinator() {
return false;
}
@Override
public final void commit(long epochId, WriterCommitMessage[] messages) {
LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, mode());
table.refresh();
Long lastCommittedEpochId = findLastCommittedEpochId();
if (lastCommittedEpochId != null && epochId <= lastCommittedEpochId) {
LOG.info("Skipping epoch {} for query {} as it was already committed", epochId, queryId);
return;
}
doCommit(epochId, messages);
}
protected abstract void doCommit(long epochId, WriterCommitMessage[] messages);
protected <T> void commit(SnapshotUpdate<T> snapshotUpdate, long epochId, String description) {
snapshotUpdate.set(QUERY_ID_PROPERTY, queryId);
snapshotUpdate.set(EPOCH_ID_PROPERTY, Long.toString(epochId));
commitOperation(snapshotUpdate, description);
}
private Long findLastCommittedEpochId() {
Snapshot snapshot = table.currentSnapshot();
Long lastCommittedEpochId = null;
while (snapshot != null) {
Map<String, String> summary = snapshot.summary();
String snapshotQueryId = summary.get(QUERY_ID_PROPERTY);
if (queryId.equals(snapshotQueryId)) {
lastCommittedEpochId = Long.valueOf(summary.get(EPOCH_ID_PROPERTY));
break;
}
Long parentSnapshotId = snapshot.parentId();
snapshot = parentSnapshotId != null ? table.snapshot(parentSnapshotId) : null;
}
return lastCommittedEpochId;
}
@Override
public void abort(long epochId, WriterCommitMessage[] messages) {
SparkWrite.this.abort(messages);
}
@Override
public String toString() {
return String.format("IcebergStreamingWrite(table=%s, format=%s)", table, format);
}
}
private class StreamingAppend extends BaseStreamingWrite {
@Override
protected String mode() {
return "append";
}
@Override
protected void doCommit(long epochId, WriterCommitMessage[] messages) {
AppendFiles append = table.newFastAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
append.appendFile(file);
numFiles++;
}
commit(append, epochId, String.format("streaming append with %d new data files", numFiles));
}
}
private class StreamingOverwrite extends BaseStreamingWrite {
@Override
protected String mode() {
return "complete";
}
@Override
public void doCommit(long epochId, WriterCommitMessage[] messages) {
OverwriteFiles overwriteFiles = table.newOverwrite();
overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());
int numFiles = 0;
for (DataFile file : files(messages)) {
overwriteFiles.addFile(file);
numFiles++;
}
commit(
overwriteFiles,
epochId,
String.format("streaming complete overwrite with %d new data files", numFiles));
}
}
public static class TaskCommit implements WriterCommitMessage {
private final DataFile[] taskFiles;
TaskCommit(DataFile[] taskFiles) {
this.taskFiles = taskFiles;
}
// Reports bytesWritten and recordsWritten to the Spark output metrics.
// Can only be called in executor.
void reportOutputMetrics() {
long bytesWritten = 0L;
long recordsWritten = 0L;
for (DataFile dataFile : taskFiles) {
bytesWritten += dataFile.fileSizeInBytes();
recordsWritten += dataFile.recordCount();
}
TaskContext taskContext = TaskContext$.MODULE$.get();
if (taskContext != null) {
OutputMetrics outputMetrics = taskContext.taskMetrics().outputMetrics();
outputMetrics.setBytesWritten(bytesWritten);
outputMetrics.setRecordsWritten(recordsWritten);
}
}
DataFile[] files() {
return taskFiles;
}
}
private static class WriterFactory implements DataWriterFactory, StreamingDataWriterFactory {
private final Broadcast<Table> tableBroadcast;
private final FileFormat format;
private final int outputSpecId;
private final long targetFileSize;
private final Schema writeSchema;
private final StructType dsSchema;
private final boolean useFanoutWriter;
private final String queryId;
private final Map<String, String> writeProperties;
protected WriterFactory(
Broadcast<Table> tableBroadcast,
String queryId,
FileFormat format,
int outputSpecId,
long targetFileSize,
Schema writeSchema,
StructType dsSchema,
boolean useFanoutWriter,
Map<String, String> writeProperties) {
this.tableBroadcast = tableBroadcast;
this.format = format;
this.outputSpecId = outputSpecId;
this.targetFileSize = targetFileSize;
this.writeSchema = writeSchema;
this.dsSchema = dsSchema;
this.useFanoutWriter = useFanoutWriter;
this.queryId = queryId;
this.writeProperties = writeProperties;
}
@Override
public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
return createWriter(partitionId, taskId, 0);
}
@Override
public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
Table table = tableBroadcast.value();
PartitionSpec spec = table.specs().get(outputSpecId);
FileIO io = table.io();
String operationId = queryId + "-" + epochId;
OutputFileFactory fileFactory =
OutputFileFactory.builderFor(table, partitionId, taskId)
.format(format)
.operationId(operationId)
.build();
SparkFileWriterFactory writerFactory =
SparkFileWriterFactory.builderFor(table)
.dataFileFormat(format)
.dataSchema(writeSchema)
.dataSparkType(dsSchema)
.writeProperties(writeProperties)
.build();
if (spec.isUnpartitioned()) {
return new UnpartitionedDataWriter(writerFactory, fileFactory, io, spec, targetFileSize);
} else {
return new PartitionedDataWriter(
writerFactory,
fileFactory,
io,
spec,
writeSchema,
dsSchema,
targetFileSize,
useFanoutWriter);
}
}
}
private static class UnpartitionedDataWriter implements DataWriter<InternalRow> {
private final FileWriter<InternalRow, DataWriteResult> delegate;
private final FileIO io;
private UnpartitionedDataWriter(
SparkFileWriterFactory writerFactory,
OutputFileFactory fileFactory,
FileIO io,
PartitionSpec spec,
long targetFileSize) {
this.delegate =
new RollingDataWriter<>(writerFactory, fileFactory, io, targetFileSize, spec, null);
this.io = io;
}
@Override
public void write(InternalRow record) throws IOException {
delegate.write(record);
}
@Override
public WriterCommitMessage commit() throws IOException {
close();
DataWriteResult result = delegate.result();
TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));
taskCommit.reportOutputMetrics();
return taskCommit;
}
@Override
public void abort() throws IOException {
close();
DataWriteResult result = delegate.result();
SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());
}
@Override
public void close() throws IOException {
delegate.close();
}
}
private static class PartitionedDataWriter implements DataWriter<InternalRow> {
private final PartitioningWriter<InternalRow, DataWriteResult> delegate;
private final FileIO io;
private final PartitionSpec spec;
private final PartitionKey partitionKey;
private final InternalRowWrapper internalRowWrapper;
private PartitionedDataWriter(
SparkFileWriterFactory writerFactory,
OutputFileFactory fileFactory,
FileIO io,
PartitionSpec spec,
Schema dataSchema,
StructType dataSparkType,
long targetFileSize,
boolean fanoutEnabled) {
if (fanoutEnabled) {
this.delegate = new FanoutDataWriter<>(writerFactory, fileFactory, io, targetFileSize);
} else {
this.delegate = new ClusteredDataWriter<>(writerFactory, fileFactory, io, targetFileSize);
}
this.io = io;
this.spec = spec;
this.partitionKey = new PartitionKey(spec, dataSchema);
this.internalRowWrapper = new InternalRowWrapper(dataSparkType, dataSchema.asStruct());
}
@Override
public void write(InternalRow row) throws IOException {
partitionKey.partition(internalRowWrapper.wrap(row));
delegate.write(row, spec, partitionKey);
}
@Override
public WriterCommitMessage commit() throws IOException {
close();
DataWriteResult result = delegate.result();
TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));
taskCommit.reportOutputMetrics();
return taskCommit;
}
@Override
public void abort() throws IOException {
close();
DataWriteResult result = delegate.result();
SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());
}
@Override
public void close() throws IOException {
delegate.close();
}
}
}SparkWrite是SparkSQL写入数据的入口啊，然后从代码上，我们不难看出它是一个抽象类啊。那它的实现在哪里呢？
@Override
public Write build() {
// Validate
Schema writeSchema = validateOrMergeWriteSchema(table, dsSchema, writeConf);
SparkUtil.validatePartitionTransforms(table.spec());
// Get application id
String appId = spark.sparkContext().applicationId();
return new SparkWrite(
spark, table, writeConf, writeInfo, appId, writeSchema, dsSchema, writeRequirements()) {
@Override
public BatchWrite toBatch() {
if (rewrittenFileSetId != null) {
return asRewrite(rewrittenFileSetId);
} else if (overwriteByFilter) {
return asOverwriteByFilter(overwriteExpr);
} else if (overwriteDynamic) {
return asDynamicOverwrite();
} else if (overwriteFiles) {
return asCopyOnWriteOperation(copyOnWriteScan, copyOnWriteIsolationLevel);
} else {
return asBatchAppend();
}
}
@Override
public StreamingWrite toStreaming() {
Preconditions.checkState(
!overwriteDynamic, "Unsupported streaming operation: dynamic partition overwrite");
Preconditions.checkState(
!overwriteByFilter || overwriteExpr == Expressions.alwaysTrue(),
"Unsupported streaming operation: overwrite by filter: %s",
overwriteExpr);
Preconditions.checkState(
rewrittenFileSetId == null, "Unsupported streaming operation: rewrite");
if (overwriteByFilter) {
return asStreamingOverwrite();
} else {
return asStreamingAppend();
}
}
};
}在SparkWriteBuilder的build方法中根据配置构建出来的。
这里会根据build前传入的不同属性，构建不同的Write。
toBatch() 方法。根据不同的写入模式返回相应的批处理写入器：
- 重写模式: asRewrite(rewrittenFileSetId) – 用于重写现有文件
- 过滤条件覆盖: asOverwriteByFilter(overwriteExpr) – 根据过滤条件覆盖数据
- 动态分区覆盖: asDynamicOverwrite() – 动态覆盖分区数据
- 文件级别覆盖: asCopyOnWriteOperation(copyOnWriteScan, copyOnWriteIsolationLevel) – Copy-on-write 操作
- 追加模式: asBatchAppend() – 默认的追加写入
toStreaming() 方法
- 用于流式写入，根据配置决定是流式追加还是流式覆盖。
这里其实是一个很典型的Template设计模式的实现啊。SparkWrite这个抽象类定义了一个算法的骨架（或称“模板”），而将一些步骤延迟到子类中实现（build中的匿名类）。模板方法使得子类可以在不改变算法结构的情况下重新定义算法的某些特定步骤。
createWriter中DataWriter的wrapper
//ignore some import
abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
private static final Logger LOG = LoggerFactory.getLogger(SparkWrite.class);
private final JavaSparkContext sparkContext;
private final SparkWriteConf writeConf;
private final Table table;
private final String queryId;
private final FileFormat format;
private final String applicationId;
private final boolean wapEnabled;
private final String wapId;
private final int outputSpecId;
private final String branch;
private final long targetFileSize;
private final Schema writeSchema;
private final StructType dsSchema;
private final Map<String, String> extraSnapshotMetadata;
private final boolean useFanoutWriter;
private final SparkWriteRequirements writeRequirements;
private final Map<String, String> writeProperties;
private boolean cleanupOnAbort = false;
SparkWrite(
SparkSession spark,
Table table,
SparkWriteConf writeConf,
LogicalWriteInfo writeInfo,
String applicationId,
Schema writeSchema,
StructType dsSchema,
SparkWriteRequirements writeRequirements) {
this.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext());
this.table = table;
this.writeConf = writeConf;
this.queryId = writeInfo.queryId();
this.format = writeConf.dataFileFormat();
this.applicationId = applicationId;
this.wapEnabled = writeConf.wapEnabled();
this.wapId = writeConf.wapId();
this.branch = writeConf.branch();
this.targetFileSize = writeConf.targetDataFileSize();
this.writeSchema = writeSchema;
this.dsSchema = dsSchema;
this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();
this.useFanoutWriter = writeConf.useFanoutWriter(writeRequirements);
this.writeRequirements = writeRequirements;
this.outputSpecId = writeConf.outputSpecId();
this.writeProperties = writeConf.writeProperties();
}
@Override
public Distribution requiredDistribution() {
Distribution distribution = writeRequirements.distribution();
LOG.debug("Requesting {} as write distribution for table {}", distribution, table.name());
return distribution;
}
@Override
public boolean distributionStrictlyRequired() {
return false;
}
@Override
public SortOrder[] requiredOrdering() {
SortOrder[] ordering = writeRequirements.ordering();
LOG.debug("Requesting {} as write ordering for table {}", ordering, table.name());
return ordering;
}
@Override
public long advisoryPartitionSizeInBytes() {
long size = writeRequirements.advisoryPartitionSize();
LOG.debug("Requesting {} bytes advisory partition size for table {}", size, table.name());
return size;
}
BatchWrite asBatchAppend() {
return new BatchAppend();
}
BatchWrite asDynamicOverwrite() {
return new DynamicOverwrite();
}
BatchWrite asOverwriteByFilter(Expression overwriteExpr) {
return new OverwriteByFilter(overwriteExpr);
}
BatchWrite asCopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {
return new CopyOnWriteOperation(scan, isolationLevel);
}
BatchWrite asRewrite(String fileSetID) {
return new RewriteFiles(fileSetID);
}
StreamingWrite asStreamingAppend() {
return new StreamingAppend();
}
StreamingWrite asStreamingOverwrite() {
return new StreamingOverwrite();
}
// the writer factory works for both batch and streaming
private WriterFactory createWriterFactory() {
// broadcast the table metadata as the writer factory will be sent to executors
Broadcast<Table> tableBroadcast =
sparkContext.broadcast(SerializableTableWithSize.copyOf(table));
return new WriterFactory(
tableBroadcast,
queryId,
format,
outputSpecId,
targetFileSize,
writeSchema,
dsSchema,
useFanoutWriter,
writeProperties);
}
private void commitOperation(SnapshotUpdate<?> operation, String description) {
LOG.info("Committing {} to table {}", description, table);
if (applicationId != null) {
operation.set("spark.app.id", applicationId);
}
if (!extraSnapshotMetadata.isEmpty()) {
extraSnapshotMetadata.forEach(operation::set);
}
if (!CommitMetadata.commitProperties().isEmpty()) {
CommitMetadata.commitProperties().forEach(operation::set);
}
if (wapEnabled && wapId != null) {
// write-audit-publish is enabled for this table and job
// stage the changes without changing the current snapshot
operation.set(SnapshotSummary.STAGED_WAP_ID_PROP, wapId);
operation.stageOnly();
}
if (branch != null) {
operation.toBranch(branch);
}
try {
long start = System.currentTimeMillis();
operation.commit(); // abort is automatically called if this fails
long duration = System.currentTimeMillis() - start;
LOG.info("Committed in {} ms", duration);
} catch (Exception e) {
cleanupOnAbort = e instanceof CleanableFailure;
throw e;
}
}
private void abort(WriterCommitMessage[] messages) {
if (cleanupOnAbort) {
SparkCleanupUtil.deleteFiles("job abort", table.io(), files(messages));
} else {
LOG.warn("Skipping cleanup of written files");
}
}
private List<DataFile> files(WriterCommitMessage[] messages) {
List<DataFile> files = Lists.newArrayList();
for (WriterCommitMessage message : messages) {
if (message != null) {
TaskCommit taskCommit = (TaskCommit) message;
files.addAll(Arrays.asList(taskCommit.files()));
}
}
return files;
}
@Override
public String toString() {
return String.format("IcebergWrite(table=%s, format=%s)", table, format);
}
private abstract class BaseBatchWrite implements BatchWrite {
@Override
public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
return createWriterFactory();
}
@Override
public boolean useCommitCoordinator() {
return false;
}
@Override
public void abort(WriterCommitMessage[] messages) {
SparkWrite.this.abort(messages);
}
@Override
public String toString() {
return String.format("IcebergBatchWrite(table=%s, format=%s)", table, format);
}
}
private class BatchAppend extends BaseBatchWrite {
@Override
public void commit(WriterCommitMessage[] messages) {
AppendFiles append = table.newAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
numFiles += 1;
append.appendFile(file);
}
commitOperation(append, String.format("append with %d new data files", numFiles));
}
}
private class DynamicOverwrite extends BaseBatchWrite {
@Override
public void commit(WriterCommitMessage[] messages) {
List<DataFile> files = files(messages);
if (files.isEmpty()) {
LOG.info("Dynamic overwrite is empty, skipping commit");
return;
}
ReplacePartitions dynamicOverwrite = table.newReplacePartitions();
IsolationLevel isolationLevel = writeConf.isolationLevel();
Long validateFromSnapshotId = writeConf.validateFromSnapshotId();
if (isolationLevel != null && validateFromSnapshotId != null) {
dynamicOverwrite.validateFromSnapshot(validateFromSnapshotId);
}
if (isolationLevel == SERIALIZABLE) {
dynamicOverwrite.validateNoConflictingData();
dynamicOverwrite.validateNoConflictingDeletes();
} else if (isolationLevel == SNAPSHOT) {
dynamicOverwrite.validateNoConflictingDeletes();
}
int numFiles = 0;
for (DataFile file : files) {
numFiles += 1;
dynamicOverwrite.addFile(file);
}
commitOperation(
dynamicOverwrite,
String.format("dynamic partition overwrite with %d new data files", numFiles));
}
}
private class OverwriteByFilter extends BaseBatchWrite {
private final Expression overwriteExpr;
private OverwriteByFilter(Expression overwriteExpr) {
this.overwriteExpr = overwriteExpr;
}
@Override
public void commit(WriterCommitMessage[] messages) {
OverwriteFiles overwriteFiles = table.newOverwrite();
overwriteFiles.overwriteByRowFilter(overwriteExpr);
int numFiles = 0;
for (DataFile file : files(messages)) {
numFiles += 1;
overwriteFiles.addFile(file);
}
IsolationLevel isolationLevel = writeConf.isolationLevel();
Long validateFromSnapshotId = writeConf.validateFromSnapshotId();
if (isolationLevel != null && validateFromSnapshotId != null) {
overwriteFiles.validateFromSnapshot(validateFromSnapshotId);
}
if (isolationLevel == SERIALIZABLE) {
overwriteFiles.validateNoConflictingDeletes();
overwriteFiles.validateNoConflictingData();
} else if (isolationLevel == SNAPSHOT) {
overwriteFiles.validateNoConflictingDeletes();
}
String commitMsg =
String.format("overwrite by filter %s with %d new data files", overwriteExpr, numFiles);
commitOperation(overwriteFiles, commitMsg);
}
}
private class CopyOnWriteOperation extends BaseBatchWrite {
private final SparkCopyOnWriteScan scan;
private final IsolationLevel isolationLevel;
private CopyOnWriteOperation(SparkCopyOnWriteScan scan, IsolationLevel isolationLevel) {
this.scan = scan;
this.isolationLevel = isolationLevel;
}
private List<DataFile> overwrittenFiles() {
if (scan == null) {
return ImmutableList.of();
} else {
return scan.tasks().stream().map(FileScanTask::file).collect(Collectors.toList());
}
}
private Expression conflictDetectionFilter() {
// the list of filter expressions may be empty but is never null
List<Expression> scanFilterExpressions = scan.filterExpressions();
Expression filter = Expressions.alwaysTrue();
for (Expression expr : scanFilterExpressions) {
filter = Expressions.and(filter, expr);
}
return filter;
}
@Override
public void commit(WriterCommitMessage[] messages) {
OverwriteFiles overwriteFiles = table.newOverwrite();
List<DataFile> overwrittenFiles = overwrittenFiles();
int numOverwrittenFiles = overwrittenFiles.size();
for (DataFile overwrittenFile : overwrittenFiles) {
overwriteFiles.deleteFile(overwrittenFile);
}
int numAddedFiles = 0;
for (DataFile file : files(messages)) {
numAddedFiles += 1;
overwriteFiles.addFile(file);
}
// the scan may be null if the optimizer replaces it with an empty relation (e.g. false cond)
// no validation is needed in this case as the command does not depend on the table state
if (scan != null) {
if (isolationLevel == SERIALIZABLE) {
commitWithSerializableIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
} else if (isolationLevel == SNAPSHOT) {
commitWithSnapshotIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
} else {
throw new IllegalArgumentException("Unsupported isolation level: " + isolationLevel);
}
} else {
commitOperation(
overwriteFiles,
String.format("overwrite with %d new data files (no validation)", numAddedFiles));
}
}
private void commitWithSerializableIsolation(
OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {
Long scanSnapshotId = scan.snapshotId();
if (scanSnapshotId != null) {
overwriteFiles.validateFromSnapshot(scanSnapshotId);
}
Expression conflictDetectionFilter = conflictDetectionFilter();
overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);
overwriteFiles.validateNoConflictingData();
overwriteFiles.validateNoConflictingDeletes();
String commitMsg =
String.format(
"overwrite of %d data files with %d new data files, scanSnapshotId: %d, conflictDetectionFilter: %s",
numOverwrittenFiles, numAddedFiles, scanSnapshotId, conflictDetectionFilter);
commitOperation(overwriteFiles, commitMsg);
}
private void commitWithSnapshotIsolation(
OverwriteFiles overwriteFiles, int numOverwrittenFiles, int numAddedFiles) {
Long scanSnapshotId = scan.snapshotId();
if (scanSnapshotId != null) {
overwriteFiles.validateFromSnapshot(scanSnapshotId);
}
Expression conflictDetectionFilter = conflictDetectionFilter();
overwriteFiles.conflictDetectionFilter(conflictDetectionFilter);
overwriteFiles.validateNoConflictingDeletes();
String commitMsg =
String.format(
"overwrite of %d data files with %d new data files",
numOverwrittenFiles, numAddedFiles);
commitOperation(overwriteFiles, commitMsg);
}
}
private class RewriteFiles extends BaseBatchWrite {
private final String fileSetID;
private RewriteFiles(String fileSetID) {
this.fileSetID = fileSetID;
}
@Override
public void commit(WriterCommitMessage[] messages) {
FileRewriteCoordinator coordinator = FileRewriteCoordinator.get();
coordinator.stageRewrite(table, fileSetID, DataFileSet.of(files(messages)));
}
}
private abstract class BaseStreamingWrite implements StreamingWrite {
private static final String QUERY_ID_PROPERTY = "spark.sql.streaming.queryId";
private static final String EPOCH_ID_PROPERTY = "spark.sql.streaming.epochId";
protected abstract String mode();
@Override
public StreamingDataWriterFactory createStreamingWriterFactory(PhysicalWriteInfo info) {
return createWriterFactory();
}
@Override
public boolean useCommitCoordinator() {
return false;
}
@Override
public final void commit(long epochId, WriterCommitMessage[] messages) {
LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, mode());
table.refresh();
Long lastCommittedEpochId = findLastCommittedEpochId();
if (lastCommittedEpochId != null && epochId <= lastCommittedEpochId) {
LOG.info("Skipping epoch {} for query {} as it was already committed", epochId, queryId);
return;
}
doCommit(epochId, messages);
}
protected abstract void doCommit(long epochId, WriterCommitMessage[] messages);
protected <T> void commit(SnapshotUpdate<T> snapshotUpdate, long epochId, String description) {
snapshotUpdate.set(QUERY_ID_PROPERTY, queryId);
snapshotUpdate.set(EPOCH_ID_PROPERTY, Long.toString(epochId));
commitOperation(snapshotUpdate, description);
}
private Long findLastCommittedEpochId() {
Snapshot snapshot = table.currentSnapshot();
Long lastCommittedEpochId = null;
while (snapshot != null) {
Map<String, String> summary = snapshot.summary();
String snapshotQueryId = summary.get(QUERY_ID_PROPERTY);
if (queryId.equals(snapshotQueryId)) {
lastCommittedEpochId = Long.valueOf(summary.get(EPOCH_ID_PROPERTY));
break;
}
Long parentSnapshotId = snapshot.parentId();
snapshot = parentSnapshotId != null ? table.snapshot(parentSnapshotId) : null;
}
return lastCommittedEpochId;
}
@Override
public void abort(long epochId, WriterCommitMessage[] messages) {
SparkWrite.this.abort(messages);
}
@Override
public String toString() {
return String.format("IcebergStreamingWrite(table=%s, format=%s)", table, format);
}
}
private class StreamingAppend extends BaseStreamingWrite {
@Override
protected String mode() {
return "append";
}
@Override
protected void doCommit(long epochId, WriterCommitMessage[] messages) {
AppendFiles append = table.newFastAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
append.appendFile(file);
numFiles++;
}
commit(append, epochId, String.format("streaming append with %d new data files", numFiles));
}
}
private class StreamingOverwrite extends BaseStreamingWrite {
@Override
protected String mode() {
return "complete";
}
@Override
public void doCommit(long epochId, WriterCommitMessage[] messages) {
OverwriteFiles overwriteFiles = table.newOverwrite();
overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());
int numFiles = 0;
for (DataFile file : files(messages)) {
overwriteFiles.addFile(file);
numFiles++;
}
commit(
overwriteFiles,
epochId,
String.format("streaming complete overwrite with %d new data files", numFiles));
}
}
public static class TaskCommit implements WriterCommitMessage {
private final DataFile[] taskFiles;
TaskCommit(DataFile[] taskFiles) {
this.taskFiles = taskFiles;
}
// Reports bytesWritten and recordsWritten to the Spark output metrics.
// Can only be called in executor.
void reportOutputMetrics() {
long bytesWritten = 0L;
long recordsWritten = 0L;
for (DataFile dataFile : taskFiles) {
bytesWritten += dataFile.fileSizeInBytes();
recordsWritten += dataFile.recordCount();
}
TaskContext taskContext = TaskContext$.MODULE$.get();
if (taskContext != null) {
OutputMetrics outputMetrics = taskContext.taskMetrics().outputMetrics();
outputMetrics.setBytesWritten(bytesWritten);
outputMetrics.setRecordsWritten(recordsWritten);
}
}
DataFile[] files() {
return taskFiles;
}
}
private static class WriterFactory implements DataWriterFactory, StreamingDataWriterFactory {
private final Broadcast<Table> tableBroadcast;
private final FileFormat format;
private final int outputSpecId;
private final long targetFileSize;
private final Schema writeSchema;
private final StructType dsSchema;
private final boolean useFanoutWriter;
private final String queryId;
private final Map<String, String> writeProperties;
protected WriterFactory(
Broadcast<Table> tableBroadcast,
String queryId,
FileFormat format,
int outputSpecId,
long targetFileSize,
Schema writeSchema,
StructType dsSchema,
boolean useFanoutWriter,
Map<String, String> writeProperties) {
this.tableBroadcast = tableBroadcast;
this.format = format;
this.outputSpecId = outputSpecId;
this.targetFileSize = targetFileSize;
this.writeSchema = writeSchema;
this.dsSchema = dsSchema;
this.useFanoutWriter = useFanoutWriter;
this.queryId = queryId;
this.writeProperties = writeProperties;
}
@Override
public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
return createWriter(partitionId, taskId, 0);
}
@Override
public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
Table table = tableBroadcast.value();
PartitionSpec spec = table.specs().get(outputSpecId);
FileIO io = table.io();
String operationId = queryId + "-" + epochId;
OutputFileFactory fileFactory =
OutputFileFactory.builderFor(table, partitionId, taskId)
.format(format)
.operationId(operationId)
.build();
SparkFileWriterFactory writerFactory =
SparkFileWriterFactory.builderFor(table)
.dataFileFormat(format)
.dataSchema(writeSchema)
.dataSparkType(dsSchema)
.writeProperties(writeProperties)
.build();
if (spec.isUnpartitioned()) {
return new UnpartitionedDataWriter(writerFactory, fileFactory, io, spec, targetFileSize);
} else {
return new PartitionedDataWriter(
writerFactory,
fileFactory,
io,
spec,
writeSchema,
dsSchema,
targetFileSize,
useFanoutWriter);
}
}
}
private static class UnpartitionedDataWriter implements DataWriter<InternalRow> {
private final FileWriter<InternalRow, DataWriteResult> delegate;
private final FileIO io;
private UnpartitionedDataWriter(
SparkFileWriterFactory writerFactory,
OutputFileFactory fileFactory,
FileIO io,
PartitionSpec spec,
long targetFileSize) {
this.delegate =
new RollingDataWriter<>(writerFactory, fileFactory, io, targetFileSize, spec, null);
this.io = io;
}
@Override
public void write(InternalRow record) throws IOException {
delegate.write(record);
}
@Override
public WriterCommitMessage commit() throws IOException {
close();
DataWriteResult result = delegate.result();
TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));
taskCommit.reportOutputMetrics();
return taskCommit;
}
@Override
public void abort() throws IOException {
close();
DataWriteResult result = delegate.result();
SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());
}
@Override
public void close() throws IOException {
delegate.close();
}
}
private static class PartitionedDataWriter implements DataWriter<InternalRow> {
private final PartitioningWriter<InternalRow, DataWriteResult> delegate;
private final FileIO io;
private final PartitionSpec spec;
private final PartitionKey partitionKey;
private final InternalRowWrapper internalRowWrapper;
private PartitionedDataWriter(
SparkFileWriterFactory writerFactory,
OutputFileFactory fileFactory,
FileIO io,
PartitionSpec spec,
Schema dataSchema,
StructType dataSparkType,
long targetFileSize,
boolean fanoutEnabled) {
if (fanoutEnabled) {
this.delegate = new FanoutDataWriter<>(writerFactory, fileFactory, io, targetFileSize);
} else {
this.delegate = new ClusteredDataWriter<>(writerFactory, fileFactory, io, targetFileSize);
}
this.io = io;
this.spec = spec;
this.partitionKey = new PartitionKey(spec, dataSchema);
this.internalRowWrapper = new InternalRowWrapper(dataSparkType, dataSchema.asStruct());
}
@Override
public void write(InternalRow row) throws IOException {
partitionKey.partition(internalRowWrapper.wrap(row));
delegate.write(row, spec, partitionKey);
}
@Override
public WriterCommitMessage commit() throws IOException {
close();
DataWriteResult result = delegate.result();
TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));
taskCommit.reportOutputMetrics();
return taskCommit;
}
@Override
public void abort() throws IOException {
close();
DataWriteResult result = delegate.result();
SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());
}
@Override
public void close() throws IOException {
delegate.close();
}
}
}这里会根据是否分区，对new出来的writer传入上游给的参数。
这两个类，都实现了DataWriter类，针对写入的类型是InternalRow，这是SparkSQL中用于做数据流动的类。而两个对应的实现类，其实实现都很轻薄：
private static class UnpartitionedDataWriter implements DataWriter<InternalRow> {
private final FileWriter<InternalRow, DataWriteResult> delegate;
private final FileIO io;
private UnpartitionedDataWriter(
SparkFileWriterFactory writerFactory,
OutputFileFactory fileFactory,
FileIO io,
PartitionSpec spec,
long targetFileSize) {
this.delegate =
new RollingDataWriter<>(writerFactory, fileFactory, io, targetFileSize, spec, null);
this.io = io;
}
@Override
public void write(InternalRow record) throws IOException {
delegate.write(record);
}
@Override
public WriterCommitMessage commit() throws IOException {
close();
DataWriteResult result = delegate.result();
TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));
taskCommit.reportOutputMetrics();
return taskCommit;
}
@Override
public void abort() throws IOException {
close();
DataWriteResult result = delegate.result();
SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());
}
@Override
public void close() throws IOException {
delegate.close();
}
}
private static class PartitionedDataWriter implements DataWriter<InternalRow> {
private final PartitioningWriter<InternalRow, DataWriteResult> delegate;
private final FileIO io;
private final PartitionSpec spec;
private final PartitionKey partitionKey;
private final InternalRowWrapper internalRowWrapper;
private PartitionedDataWriter(
SparkFileWriterFactory writerFactory,
OutputFileFactory fileFactory,
FileIO io,
PartitionSpec spec,
Schema dataSchema,
StructType dataSparkType,
long targetFileSize,
boolean fanoutEnabled) {
if (fanoutEnabled) {
this.delegate = new FanoutDataWriter<>(writerFactory, fileFactory, io, targetFileSize);
} else {
this.delegate = new ClusteredDataWriter<>(writerFactory, fileFactory, io, targetFileSize);
}
this.io = io;
this.spec = spec;
this.partitionKey = new PartitionKey(spec, dataSchema);
this.internalRowWrapper = new InternalRowWrapper(dataSparkType, dataSchema.asStruct());
}
@Override
public void write(InternalRow row) throws IOException {
partitionKey.partition(internalRowWrapper.wrap(row));
delegate.write(row, spec, partitionKey);
}
@Override
public WriterCommitMessage commit() throws IOException {
close();
DataWriteResult result = delegate.result();
TaskCommit taskCommit = new TaskCommit(result.dataFiles().toArray(new DataFile[0]));
taskCommit.reportOutputMetrics();
return taskCommit;
}
@Override
public void abort() throws IOException {
close();
DataWriteResult result = delegate.result();
SparkCleanupUtil.deleteTaskFiles(io, result.dataFiles());
}
@Override
public void close() throws IOException {
delegate.close();
}
}一看到delegate，有点代码经验的同学都知道了啊——它才是真正干活的人，像极了你的老板对外疯狂揽活而在内部疯狂消化需求的你。
两者内部对应的delegate实现类为：
- PartitionedDataWriter：FanoutDataWriter、ClusteredDataWriter
- UnpartitionedDataWriter：RollingDataWriter
/**
* A data writer capable of writing to multiple specs and partitions that keeps data writers for
* each seen spec/partition pair open until this writer is closed.
*/
public class FanoutDataWriter<T> extends FanoutWriter<T, DataWriteResult> {
private final FileWriterFactory<T> writerFactory;
private final OutputFileFactory fileFactory;
private final FileIO io;
private final long targetFileSizeInBytes;
private final List<DataFile> dataFiles;
public FanoutDataWriter(
FileWriterFactory<T> writerFactory,
OutputFileFactory fileFactory,
FileIO io,
long targetFileSizeInBytes) {
this.writerFactory = writerFactory;
this.fileFactory = fileFactory;
this.io = io;
this.targetFileSizeInBytes = targetFileSizeInBytes;
this.dataFiles = Lists.newArrayList();
}
@Override
protected FileWriter<T, DataWriteResult> newWriter(PartitionSpec spec, StructLike partition) {
return new RollingDataWriter<>(
writerFactory, fileFactory, io, targetFileSizeInBytes, spec, partition);
}
@Override
protected void addResult(DataWriteResult result) {
dataFiles.addAll(result.dataFiles());
}
@Override
protected DataWriteResult aggregatedResult() {
return new DataWriteResult(dataFiles);
}
}
/**
* A data writer capable of writing to multiple specs and partitions that requires the incoming
* records to be properly clustered by partition spec and by partition within each spec.
*/
public class ClusteredDataWriter<T> extends ClusteredWriter<T, DataWriteResult> {
private final FileWriterFactory<T> writerFactory;
private final OutputFileFactory fileFactory;
private final FileIO io;
private final long targetFileSizeInBytes;
private final List<DataFile> dataFiles;
public ClusteredDataWriter(
FileWriterFactory<T> writerFactory,
OutputFileFactory fileFactory,
FileIO io,
long targetFileSizeInBytes) {
this.writerFactory = writerFactory;
this.fileFactory = fileFactory;
this.io = io;
this.targetFileSizeInBytes = targetFileSizeInBytes;
this.dataFiles = Lists.newArrayList();
}
@Override
protected FileWriter<T, DataWriteResult> newWriter(PartitionSpec spec, StructLike partition) {
return new RollingDataWriter<>(
writerFactory, fileFactory, io, targetFileSizeInBytes, spec, partition);
}
@Override
protected void addResult(DataWriteResult result) {
dataFiles.addAll(result.dataFiles());
}
@Override
protected DataWriteResult aggregatedResult() {
return new DataWriteResult(dataFiles);
}
}无论是FanoutDataWriter还是ClusteredDataWriter，最终干活的类都是RollingDataWriter。
RollingFileWriter很明显是一个File级别的写入，而RollingDataWriter则是Data级别的写入——针对上层各种引擎可以适配它们自己的内部类型。
这里套了一层又一层，本质就是个Wrapper模式的实现——因为Iceberg本质就是管理S3上那些文件的，上层计算引擎怎么定义数据类型，它其实并不关心，最后只要能变成DataFile就行了。
那今天内容就到这里了，当然实际可以讲的还有很多啊。星球不止有源码分析，还有一些我之前的面试分析、辅导实录，还有一些项目亮点，都是能拎出去打的啊。
最后关注不走丢，我们下期见。