glassflow-clickhouse-etl 0.2.3

GlassFlow Clickhouse ETL Python SDK: Create GlassFlow pipelines between Kafka and ClickHouse

Clickhouse ETL Python SDK

A Python SDK for creating and managing data pipelines between Kafka and ClickHouse.

Features

• Create and manage data pipelines between Kafka and ClickHouse
• Deduplication of events during a time window based on a key
• Temporal joins between topics based on a common key with a given time window
• Schema validation and configuration management

Installation

pip install glassflow-clickhouse-etl

Quick Start

from glassflow_clickhouse_etl import Pipeline


pipeline_config = {
  "pipeline_id": "test-pipeline",
  "source": {
    "type": "kafka",
    "provider": "aiven",
    "connection_params": {
      "brokers": ["localhoust:9092"],
      "protocol": "SASL_SSL",
      "mechanism": "SCRAM-SHA-256",
      "username": "user",
      "password": "pass"
    }
    "topics": [
      {
        "consumer_group_initial_offset": "earliest",
        "id": "test-topic",
        "name": "test-topic",
        "schema": {
          "type": "json",
          "fields": [
            {"name": "id", "type": "string" },
            {"name": "email", "type": "string"}
          ]
        },
        "deduplication": {
          "id_field": "id",
          "id_field_type": "string",
          "time_window": "1h",
          "enabled": True
        }
      }
    ],
  },
  "sink": {
    "type": "clickhouse",
    "host": "localhost:8443",
    "port": 8443,
    "database": "test",
    "username": "default",
    "password": "pass",
    "table_mapping": [
      {
        "source_id": "test_table",
        "field_name": "id",
        "column_name": "user_id",
        "column_type": "UUID"
      },
      {
        "source_id": "test_table",
        "field_name": "email",
        "column_name": "email",
        "column_type": "String"
      }
    ]
  }
}

# Create a pipeline from a JSON configuration
pipeline = Pipeline(pipeline_config)

# Create the pipeline
pipeline.create()

Configuration

For detailed information about the pipeline configuration, see CONFIGURATION.

Tracking

The SDK includes anonymous usage tracking to help improve the product. Tracking is enabled by default but can be disabled in two ways:

1. Using an environment variable:

export GF_TRACKING_ENABLED=false

2. Programmatically using the disable_tracking method:

pipeline = Pipeline(pipeline_config)
pipeline.disable_tracking()

The tracking collects anonymous information about:

• SDK version
• Platform (operating system)
• Python version
• Pipeline ID
• Whether joins or deduplication are enabled

Development

Setup

1. Clone the repository
2. Create a virtual environment
3. Install dependencies:

uv venv
source .venv/bin/activate
uv pip install -e .[dev]

Testing

pytest