Delta Lake Sink Node

The deltalakeSink node in ZephFlow enables you to write processed data directly to a Delta Lake table on cloud storage or HDFS.

warning

When using S3 as the cloud storage backend, Delta Lake does not support distributed writing. Deploying a data pipeline job with multiple replicas writing data in Delta Lake format to S3 will likely cause data corruption due to the lack of a distributed lock implementation. For S3 deployments, ensure your pipeline runs with a single replica.

Overview

The Delta Lake sink writes pipeline output to an existing Delta Lake table. Data is validated against the table's schema and written in transactional batches. The table must already exist at the specified path with a defined schema.

Key Features

• Delta Lake Integration: Write data directly to Delta Lake tables on S3, GCS, Azure Blob, or HDFS
• Transactional Writes: ACID-compliant batch commits
• Built-in Batching: Automatically accumulates records and flushes at configurable batch size or time interval
• Partitioning: Support for partition columns for optimized storage and reads
• Schema Validation: Incoming data is validated against the existing Delta table schema
• Auto Checkpoint: Optional automatic checkpointing for Delta log management

Parameters

Parameter	Type	Description	Required
tablePath	String	Full URI to the Delta table (e.g., s3a://bucket/path)	Yes
avroSchema	Map<String, Object>	Avro schema defining the record structure	Yes
partitionColumns	List<String>	Columns used to partition the Delta table	No
hadoopConfiguration	Map<String, String>	Custom Hadoop properties for storage access	No
credentialId	String	ID of a credential in jobContext.otherProperties for storage authentication (omit for default)	No

Config Object

The full config object for the Delta Lake sink node (DeltaLakeSinkDto.Config):

Field	Type	Required	Default	Description
tablePath	String	Yes	—	Full URI to the Delta table (e.g., s3a://bucket/path)
avroSchema	Map<String, Object>	Yes	—	Avro schema defining the record structure
partitionColumns	List<String>	No	null	Columns used to partition the Delta table
batchSize	int	No	50000	Number of records accumulated before committing
hadoopConfiguration	Map<String, String>	No	{}	Custom Hadoop properties for storage access
credentialId	String	No	null	Credential ID for storage authentication
enableAutoCheckpoint	boolean	No	true	Automatically checkpoint the Delta log
flushIntervalSeconds	int	No	30	Maximum time in seconds between commits regardless of batch size

Supported Storage Backends

Storage Provider	Path Scheme	Credential Type
AWS S3	s3:// or s3a://	Username/Password (Access Key / Secret Key)
Google Cloud	gs://	GCP Credential
Azure Blob	abfs:// or abfss://	API Key (Storage Account Key)
HDFS	hdfs://	None (use Hadoop config)

DAG Examples

The examples below go from the simplest config to the most configurable.

Simple

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "broker:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "deltalakesink"
    config:
      tablePath: "s3a://my-bucket/data/events_delta"  # Delta table path
      avroSchema:
        type: "record"
        name: "UserEvent"
        fields:
          - name: "user_id"
            type: "string"
          - name: "event_type"
            type: "string"
          - name: "timestamp"
            type: "long"
          - name: "event_date"
            type: "string"
    outputs: []

With Partitions

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "broker:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "deltalakesink"
    config:
      tablePath: "s3a://my-bucket/data/events_delta"  # Delta table path
      avroSchema:
        type: "record"
        name: "UserEvent"
        fields:
          - name: "user_id"
            type: "string"
          - name: "event_type"
            type: "string"
          - name: "timestamp"
            type: "long"
          - name: "event_date"
            type: "string"
      partitionColumns:                                # Partition columns
        - "event_date"
    outputs: []

Full Configuration

jobContext:
  otherProperties:
    aws-cred:
      username: "AKIAIOSFODNN7EXAMPLE"
      password: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "broker:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "deltalakesink"
    config:
      tablePath: "s3a://my-bucket/data/events_delta"  # Delta table path
      avroSchema:
        type: "record"
        name: "UserEvent"
        fields:
          - name: "user_id"
            type: "string"
          - name: "event_type"
            type: "string"
          - name: "timestamp"
            type: "long"
          - name: "event_date"
            type: "string"
      partitionColumns:                                # Partition columns
        - "event_date"
      hadoopConfiguration:                             # Hadoop configuration
        fs.s3a.endpoint: "s3.us-west-2.amazonaws.com"
        fs.s3a.path.style.access: "true"
      credentialId: "aws-cred"                         # Credential (optional)
    outputs: []

Writing to S3 with Credentials

jobContext:
  otherProperties:
    aws-cred:
      username: "AKIAIOSFODNN7EXAMPLE"
      password: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka:9092"
      topic: "records"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "transform"

  - id: "transform"
    commandName: "eval"
    config:
      expression: dict_merge($, dict(event_date=epoch_to_ts_str($.timestamp, "yyyy-MM-dd")))
    outputs:
      - "sink"

  - id: "sink"
    commandName: "deltalakesink"
    config:
      tablePath: "s3a://data-lake/events_delta"
      avroSchema:
        type: "record"
        name: "UserEvent"
        fields:
          - name: "user_id"
            type: "string"
          - name: "event_type"
            type: "string"
          - name: "timestamp"
            type: "long"
          - name: "event_date"
            type: "string"
      partitionColumns:
        - "event_date"
      credentialId: "aws-cred"
    outputs: []

Writing to GCS

jobContext:
  otherProperties:
    gcp-cred:
      authType: "SERVICE_ACCOUNT_JSON_KEYFILE"
      jsonKeyContent: |
        {
          "type": "service_account",
          "project_id": "my-project",
          "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n",
          "client_email": "writer@my-project.iam.gserviceaccount.com"
        }
      projectId: "my-project"

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka:9092"
      topic: "records"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "deltalakesink"
    config:
      tablePath: "gs://data-lake-bucket/events_delta"
      avroSchema:
        type: "record"
        name: "UserEvent"
        fields:
          - name: "user_id"
            type: "string"
          - name: "event_type"
            type: "string"
          - name: "timestamp"
            type: "long"
          - name: "event_date"
            type: "string"
      credentialId: "gcp-cred"
    outputs: []

Related Nodes

• s3Sink: Write data to S3 in JSON or CSV formats
• kafkaSink: Write data to Kafka topics