Skip to content

Commit 34f9944

Browse files
ashish-bagriashish-bagri
authored
Merge pull request #66 from glassflow/add-support-for-pipeline-v3
ETL-811: migrate to pipeline v3
2 parents f61e2ce + b83dad9 commit 34f9944

44 files changed

Lines changed: 4509 additions & 3400 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎README.md‎

Lines changed: 105 additions & 131 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
<img src="https://github.com/glassflow/glassflow-python-sdk/workflows/Test/badge.svg?labelColor=&color=e69e3a">
1616
</a>
1717
<!-- Pytest Coverage Comment:Begin -->
18-
<img src="https://img.shields.io/badge/coverage-90%25-brightgreen">
18+
<img src="https://img.shields.io/badge/coverage-94%25-brightgreen">
1919
<!-- Pytest Coverage Comment:End -->
2020
</p>
2121

@@ -24,9 +24,13 @@ A Python SDK for creating and managing data pipelines between Kafka and ClickHou
2424
## Features
2525

2626
- Create and manage data pipelines between Kafka and ClickHouse
27-
- Deduplication of events during a time window based on a key
28-
- Temporal joins between topics based on a common key with a given time window
27+
- Ingest from Kafka sources or OTLP signals (logs, metrics, traces)
28+
- Unified transforms pipeline: dedup, filter, and stateless transformations
29+
- Temporal joins between sources based on a common key with a given time window
30+
- Per-source Schema Registry integration
31+
- Pipeline configuration via YAML or JSON
2932
- Schema validation and configuration management
33+
- Fine-grained resource control per pipeline component
3034

3135
## Installation
3236

@@ -41,191 +45,161 @@ pip install glassflow
4145
```python
4246
from glassflow.etl import Client
4347

44-
# Initialize GlassFlow client
4548
client = Client(host="your-glassflow-etl-url")
4649
```
4750

4851
### Create a pipeline
4952

