Databricks Sink Node

The databricksSink node in ZephFlow enables you to ingest processed data into Databricks Unity Catalog tables using the Databricks SQL Compute engine.

Overview

The Databricks sink writes pipeline output to a Databricks Unity Catalog table using a "Stage and Load" approach: data is converted to Parquet, uploaded to a Databricks Volume, and then loaded into the target table via a COPY INTO SQL command on a SQL Warehouse. This ensures ACID compliance and governance within the Databricks ecosystem.

Key Features

• Unity Catalog Integration: Write data to governed Databricks tables
• SQL Warehouse Execution: Leverages Databricks SQL Compute for transactional loads
• Stage and Load: Parquet staging via Databricks Volumes for reliable ingestion
• Schema Validation: Validates Avro schema against the target table at startup
• Configurable Batching: Control batch size for throughput tuning

Parameters

Parameter	Type	Description	Required
tableName	String	Fully qualified table name (e.g., catalog.schema.table)	Yes
warehouseId	String	Databricks SQL Warehouse ID to execute the COPY INTO command	Yes
avroSchema	Map<String, Object>	Avro schema as a map defining the record structure	Yes
databricksCredentialId	String	Credential ID for Databricks authentication	Yes
volumePath	String	Databricks Volume path for staging Parquet files	Yes

Config Object

The full config object for the Databricks sink node (DatabricksSinkDto.Config):

Field	Type	Required	Default	Description
tableName	String	Yes	—	Fully qualified table name (e.g., catalog.schema.table)
warehouseId	String	Yes	—	Databricks SQL Warehouse ID for executing COPY INTO commands
avroSchema	Map<String, Object>	Yes	—	Avro schema defining the record structure
databricksCredentialId	String	Yes	—	Credential ID for Databricks authentication
volumePath	String	Yes	—	Databricks Volume path for staging Parquet files (e.g., /Volumes/<catalog>/<schema>/<volume>/[path])
batchSize	int	No	10000	Number of records accumulated before triggering a Stage and Load. Must be positive and no greater than 100000.
flushIntervalMillis	long	No	30000	Maximum time in milliseconds between flushes. Must be at least 1000.
cleanupAfterCopy	boolean	No	true	Remove temporary staged files, locally and in the volume, after the COPY INTO
copyOptions	Map<String, String>	No	{}	Additional options for the COPY INTO SQL command
formatOptions	Map<String, String>	No	{}	Format options for the COPY INTO SQL command

tableName accepts either catalog.schema.table or schema.table. volumePath must be a Databricks Volume path of the form /Volumes/<catalog>/<schema>/<volume>[/path]. All configuration problems are collected and reported together in a single error.

Run Behavior

Buffering and Flush Triggers

Records are not written one at a time. They accumulate in an in-memory buffer, and a Stage and Load runs when any of these happens:

1. The buffer reaches batchSize records (default 10000)
2. The background flush timer fires, every flushIntervalMillis (default 30000, i.e. 30 seconds)
3. The node is closed, which flushes whatever partial batch remains

Because of the timer, a low-volume pipeline still lands data within flushIntervalMillis rather than waiting indefinitely for the batch to fill.

Test Runs Write Immediately

A test run puts the sink in synchronous mode: the flush timer is not started and every batch is flushed as soon as it arrives, regardless of batchSize. This means a test run performs a real Stage and Load — Parquet is written, uploaded to the volume, and COPY INTO is executed against the actual table. Point test runs at a table you are willing to write to.

In a deployed run the buffer and timer behave as described above.

Each Flush

A single flush performs the full Stage and Load for its batch, under a batch-scoped subdirectory of the volume:

1. Stage — records are converted to Parquet locally. A record that cannot be converted becomes an error output for that record; the rest of the batch continues.
2. Upload — Parquet files are uploaded to the volume, retrying up to 3 times per file. If any file still fails, the whole batch is aborted and reported as failed, and the staged files are cleaned up. This is deliberate: loading only some of the files would leave a silent data gap.
3. Load — COPY INTO <table> FROM <batch dir> FILEFORMAT = PARQUET runs on the SQL Warehouse. FORMAT_OPTIONS and COPY_OPTIONS are emitted from formatOptions and copyOptions. When no copyOptions are supplied, the node applies COPY_OPTIONS ('mergeSchema' = 'true').
4. Clean up — local temp files and the remote batch directory are removed, unless cleanupAfterCopy is false. Clean-up also runs when the batch fails.

The node waits for COPY INTO to finish, polling until the statement completes, and gives up after 30 minutes. A COPY INTO that fails after partially loading rows logs the number loaded and reports the whole batch as failed.

Retry and Recovery

A failing flush is retried up to 3 times with exponential backoff starting at 1 second (1s, then 2s). If it still fails, the node enters recovery: each record in the batch is validated individually, records that fail validation are reported as errors with the reason, and the remaining good records are flushed again. If that bulk write also fails, every record in the batch is reported as an error.

This is what keeps one malformed record from costing you the whole batch.

Dead Letter Queue

Errors from a flush triggered by the timer or by close cannot be returned to the caller, so those records are logged as lost and counted in sink_error_count. They are not written to the DLQ, whether or not one is configured. Errors from a flush triggered by a full buffer travel back through the normal pipeline error path.

