glassflow 3.7.2

GlassFlow Python SDK: Create GlassFlow pipelines between Kafka and ClickHouse

GlassFlow 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

Quick Start

Initialize client

from glassflow.etl import Client

# Initialize GlassFlow client
client = Client(host="your-glassflow-etl-url")

Create a pipeline

pipeline_config = {
    "version": "v2",
    "pipeline_id": "my-pipeline-id",
    "source": {
      "type": "kafka",
      "connection_params": {
        "brokers": [
          "http://my.kafka.broker:9093"
        ],
        "protocol": "PLAINTEXT",
        "mechanism": "NO_AUTH"
      },
      "topics": [
        {
          "consumer_group_initial_offset": "latest",
          "name": "users",
          "deduplication": {
            "enabled": True,
            "id_field": "event_id",
            "id_field_type": "string",
            "time_window": "1h"
          }
        }
      ]
    },
    "join": {
      "enabled": False
    },
    "sink": {
      "type": "clickhouse",
      "host": "http://my.clickhouse.server",
      "port": "9000",
      "database": "default",
      "username": "default",
      "password": "c2VjcmV0",
      "secure": False,
      "max_batch_size": 1000,
      "max_delay_time": "30s",
      "table": "users_dedup"
    },
    "schema": {
      "fields": [
        {
          "source_id": "users",
          "name": "event_id",
          "type": "string",
          "column_name": "event_id",
          "column_type": "UUID"
        },
        {
          "source_id": "users",
          "field_name": "user_id",
          "column_name": "user_id",
          "column_type": "UUID"
        },
        {
          "source_id": "users",
          "name": "created_at",
          "type": "string",
          "column_name": "created_at",
          "column_type": "DateTime"
        },
        {
          "source_id": "users",
          "name": "name",
          "type": "string",
          "column_name": "name",
          "column_type": "String"
        },
        {
          "source_id": "users",
          "name": "email",
          "type": "string",
          "column_name": "email",
          "column_type": "String"
        }
      ]
    }
}

# Create a pipeline
pipeline = client.create_pipeline(pipeline_config)

Get pipeline

# Get a pipeline by ID
pipeline = client.get_pipeline("my-pipeline-id")

List pipelines

pipelines = client.list_pipelines()
for pipeline in pipelines:
    print(f"Pipeline ID: {pipeline['pipeline_id']}")
    print(f"Name: {pipeline['name']}")
    print(f"Transformation Type: {pipeline['transformation_type']}")
    print(f"Created At: {pipeline['created_at']}")
    print(f"State: {pipeline['state']}")

Stop / Terminate / Resume Pipeline

pipeline = client.get_pipeline("my-pipeline-id")
pipeline.stop()
print(pipeline.status)

STOPPING

# Stop a pipeline ungracefully (terminate)
client.stop_pipeline("my-pipeline-id", terminate=True)
print(pipeline.status)

TERMINATING

pipeline = client.get_pipeline("my-pipeline-id")
pipeline.resume()
print(pipeline.status)

RESUMING

Delete pipeline

Only stopped or terminated pipelines can be deleted.

# Delete a pipeline
client.delete_pipeline("my-pipeline-id")

# Or delete via pipeline instance
pipeline.delete()

Pipeline Configuration

For detailed information about the pipeline configuration, see GlassFlow docs.

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:

from glassflow.etl import Client

client = Client(host="my-glassflow-host")
client.disable_tracking()

The tracking collects anonymous information about:

• SDK version
• Platform (operating system)
• Python version
• Pipeline ID
• Whether joins or deduplication are enabled
• Kafka security protocol, auth mechanism used and whether authentication is disabled
• Errors during pipeline creation and deletion

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