Pipeline JSON Reference

The pipeline configuration is defined in a JSON file that specifies the source, sink, and any transformations. While the web interface automatically generates this configuration, understanding its structure can be helpful for advanced users.

The pipeline configuration is a JSON object that defines how data flows from Kafka topics to ClickHouse tables.

Root Configuration

Source Configuration

The source configuration defines how to connect to and consume from Kafka topics.

Kafka Connection Parameters

Supported Protocol:

• PLAINTEXT - Plaintext protocol.
• SSL - SSL protocol.
• SASL_SSL - SASL over SSL protocol.
• SASL_PLAINTEXT - SASL over plaintext protocol.

Supported Mechanism:

• NO_AUTH - No authentication mechanism.
• PLAIN - Plaintext authentication.
• SCRAM-SHA-256 - SCRAM-SHA-256 authentication.
• SCRAM-SHA-512 - SCRAM-SHA-512 authentication.
• GSSAPI - GSSAPI authentication.

Topic Configuration

Each topic in the topics array has the following configuration:

Deduplication Configuration

Sink Configuration

The sink configuration defines how to connect to and write to ClickHouse.

Schema Configuration

The schema configuration defines the schema of the pipeline.

Field Configuration

The type parameter accepts the following Kafka data types:

• 'string' - For text data, including datetime strings
• 'int', 'int8', 'int16', 'int32', 'int64' - For integer values
• 'uint', 'uint8', 'uint16', 'uint32', 'uint64' - For unsigned integer values
• 'float', 'float32', 'float64' - For floating-point numbers
• 'bool' - For boolean values
• 'bytes' - For binary data
• 'array' - For array data
• 'map' - For key-value pair data

The column_type parameter accepts the following ClickHouse data types:

Basic Types:

• 'Int8', 'Int16', 'Int32', 'Int64' - For integer values
• 'UInt8', 'UInt16', 'UInt32', 'UInt64' - For unsigned integer values
• 'Float32', 'Float64' - For float values
• 'String' - For string values
• 'FixedString' - For fixedstring values
• 'DateTime', 'DateTime64' - For datetime values
• 'Bool' - For bool values
• 'UUID' - For uuid values
• 'Enum8', 'Enum16' - For enum values

LowCardinality Types:

• 'LowCardinality(Int8)', 'LowCardinality(Int16)', 'LowCardinality(Int32)', 'LowCardinality(Int64)' - For low cardinality int values
• 'LowCardinality(UInt8)', 'LowCardinality(UInt16)', 'LowCardinality(UInt32)', 'LowCardinality(UInt64)' - For low cardinality uint values
• 'LowCardinality(Float32)', 'LowCardinality(Float64)' - For low cardinality float values
• 'LowCardinality(String)' - For low cardinality string values
• 'LowCardinality(FixedString)' - For low cardinality fixedstring values
• 'LowCardinality(DateTime)' - For low cardinality datetime values

Array Types:

• 'Array(String)' - For array of string values
• 'Array(Int8)', 'Array(Int16)', 'Array(Int32)', 'Array(Int64)' - For array of int values
• 'Array(UInt8)', 'Array(UInt16)', 'Array(UInt32)', 'Array(UInt64)' - For array of uint values
• 'Array(Float32)', 'Array(Float64)' - For array of float values
• 'Array(Bool)' - For array of bool values
• 'Array(Map(String, String))' - For array of maps

Map Types:

• 'Map(String, String)' - For key-value pairs

💡 Note: For comprehensive type mapping information, see the Supported Data Formats documentation.

Join Configuration

The join configuration defines how to join data from multiple Kafka topics.

Join Source Configuration

Each source in the sources array has the following configuration:

Filter Configuration

Metadata Configuration

Other configuration notes

Time windows

Time windows use string format, for example:

• "30s" - 30 seconds
• "1m" - 1 minute
• "1h" - 1 hour
• "12h" - 12 hours
• "24h" - 24 hours

Example Configuration

{
  "version": "v2",
  "pipeline_id": "kafka-to-clickhouse-pipeline",
  "source": {
    "type": "kafka",
    "connection_params": {
      "brokers": [
        "kafka-broker-0:9092",
        "kafka-broker-1:9092"
      ],
      "protocol": "SASL_SSL",
      "mechanism": "SCRAM-SHA-256",
      "username": "<user>",
      "password": "<password>",
      "root_ca": "<base64 encoded ca>"
    },
    "topics": [
      {
        "consumer_group_initial_offset": "earliest",
        "name": "user_logins",
        "deduplication": {
          "enabled": true,
          "id_field": "session_id",
          "id_field_type": "string",
          "time_window": "12h"
        }
      },
      {
        "consumer_group_initial_offset": "earliest",
        "name": "orders",
        "deduplication": {
          "enabled": true,
          "id_field": "order_id",
          "id_field_type": "string",
          "time_window": "12h"
        }
      }
    ]
  },
  "join": {
    "enabled": false,
    "type": "temporal",
    "sources": [
      {
        "source_id": "user_logins",
        "join_key": "user_id",
        "time_window": "1h",
        "orientation": "left"
      },
      {
        "source_id": "orders",
        "join_key": "user_id",
        "time_window": "1h",
        "orientation": "right"
      }
    ]
  },
  "sink": {
    "type": "clickhouse",
    "provider": "aiven",
    "host": "<host>",
    "port": "12753",
    "database": "default",
    "username": "<user>",
    "password": "<password>",
    "secure": true,
    "max_batch_size": 1,
    "max_delay_time": "10m",
    "table": "user_orders"
  },
  "schema": {
    "fields": [
      {
        "source_id": "user_logins",
        "name": "session_id",
        "type": "string",
        "column_name": "session_id",
        "column_type": "UUID"
      },
      {
        "source_id": "user_logins",
        "name": "user_id",
        "type": "string",
        "column_name": "user_id",
        "column_type": "UUID"
      },
      {
        "source_id": "orders",
        "name": "user_id",
        "type": "string"
      },
      {
        "source_id": "orders",
        "name": "order_id",
        "type": "string",
        "column_name": "order_id",
        "column_type": "UUID"
      },
      {
        "source_id": "user_logins",
        "name": "timestamp",
        "type": "datetime",
        "column_name": "login_at",
        "column_type": "DateTime"
      },
      {
        "source_id": "orders",
        "name": "timestamp",
        "type": "datetime",
        "column_name": "order_placed_at",
        "column_type": "DateTime"
      }
    ]
  }
}

💡 Note: The web interface automatically generates this configuration based on user input, so manual editing is not required.