bulk-cdk-toolkit-load-object-storage/io.airbyte.cdk.load.pipline.object_storage

Package-level declarations

Types

data class ObjectKey(val stream: DestinationStream.Descriptor, val objectKey: String, val uploadId: String? = null) : WithStream

ObjectLoaderCompletedUploadPartitioner

interface ObjectLoaderCompletedUploadPartitioner<K1 : WithStream, T, K2 : WithStream, U : RemoteObject<*>> : OutputPartitioner<K1, T, K2, ObjectLoaderUploadCompleter.UploadResult<U>>

ObjectLoaderFormattedPartPartitioner

class ObjectLoaderFormattedPartPartitioner<K : WithStream, T> : OutputPartitioner<K, T, ObjectKey, ObjectLoaderPartFormatter.FormattedPart>

Distribute the parts randomly across loaders. (Testing shows this is the most efficient pattern.)

ObjectLoaderLoadedPartPartitioner

class ObjectLoaderLoadedPartPartitioner<K : WithStream, T, U : RemoteObject<*>> : OutputPartitioner<K, T, ObjectKey, ObjectLoaderPartLoader.PartResult<U>>

The technically correct partitioning is round-robin, but since we use io.airbyte.cdk.load.message.SinglePartitionQueueWithMultiPartitionBroadcast, the partition is immaterial, so it's simpler just to return 0 here.

ObjectLoaderOneShotUploader

@Singleton

@Requires(bean = ObjectLoader::class)

class ObjectLoaderOneShotUploader<O : OutputStream, T : RemoteObject<*>>(partFormatter: ObjectLoaderPartFormatter<O>, partLoader: ObjectLoaderPartLoader<T>, uploadCompleter: ObjectLoaderUploadCompleter<T>, @Named(value = "uploadParallelismForSocket") uploadParallelism: Int) : BatchAccumulator<ObjectLoaderOneShotUploader.State<O, T>, StreamKey, DestinationRecordRaw, ObjectLoaderUploadCompleter.UploadResult<T>>

Wraps the PartFormatter, PartLoader, and UploadCompleter in a single BatchAccumulator.

ObjectLoaderOneShotUploaderStep

class ObjectLoaderOneShotUploaderStep<K : WithStream, T : RemoteObject<*>>(objectLoaderOneShotUploader: ObjectLoaderOneShotUploader<*, T>, completedUploadQueue: PartitionedQueue<PipelineEvent<K, ObjectLoaderUploadCompleter.UploadResult<T>>>? = null, completedUploadPartitioner: ObjectLoaderCompletedUploadPartitioner<StreamKey, DestinationRecordRaw, K, T>? = null, taskFactory: LoadPipelineStepTaskFactory, val numWorkers: Int) : LoadPipelineStep

ObjectLoaderPartFormatter

@Singleton

@Requires(bean = ObjectLoader::class)

class ObjectLoaderPartFormatter<T : OutputStream>(pathFactory: PathFactory, catalog: DestinationCatalog, writerFactory: BufferedFormattingWriterFactory<T>, loader: ObjectLoader, stateManager: DestinationStateManager<ObjectStorageDestinationState>, @Value(value = "${airbyte.destination.core.record-batch-size-override:null}") val batchSizeOverride: Long? = null, @Named(value = "objectLoaderClampedPartSizeBytes") val clampedPartSizeBytes: Long) : BatchAccumulator<ObjectLoaderPartFormatter.State<T>, StreamKey, DestinationRecordRaw, ObjectLoaderPartFormatter.FormattedPart>

ObjectLoaderPartFormatterStep

class ObjectLoaderPartFormatterStep(val numWorkers: Int, partFormatter: ObjectLoaderPartFormatter<*>, inputFlows: Array<Flow<PipelineEvent<StreamKey, DestinationRecordRaw>>>, outputQueue: PartitionedQueue<PipelineEvent<ObjectKey, ObjectLoaderPartFormatter.FormattedPart>>, taskFactory: LoadPipelineStepTaskFactory, stepId: String, flushStrategy: PipelineFlushStrategy?) : LoadPipelineStep

ObjectLoaderPartLoader

