Data formats

Choosing a Format

Format	Best For	Source	Sink
JSON	General-purpose streaming, webhooks, APIs	Yes	Yes
Avro	Kafka ecosystems with Schema Registry	Yes	Yes
Parquet	Lakehouse sinks (Iceberg, Delta, Filesystem)	Yes	Yes
Protobuf	High-throughput, strongly-typed gRPC services	Yes	Yes
Raw String	Log lines, CSV, custom text formats	Yes	Yes
Raw Bytes	Binary protocols, custom serialization	Yes	Yes

JSON

The most common format for streaming data. Supports structured (schema-mapped) and unstructured (single text column) modes.

schema:
  format:
    json: {}

Options

Option	Type	Default	Description
`confluent_schema_registry`	boolean	`false`	Use Confluent Schema Registry for schema resolution
`schema_id`	number	-	Schema version ID from registry
`include_schema`	boolean	`false`	Include schema in each message (for Kafka Connect compatibility)
`debezium`	boolean	`false`	Parse Debezium CDC envelope format
`unstructured`	boolean	`false`	Treat payload as a single unstructured text field
`timestamp_format`	string	`rfc3339`	How timestamps are encoded: `rfc3339` or `unix_millis`
`decimal_encoding`	string	`number`	How decimals are encoded: `number`, `string`, or `bytes`

Structured JSON

By default, JSON is parsed into columns according to the table schema. Each field in the JSON object maps to a field defined in the fields array:

schema:
  format:
    json:
      timestamp_format: rfc3339
  fields:
    - field_name: user_id
      field_type:
        type:
          primitive: Int64
      nullable: false
    - field_name: event_time
      field_type:
        type:
          primitive: DateTime
      nullable: false

Unstructured JSON

When unstructured: true, the entire JSON payload is stored as a single text column. Use SQL JSON functions to extract fields at query time:

schema:
  format:
    json:
      unstructured: true
  fields:
    - field_name: value
      field_type:
        type:
          primitive: Utf8
      nullable: false

Debezium CDC

When debezium: true, Ingest Engine expects the Debezium envelope format with before, after, and op fields. Use this with the CDC connector or any Debezium-compatible source:

schema:
  format:
    json:
      debezium: true
  primary_keys:
    - id
  fields:
    - field_name: id
      field_type:
        type:
          primitive: Int64
      nullable: false

Avro

Apache Avro binary format, commonly used in Kafka ecosystems with Confluent Schema Registry.

schema:
  format:
    avro:
      confluent_schema_registry: true

Options

Option	Type	Default	Description
`confluent_schema_registry`	boolean	`false`	Fetch/register schemas via Confluent Schema Registry
`raw_datums`	boolean	`false`	Serialize as raw Avro datums (no container/header)
`into_unstructured_json`	boolean	`false`	Convert Avro to unstructured JSON for flexible processing
`reader_schema`	string	-	Override the reader schema
`schema_id`	number	-	Schema version ID

Serialization Modes

Avro supports two serialization modes:

Raw datums (raw_datums: true) - just the data, no schema. Readers must have the exact schema. This is the mode used with Schema Registry, which distributes schemas separately.
Complete records (default) - schema is embedded in every record. Self-describing but larger on the wire.

Schema Registry Integration

When confluent_schema_registry: true:

Sources: Ingest Engine fetches the schema from the registry and uses it for deserialization
Sinks: Ingest Engine registers the schema with the registry so downstream consumers can read the data

Parquet

Apache Parquet columnar format. Primarily used for sink tables writing to lakehouse destinations (Iceberg, Delta Lake, Filesystem).

schema:
  format:
    parquet:
      compression: snappy

Options

Option	Type	Default	Description
`compression`	string	`uncompressed`	Compression codec: `uncompressed`, `snappy`, `gzip`, `zstd`, `lz4`
`row_group_bytes`	number	-	Target row group size in bytes

Compression Guide

Codec	Speed	Ratio	Use Case
`snappy`	Fast	Moderate	General purpose, best default
`zstd`	Moderate	High	Storage-optimized, cold data
`gzip`	Slow	High	Maximum compatibility
`lz4`	Fastest	Low	Latency-sensitive workloads
`uncompressed`	N/A	None	Debugging, already-compressed data

Protobuf

Protocol Buffers binary format for high-throughput, strongly-typed data interchange.

schema:
  format:
    protobuf:
      confluent_schema_registry: true
      message_name: MyMessage

Options

Option	Type	Default	Description
`message_name`	string	-	Fully-qualified Protobuf message type name
`compiled_schema`	bytes	-	Compiled protobuf descriptor (FileDescriptorSet)
`confluent_schema_registry`	boolean	`false`	Fetch schema from Confluent Schema Registry
`length_delimited`	boolean	`false`	Use length-delimited framing
`into_unstructured_json`	boolean	`false`	Convert to unstructured JSON

Raw String

Ingests or emits arbitrary UTF-8 text as a single field. Useful for log lines, CSV data, or custom text formats that you parse in SQL.

schema:
  format:
    raw_string: {}
  fields:
    - field_name: raw
      field_type:
        type:
          primitive: String
      nullable: false

The table must have exactly one field of type Utf8 / String. Use SQL text functions or regex functions to parse the data in your pipeline query.

Raw Bytes

Ingests or emits arbitrary binary data as a single field. Use for binary protocols or custom serialization formats.

schema:
  format:
    raw_bytes: {}
  fields:
    - field_name: raw
      field_type:
        type:
          primitive: Bytes
      nullable: false

The table must have exactly one field of type Bytes.

Framing

Framing controls how incoming data is split into individual records. Only applies to source tables.

Mode	Description
`newline_delimited`	Each line is a separate record (e.g., NDJSON)

schema:
  format:
    json: {}
  framing:
    newline_delimited: {}
  fields:
    # ...

Bad Data Handling

Controls how malformed or unparseable records are handled. Only applies to source tables.

Mode	Description
`fail`	Pipeline fails on the first bad record (default)
`drop`	Skip bad records silently and continue processing

schema:
  format:
    json: {}
  bad_data:
    drop: {}
  fields:
    # ...

Use drop for sources where occasional malformed records are expected (e.g., user-generated webhook data). Use fail (the default) when data quality issues should halt the pipeline.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Choosing a Format

JSON

Options

Structured JSON

Unstructured JSON

Debezium CDC

Avro

Options

Serialization Modes

Schema Registry Integration

Parquet

Options

Compression Guide

Protobuf

Options

Raw String

Raw Bytes

Framing

Bad Data Handling

​Choosing a Format

​JSON

​Options

​Structured JSON

​Unstructured JSON

​Debezium CDC

​Avro

​Options

​Serialization Modes

​Schema Registry Integration

​Parquet

​Options

​Compression Guide

​Protobuf

​Options

​Raw String

​Raw Bytes

​Framing

​Bad Data Handling

Choosing a Format

JSON

Options

Structured JSON

Unstructured JSON

Debezium CDC

Avro

Options

Serialization Modes

Schema Registry Integration

Parquet

Options

Compression Guide

Protobuf

Options

Raw String

Raw Bytes

Framing

Bad Data Handling