Dataflow Pipeline & Buffering
Relevant source files
The following files were used as context for generating this wiki page:
- airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/dataflow/pipeline/DataFlowPipeline.kt
- airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/dataflow/aggregate/Aggregate.kt
- airbyte-integrations/developer-docs/destinations/step-by-step/0-introduction.md
- airbyte-integrations/developer-docs/destinations/step-by-step/1-getting-started.md
- docusaurus/platform_versioned_docs/version-2.0/contributing-to-airbyte/writing-docs.md
The Dataflow Pipeline & Buffering system represents the core architectural pattern for high-performance data movement within Airbyte Bulk CDK destination connectors. Its primary purpose is to manage the lifecycle of record processing, from initial ingestion from the Airbyte platform to efficient batch writes in a target database. This system minimizes I/O overhead by aggregating records in-memory or in temporary storage before committing them to the destination.
The pipeline orchestrates several specialized components: the Writer, which provides high-level orchestration; the Aggregate, which manages record grouping and batching logic; and the InsertBuffer, which handles the database-specific mechanics of efficient batch writes. This structure allows for a clear separation of concerns, where SQL generation is decoupled from execution, ensuring both reliability and testability.
Pipeline Architecture & Lifecycle
The Dataflow Pipeline follows a structured lifecycle to ensure data integrity and efficient resource usage. The lifecycle begins with the setup() phase and proceeds through continuous record aggregation until the close() signal is received, which triggers the finalization of all pending data and state.
High-Level Data Flow
Records flow from the Airbyte platform into a lifecycle manager that coordinates the interaction between the buffering logic and the database loader.
The diagram shows the sequence from record ingestion via stdin to the final state update via stdout.
Sources: 0-introduction.md
Component Roles
The pipeline is composed of five primary roles that define how data is transformed and loaded:
| Component | Responsibility |
|---|---|
| Writer | Orchestrates the high-level sync; implements setup() and createStreamLoader(). |
| SqlGenerator | A pure function component that generates SQL strings without executing them. |
| Client | Executes SQL strings against the target database and handles I/O errors. |
| InsertBuffer | Implements database-specific batching and buffering strategies. |
| StreamLoader | CDK-provided component that manages the table lifecycle (temp tables, swaps). |
Sources: 0-introduction.md
Aggregation and Buffering
The Aggregate and InsertBuffer components are responsible for the "Buffering" aspect of the pipeline. They prevent the performance degradation associated with single-row inserts by grouping records based on stream and configuration.
The Aggregate Component
The Aggregate interface serves as the primary container for records belonging to a specific stream. It maintains the current state of the buffer and determines when a "flush" to the destination is required.
- Logic: It receives records and stores them until a threshold (memory or count) is reached.
- Factory:
AggregateFactoryis used by the pipeline to instantiate the correct aggregate implementation for a given stream context.
Sources: Aggregate.kt, 0-introduction.md
InsertBuffer Pattern
The InsertBuffer is a database-specific implementation that translates the abstracted records in an Aggregate into the most efficient format for the target database (e.g., a bulk COPY command or a multi-row INSERT).
- Append Mode: In this mode, the buffer writes directly to the destination tables.
- Overwrite Mode: The buffer writes to temporary tables, which are then atomically swapped with the final table by the
StreamLoaderduring the close phase.
Sources: 0-introduction.md, 1-getting-started.md
Pipeline Orchestration
The DataFlowPipeline class manages the execution flow, ensuring that components are initialized in the correct order and that resources are cleaned up even in the event of failure.
Sequence of Operations
The orchestration sequence follows a synchronous pattern where each record is passed through the buffer before state is emitted.
This sequence illustrates the interaction between the platform and the pipeline components during a sync.
Sources: 0-introduction.md, DataFlowPipeline.kt
Configuration & Dependency Injection
The pipeline relies on Micronaut-based Dependency Injection (DI) to assemble its components. Required configuration elements include:
- Data Channel: Configured via
data-channel.medium(STDIO or SOCKET). - Serialization: Supports JSONL and PROTOBUF formats.
- Namespace Mapping: Defines how source namespaces are transformed into destination schemas.
Sources: 1-getting-started.md:143-160
Implementation Details
The implementation of a Dataflow Pipeline requires specific scaffolding and configuration files to be present in the connector project.
Required Configuration File
A file named application-connector.yml must be present in src/main/resources to prevent DI errors during pipeline initialization. It defines the core data channel and mapping behaviors.
airbyte:
destination:
core:
data-channel:
medium: STDIO
format: JSONL
mappers:
namespace-mapping-config-path: ""
file-transfer:
enabled: false
Sources: 1-getting-started.md:143-160
Error Handling & Validation
The pipeline uses a "Fail fast & actionably" approach. Validation is performed during the spec and check phases to ensure that the user-provided configuration adheres to the expected schema before the pipeline begins processing records.
Sources: version-2.0/contributing-to-airbyte/writing-docs.md, 1-getting-started.md
Summary
The Dataflow Pipeline & Buffering system provides the necessary infrastructure for Airbyte Bulk CDK destinations to process large volumes of data efficiently. By utilizing an aggregation-based architecture, the pipeline decouples the ingestion of Airbyte records from the database's write operations. This enables the use of optimized bulk loading techniques like temporary table swaps and batch inserts, ensuring high throughput and data integrity throughout the synchronization process.