@Singleton

@Requires(bean = ObjectLoader::class)

class ObjectLoaderPartLoader<T : RemoteObject<*>>(client: ObjectStorageClient<T>, catalog: DestinationCatalog, uploads: UploadsInProgress<T>, destinationConfig: DestinationConfiguration) : BatchAccumulator<ObjectLoaderPartLoader.State<T>, ObjectKey, ObjectLoaderPartFormatter.FormattedPart, ObjectLoaderPartLoader.PartResult<T>>

ObjectLoaderPartLoaderStep

class ObjectLoaderPartLoaderStep<T : RemoteObject<*>>(loader: ObjectLoader, partLoader: ObjectLoaderPartLoader<T>, inputQueue: PartitionedQueue<PipelineEvent<ObjectKey, ObjectLoaderPartFormatter.FormattedPart>>, outputQueue: PartitionedQueue<PipelineEvent<ObjectKey, ObjectLoaderPartLoader.PartResult<T>>>, taskFactory: LoadPipelineStepTaskFactory, stepId: String) : LoadPipelineStep

ObjectLoaderPipeline

@Singleton

@Requires(bean = ObjectLoader::class)

class ObjectLoaderPipeline<K : WithStream, T : RemoteObject<*>>(routeEventStep: RouteEventStep?, fileChunkStep: FileChunkStep<T>?, @Named(value = "filePartLoaderStep") fileChunkUploader: ObjectLoaderPartLoaderStep<T>?, @Named(value = "fileUploadCompleterStep") fileCompleterStep: ObjectLoaderUploadCompleterStep<K, T>?, forwardFileRecordStep: ForwardFileRecordStep<T>?, @Named(value = "fileRecordPartFormatterStep") fileRecordFormatStep: ObjectLoaderPartFormatterStep?, @Named(value = "recordPartFormatterStep") recordFormatStep: ObjectLoaderPartFormatterStep, @Named(value = "recordPartLoaderStep") recordUploadStep: ObjectLoaderPartLoaderStep<T>, @Named(value = "recordUploadCompleterStep") recordCompleterStep: ObjectLoaderUploadCompleterStep<K, T>, @Value(value = "${airbyte.destination.core.file-transfer.enabled}") isLegacyFileTransfer: Boolean, processFileTaskLegacyStep: ProcessFileTaskLegacyStep, @Named(value = "isFileTransfer") isFileTransfer: Boolean, @Named(value = "oneShotObjectLoaderStep") oneShotObjectLoaderStep: ObjectLoaderOneShotUploaderStep<K, T>, @Named(value = "dataChannelMedium") dataChannelMedium: DataChannelMedium) : LoadPipeline

Three steps for default record flow:

ObjectLoaderUploadCompleter

@Singleton

@Requires(bean = ObjectLoader::class)

class ObjectLoaderUploadCompleter<T : RemoteObject<*>>(val objectLoader: ObjectLoader) : BatchAccumulator<ObjectLoaderUploadCompleter.State, ObjectKey, ObjectLoaderPartLoader.PartResult<T>, ObjectLoaderUploadCompleter.UploadResult<T>>

ObjectLoaderUploadCompleterStep

class ObjectLoaderUploadCompleterStep<K : WithStream, T : RemoteObject<*>>(objectLoader: ObjectLoader, uploadCompleter: ObjectLoaderUploadCompleter<T>, inputQueue: PartitionedQueue<PipelineEvent<ObjectKey, ObjectLoaderPartLoader.PartResult<T>>>, completedUploadQueue: PartitionedQueue<PipelineEvent<K, ObjectLoaderUploadCompleter.UploadResult<T>>>? = null, completedUploadPartitioner: ObjectLoaderCompletedUploadPartitioner<ObjectKey, ObjectLoaderPartLoader.PartResult<T>, K, T>? = null, taskFactory: LoadPipelineStepTaskFactory, stepId: String) : LoadPipelineStep

UploadsInProgress

@Singleton

@Requires(bean = ObjectLoader::class)

class UploadsInProgress<T : RemoteObject<*>>

In order to allow streaming uploads on the same key to be parallelized, upload state needs to be shared across workers.