Startup Schema Validation

At startup the node reads the target Delta table's schema and compares it with the schema derived from avroSchema. A mismatch raises a schema validation error and the node does not start, so schema drift is caught before any data is written.

Input Format

Each record reaching the sink is treated as a map of field name to value. For every field in avroSchema, the node looks up that exact field name in the record. Fields present in the record but absent from the schema are ignored; fields in the schema but absent from the record are treated as null.

Null and Required Fields

A field is nullable only if its Avro type is a union with null. Writing a null — or omitting the field entirely — for a non-nullable field fails that record with an error naming the field and listing the field names actually present in the input, which makes a field-name typo easy to spot.

Avro to Delta Type Mapping

Avro type	Delta type
string, enum	STRING
int	INTEGER, or DATE with the date logical type
long	LONG, or TIMESTAMP with the timestamp-millis / timestamp-micros logical type
float	FLOAT
double	DOUBLE
boolean	BOOLEAN
bytes, fixed	BINARY, or DECIMAL with the decimal logical type
array	ARRAY
map	MAP
record	STRUCT

A standalone null type is rejected. A union is supported only as a nullable type — that is, null plus exactly one other type. Any other union is rejected when the schema is parsed, so the node fails at config time rather than at write time.

Value Coercion

Values do not have to arrive already in the target type. Each value is coerced as follows, and a value that cannot be coerced fails that record with an error naming the field:

Target type	Accepted input
STRING	Any value, converted with toString()
BYTE, SHORT, INTEGER, LONG, FLOAT, DOUBLE	Any number, or a string that parses as that type
DATE	An integer, or a string that parses as an integer (days since epoch)
BOOLEAN	Boolean values retain their value; for other non-null values, text "true" (case-insensitive) becomes true and every other textual value becomes false. Null remains null, subject to the column's nullable constraint.
TIMESTAMP, TIMESTAMP_NTZ	A java.sql.Timestamp, a java.time.Instant, a long of epoch milliseconds, or a string that parses as a long
DECIMAL	A BigDecimal, any other number, or a string accepted by the BigDecimal constructor
BINARY	A byte[], or any other value's toString() bytes
ARRAY	A list. Each element is coerced with these same rules. Anything else fails the record.
STRUCT	A map. Anything else fails the record.

warning

Boolean values retain their value. For other non-null values, Boolean.parseBoolean returns true for the textual value "true" (case-insensitive) and false for other textual values, without a parsing error. Null remains null; a missing or null value still fails for a non-nullable column. Equality with Boolean true is a type-sensitive predicate, not a general Boolean cast or a conversion recipe. Text "true" compared with Boolean true yields false; null compared with Boolean true also yields false. That predicate does not preserve the string or null behavior described above. Do not present an equality expression as the implementation of this sink conversion or invent an equality-based recipe for an unspecified input policy. For a different conversion, first define the accepted input values and the intended true/false/null or error policy; if no such policy is defined, describe the sink behavior without proposing a custom expression.

When to Use

• Choose Databricks Sink if you need Unity Catalog governance, ACID compliance, and immediate query access for analysts.
• Choose Delta Lake Sink if you need a raw data landing zone, want to minimize compute costs, or write to storage not coupled to a Databricks workspace.

DAG Examples

Basic Usage

jobContext:
  otherProperties:
    my-databricks-credential:
      host: "https://my-workspace.cloud.databricks.com"
      clientId: "my-service-principal-client-id"
      clientSecret: "my-service-principal-secret"

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

  - id: "sink"
    commandName: "databrickssink"
    config:
      tableName: "prod.finance.revenue_reports"
      warehouseId: "my-sql-warehouse-id"
      databricksCredentialId: "my-databricks-credential"
      volumePath: "/Volumes/prod/finance/staging/parquet"
      avroSchema:
        type: "record"
        name: "RevenueReport"
        fields:
          - name: "report_id"
            type: "string"
          - name: "amount"
            type: "double"
          - name: "region"
            type: "string"
    outputs: []

With Filtering and Transformation

jobContext:
  otherProperties:
    my-databricks-credential:
      host: "https://my-workspace.cloud.databricks.com"
      clientId: "my-service-principal-client-id"
      clientSecret: "my-service-principal-secret"

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka:9092"
      topic: "raw-events"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "filter"

  - id: "filter"
    commandName: "filter"
    config:
      expression: $.status == "completed"
    outputs:
      - "transform"

  - id: "transform"
    commandName: "eval"
    config:
      expression: dict(order_id=$.id, total=$.amount * 100, region=upper($.region))
    outputs:
      - "sink"

  - id: "sink"
    commandName: "databrickssink"
    config:
      tableName: "prod.finance.revenue_reports"
      warehouseId: "my-sql-warehouse-id"
      databricksCredentialId: "my-databricks-credential"
      volumePath: "/Volumes/prod/finance/staging/parquet"
      avroSchema:
        type: "record"
        name: "RevenueReport"
        fields:
          - name: "report_id"
            type: "string"
          - name: "amount"
            type: "double"
          - name: "region"
            type: "string"
    outputs: []

Related Nodes

• deltalakeSink: Write data to Delta Lake tables on cloud storage
• s3Sink: Write data to S3 in JSON or CSV formats
• kafkaSink: Write data to Kafka topics