53+
The example below uses pipeline version `v3`. See [Migrating from V2 to V3](#migrating-from-v2-to-v3) if you have existing `v2` configurations.
54+
5055
```python
5156
pipeline_config = {
52-
"version": "v2",
57+
"version": "v3",
5358
"pipeline_id": "my-pipeline-id",
54-
"source": {
55-
"type": "kafka",
56-
"connection_params": {
57-
"brokers": [
58-
"http://my.kafka.broker:9093"
59-
],
60-
"protocol": "PLAINTEXT",
61-
"mechanism": "NO_AUTH"
62-
},
63-
"topics": [
59+
"sources": [
6460
{
65-
"consumer_group_initial_offset": "latest",
66-
"name": "users",
67-
"deduplication": {
68-
"enabled": True,
69-
"id_field": "event_id",
70-
"id_field_type": "string",
71-
"time_window": "1h"
72-
}
61+
"type": "kafka",
62+
"source_id": "users",
63+
"connection_params": {
64+
"brokers": ["my.kafka.broker:9093"],
65+
"protocol": "PLAINTEXT",
66+
},
67+
"topic": "users",
68+
"consumer_group_initial_offset": "latest",
69+
"schema_fields": [
70+
{"name": "event_id", "type": "string"},
71+
{"name": "user_id", "type": "string"},
72+
{"name": "created_at", "type": "string"},
73+
{"name": "name", "type": "string"},
74+
{"name": "email", "type": "string"},
75+
],
7376
}
74-
]
75-
},
76-
"join": {
77-
"enabled": False
78-
},
79-
"sink": {
80-
"type": "clickhouse",
81-
"host": "http://my.clickhouse.server",
82-
"port": "9000",
83-
"database": "default",
84-
"username": "default",
85-
"password": "c2VjcmV0",
86-
"secure": False,
87-
"max_batch_size": 1000,
88-
"max_delay_time": "30s",
89-
"table": "users_dedup"
90-
},
91-
"schema": {
92-
"fields": [
93-
{
94-
"source_id": "users",
95-
"name": "event_id",
96-
"type": "string",
97-
"column_name": "event_id",
98-
"column_type": "UUID"
99-
},
100-
{
101-
"source_id": "users",
102-
"field_name": "user_id",
103-
"column_name": "user_id",
104-
"column_type": "UUID"
105-
},
106-
{
107-
"source_id": "users",
108-
"name": "created_at",
109-
"type": "string",
110-
"column_name": "created_at",
111-
"column_type": "DateTime"
112-
},
77+
],
78+
"transforms": [
11379
{
114-
"source_id": "users",
115-
"name": "name",
116-
"type": "string",
117-
"column_name": "name",
118-
"column_type": "String"
119-
},
120-
{
121-
"source_id": "users",
122-
"name": "email",
123-
"type": "string",
124-
"column_name": "email",
125-
"column_type": "String"
80+
"type": "dedup",
81+
"source_id": "users",
82+
"config": {
83+
"key": "event_id",
84+
"time_window": "1h",
85+
},
12686
}
127-
]
128-
}
87+
],
88+
"sink": {
89+
"type": "clickhouse",
90+
"connection_params": {
91+
"host": "my.clickhouse.server",
92+
"port": "9000",
93+
"database": "default",
94+
"username": "default",
95+
"password": "mysecret",
96+
"secure": False,
97+
},
98+
"table": "users",
99+
"mapping": [
100+
{"name": "event_id", "column_name": "event_id", "column_type": "UUID"},
101+
{"name": "user_id", "column_name": "user_id", "column_type": "UUID"},
102+
{"name": "created_at", "column_name": "created_at", "column_type": "DateTime"},
103+
{"name": "name", "column_name": "name", "column_type": "String"},
104+
{"name": "email", "column_name": "email", "column_type": "String"},
105+
],
106+
},
129107
}
130108

131-
# Create a pipeline
132109
pipeline = client.create_pipeline(pipeline_config)
133110
```
134111

135-
136-
## Get pipeline
112+
You can also load configurations from YAML or JSON files:
137113

138114
```python
139-
# Get a pipeline by ID
140-
pipeline = client.get_pipeline("my-pipeline-id")
115+
pipeline = client.create_pipeline(
116+
pipeline_config_yaml_path="pipeline.yaml"
117+
)
118+
# or
119+
pipeline = client.create_pipeline(
120+
pipeline_config_json_path="pipeline.json"
121+
)
141122
```
142123

143-
### List pipelines
124+
For full configuration reference — including Schema Registry, joins, OTLP sources, and resource controls — see the [GlassFlow docs](https://docs.glassflow.dev/configuration/pipeline-json-reference).
144125

145-
```python
146-
pipelines = client.list_pipelines()
147-
for pipeline in pipelines:
148-
print(f"Pipeline ID: {pipeline['pipeline_id']}")
149-
print(f"Name: {pipeline['name']}")
150-
print(f"Transformation Type: {pipeline['transformation_type']}")
151-
print(f"Created At: {pipeline['created_at']}")
152-
print(f"State: {pipeline['state']}")
153-
```
154-
155-
### Stop / Terminate / Resume Pipeline
126+
### Get pipeline
156127

157128
```python
158129
pipeline = client.get_pipeline("my-pipeline-id")
159-
pipeline.stop()
160-
print(pipeline.status)
161130
```
162131

163-
```
164-
STOPPING
165-
```
132+
### List pipelines
166133

167134
```python
168-
# Stop a pipeline ungracefully (terminate)
169-
client.stop_pipeline("my-pipeline-id", terminate=True)
170-
print(pipeline.status)
135+
pipelines = client.list_pipelines()
136+
for pipeline in pipelines:
137+
print(f"Pipeline ID: {pipeline['pipeline_id']}, State: {pipeline['state']}")
171138
```
172139

173-
```
174-
TERMINATING
175-
```
140+
### Stop / Terminate / Resume pipeline
176141

177142
```python
178143
pipeline = client.get_pipeline("my-pipeline-id")
179-
pipeline.resume()
180-
print(pipeline.status)
181-
```
182144

183-
```
184-
RESUMING
145+
pipeline.stop() # graceful stop → STOPPING
146+
client.stop_pipeline("my-pipeline-id", terminate=True) # ungraceful → TERMINATING
147+
pipeline.resume() # restart → RESUMING
185148
```
186149

187150
### Delete pipeline
188151

189152
Only stopped or terminated pipelines can be deleted.
190153

191154
```python
192-
# Delete a pipeline
193155
client.delete_pipeline("my-pipeline-id")
194-
195-
# Or delete via pipeline instance
156+
# or
196157
pipeline.delete()
197158
```
198159

199-
## Pipeline Configuration
160+
## Migrating from V2 to V3
161+
162+
Pipeline version `v2` has been removed. Use `Client.migrate_pipeline_v2_to_v3()` to convert an existing configuration automatically:
200163

201-
For detailed information about the pipeline configuration, see [GlassFlow docs](https://docs.glassflow.dev/configuration/pipeline-json-reference).
164+
```python
165+
from glassflow.etl import Client
166+
167+
client = Client(host="your-glassflow-etl-url")
168+
v2_config = ... # your existing v2 pipeline config dict
169+
v3_config = client.migrate_pipeline_v2_to_v3(v2_config)
170+
pipeline = client.create_pipeline(v3_config)
171+
```
172+
173+
If you prefer to migrate manually, the key changes are:
174+
175+
| Area | V2 | V3 |
176+
|------|----|----|
177+
| `version` | `"v2"` | `"v3"` |
178+
| Sources | `source: {type, connection_params, topics: [...]}` | `sources: [{type, source_id, connection_params, topic, ...}]` flat list |
179+
| Schema | top-level `schema.fields` block | `sources[].schema_fields` per source |
180+
| Deduplication | per-topic `deduplication: {enabled, id_field, ...}` | `transforms: [{type: "dedup", source_id, config: {key, time_window}}]` |
181+
| Filter | top-level `filter: {enabled, expression}` | `transforms: [{type: "filter", source_id, config: {expression}}]` |
182+
| Transformation | top-level `stateless_transformation` | `transforms: [{type: "stateless", source_id, config: {transforms: [...]}}]` |
183+
| Join | `join.sources: [{source_id, key, orientation}]` | `join: {left_source: {...}, right_source: {...}, output_fields: [...]}` |
184+
| Sink connection | flat fields (`host`, `port`, ...) at top level | nested `sink.connection_params` object |
185+
| Sink field mapping | top-level `schema.fields` with `source_id` | `sink.mapping` list of `{name, column_name, column_type}` |
186+
| Resources | `pipeline_resources: {ingestor, transform, ...}` | `resources: {sources: [...], transform: [...], ...}` |
187+
| Sink password | base64-encoded | plain text |
202188

203189
## Tracking
204190

205-
The SDK includes anonymous usage tracking to help improve the product. Tracking is enabled by default but can be disabled in two ways:
191+
The SDK includes anonymous usage stats collection to help improve the product. It collects non-identifying information such as SDK version, Python version, and feature flags (e.g., whether joins or deduplication are enabled). No personally identifiable information is collected.
192+
193+
Usage states collection is enabled by default. To disable it:
206194

207-
1. Using an environment variable:
208195
```bash
209-
export GF_TRACKING_ENABLED=false
196+
export GF_USAGESTATS_ENABLED=false
210197
```
211198

212-
2. Programmatically using the `disable_tracking` method:
213199
```python
214-
from glassflow.etl import Client
215-
216-
client = Client(host="my-glassflow-host")
217-
client.disable_tracking()
200+
client.disable_usagestats()
218201
```
219202

220-
The tracking collects anonymous information about:
221-
- SDK version
222-
- Platform (operating system)
223-
- Python version
224-
- Pipeline ID
225-
- Whether joins or deduplication are enabled
226-
- Kafka security protocol, auth mechanism used and whether authentication is disabled
227-
- Errors during pipeline creation and deletion
228-
229203
## Development
230204

231205
### Setup

‎VERSION‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
3.8.0
1+
4.0.0

0 commit comments

Comments
 (0)