Show HN: GlassFlow – OSS streaming dedup and joins from Kafka to ClickHouse

Source | HN Comments

GlassFlow 是一个开源的流式数据处理工具，主要功能是实现 Kafka 数据到 ClickHouse 的实时去重和时间 Join。它提供 Web UI 用于配置和管理数据管道，支持流式去重、时间 Join、内置 Kafka 连接器、ClickHouse 优化、Docker 部署等特性。用户可以通过配置连接参数、选择 Kafka 输入流、定义去重规则和 Join 规则来创建数据管道，并进行本地测试。该项目基于 Apache License 2.0 授权，欢迎贡献。

Show HN: GlassFlow – 开源流式数据去重与 Kafka 到 ClickHouse 的 Join

跳至内容

glassflow / **clickhouse-etl ** Public

Notifications You must be signed in to change notification settings
Fork 2
Star 82

实时去重和流式数据的时间 Join

glassflow.dev/

License

Apache-2.0 license 82 stars 2 forks Branches Tags Activity Star Notifications You must be signed in to change notification settings

glassflow/clickhouse-etl

main Branches Tags Go to file Code

Folders and files

Name| Name| Last commit message| Last commit date
---|---|---|---

Latest commit

History

Repository files navigation

Report Bug · Feature Request · Get Help Join our weekly office hours every Wednesday 15:00-18:00 CET Email Support

GlassFlow for ClickHouse Streaming ETL

GlassFlow for ClickHouse Streaming ETL 是一个实时流处理器，旨在简化 Kafka 和 ClickHouse 之间的数据管道创建和管理。它提供了一个强大的、用户友好的界面，用于构建和管理具有内置去重和时间 Join 支持的实时数据管道。

GlassFlow 专为数据工程师构建，可以处理迟到的事件，确保 exactly-once 的正确性，并通过高吞吐量数据进行扩展。它可以从流式数据中提供准确的、低延迟的结果，而不会影响简单性或性能。该工具的直观 Web 界面使配置和监控管道变得容易，而其强大的架构可确保可靠的数据处理。

主要特性

流式去重 :
- 在提取到 ClickHouse 之前，实时去重 Kafka 流
- 可配置的时间窗口，最长可达 7 天，用于去重
- 简单的去重键和时间窗口配置
- 一键设置去重的数据管道
- 阻止重复数据到达 ClickHouse
时间流 Join :
- 实时 Join 两个 Kafka 流
- 可配置的时间窗口，最长可达 7 天，用于流 Join
- 通过 UI 配置 Join 键和时间窗口
- 简化的 Join 设置过程
- 生成已 Join 的流，可供 ClickHouse 提取
内置 Kafka 连接器 :
- 由 NATS-Kafka Bridge 提供支持
- 从 Kafka 主题自动提取数据
- 与 Kafka 集群无缝集成
- 无需手动拉取数据
- 支持多个 Kafka 主题和分区
- 原生支持 JSON 数据类型
优化 ClickHouse Sink :
- 本地 ClickHouse 连接，实现最佳性能
- 可配置的批量大小，实现高效的数据提取
- 可调节的等待时间，实现最佳吞吐量
- 内置重试机制
- 自动模式检测和管理
- 完全支持 ClickHouse 中的 JSON 数据类型
用户友好的界面 : 基于 Web 的 UI，用于管道配置和管理
本地开发 : 包括带有本地 Kafka 和 ClickHouse 实例的演示设置
Docker 支持 : 使用 Docker 和 docker-compose 轻松部署
自托管 : 可以在您的基础设施中自托管的开源解决方案

快速开始

前提条件

Docker 和 Docker Compose
Git

本地开发设置

克隆存储库:

git clone https://github.com/glassflow/clickhouse-etl.git
cd clickhouse-etl

使用 Docker Compose 启动服务:

docker-compose up

访问 http://localhost:8080 上的 Web 界面以配置您的管道:

连接设置 :
- 配置 Kafka 源连接参数
- 配置 ClickHouse sink 连接参数
管道配置 :
- 选择 Kafka 输入流
- 定义去重键和时间窗口
- 对于 Join：为每个流指定 Join 键和时间窗口
- 配置 ClickHouse 表设置
管道管理 :
- 启动/停止管道
- 检查管道日志

用法

创建管道

访问 http://localhost:8080 上的 Web 界面
配置您的源 (Kafka) 和 sink (ClickHouse) 连接
定义您的管道转换:
- 设置去重规则
- 配置流之间的时间 Join
- 定义数据转换
启动管道并监控其进度

本地测试

该项目在 demos 文件夹中包含一个全面的演示设置，该设置提供了完整的本地测试环境。此设置包括:

带有 Zookeeper 的本地 Kafka 实例
本地 ClickHouse 实例
示例数据生成
示例管道配置
运行演示的分步说明

有关设置和运行本地测试环境的详细说明，请参阅 Demo README file.

架构

该项目包含几个关键组件:

GlassFlow API : 用 Go 编写的核心 ETL 引擎
Web UI : 用于管道管理的用户界面
NATS : 用于内部通信的消息代理
Kafka Bridge : 用于 Kafka 集成的连接器，使用 NATS-Kafka Bridge

有关第三方组件及其许可证的信息，请参阅我们的 NOTICE 文件。

管道配置

管道配置在 JSON 文件中定义，该文件指定源、sink 和任何转换。虽然 Web 界面会自动生成此配置，但了解其结构可能对高级用户有所帮助。

管道配置是一个 JSON 对象，用于定义数据如何从 Kafka 主题流向 ClickHouse 表。它由三个主要组件组成：

源配置 (Kafka)
Sink 配置 (ClickHouse)
Join 配置 (可选)

根配置

字段 | 类型 | 必需 | 描述
---|---|---|---
pipeline_id | string | 是 | 管道的唯一标识符。必须为非空。
source | object | 是 | Kafka 源的配置。请参阅源配置。
sink | object | 是 | ClickHouse sink 的配置。请参阅Sink 配置。
join | object | 否 | 用于 Join 多个 Kafka 主题的配置。请参阅Join 配置。

源配置

源配置定义了如何连接到 Kafka 主题并从中消费。字段 | 类型 | 必需 | 描述
---|---|---|---
type | string | 是 | "kafka" 是唯一支持的源
provider | string | 否 | Kafka 提供者，例如 "aiven"
topics | array | 是 | 要消费的 Kafka 主题列表。请参阅主题配置。
connection_params | object | 是 | Kafka 连接参数。请参阅连接参数。

连接参数

字段 | 类型 | 必需 | 描述
---|---|---|---
brokers | array | 是 | Kafka 代理地址列表 (例如，["localhost:9092"])。
protocol | string | 是 | Kafka 连接的安全协议 (例如，"SASL_SSL")。
mechanism | string | 是 | 身份验证机制 (例如，"SCRAM-SHA-256")。
username | string | 是 | Kafka 身份验证的用户名。
password | string | 是 | Kafka 身份验证的密码。
root_ca | string | 否 | Kafka 身份验证的 Cert 文件。

主题配置

topics 数组中的每个主题都具有以下配置: 字段 | 类型 | 必需 | 描述
---|---|---|---
name | string | 是 | Kafka 主题的名称。
consumer_group_initial_offset | string | 是 | 消费者组的初始偏移量("earliest" 或 "newest")。
schema | object | 是 | 事件模式定义。请参阅模式配置。
deduplication | object | 是 | 去重设置。请参阅去重配置。

模式配置

字段 | 类型 | 必需 | 描述
---|---|---|---
type | string | 是 | 模式类型(目前仅支持 "json")。
fields | array | 是 | 字段定义列表。请参阅字段配置。

字段配置

字段 | 类型 | 必需 | 描述
---|---|---|---
name | string | 是 | 字段名称。
type | string | 是 | 字段类型 (例如，"String"、"Integer")。

去重配置

字段 | 类型 | 必需 | 描述
---|---|---|---
enabled | boolean | 是 | 是否启用去重。
id_field | string | 是 | 用于消息去重的字段名称。
id_field_type | string | 是 | ID 字段的类型 (例如，"string")。
time_window | string | 是 | 去重的时间窗口 (例如，"1h" 表示一小时)。

Sink 配置

Sink 配置定义了如何连接到 ClickHouse 并向其写入。字段 | 类型 | 必需 | 描述
---|---|---|---
type | string | 是 | 必须为 "clickhouse"。
host | string | 是 | ClickHouse 服务器主机名。
port | integer | 是 | ClickHouse 服务器端口。
database | string | 是 | ClickHouse 数据库名称。
username | string | 是 | ClickHouse 用户名。
password | string | 是 | ClickHouse 密码。
table | string | 是 | 目标表名称。
secure | boolean | 否 | 是否使用安全连接。默认为 false。
max_batch_size | integer | 否 | 写入之前要批处理的最大记录数。默认为 1000。
max_delay_time | string | 否 | 消息刷新到 sink 之前的最大延迟时间。默认为 "10m"。
table_mapping | array | 是 | 字段到列的映射列表。请参阅表映射配置。

表映射配置

table_mapping 数组中的每个映射都具有以下配置: 字段 | 类型 | 必需 | 描述
---|---|---|---
source_id | string | 是 | 源主题的名称。
field_name | string | 是 | 源字段名称。
column_name | string | 是 | 目标列名称。
column_type | string | 是 | 目标列类型。

Join 配置

Join 配置定义了如何 Join 来自多个 Kafka 主题的数据。字段 | 类型 | 必需 | 描述
---|---|---|---
enabled | boolean | 是 | 是否启用 Join。
type | string | 是 | Join 类型 (例如，"temporal")。
sources | array | 是 | 要 Join 的源列表。请参阅Join 源配置。

Join 源配置

sources 数组中的每个源都具有以下配置: 字段 | 类型 | 必需 | 描述
---|---|---|---
source_id | string | 是 | 要 Join 的 Kafka 主题的名称。
join_key | string | 是 | 用于 Join 记录的字段名称。
time_window | string | 是 | 用于 Join 记录的时间窗口 (例如，"1h" 表示一小时)。
orientation | string | 是 | Join 方向("left" 或 "right")。

示例配置

{
 "pipeline_id": "kafka-to-clickhouse-pipeline",
 "source": {
  "type": "kafka",
  "provider": "aiven",
  "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",
    "schema": {
     "type": "json",
     "fields": [
      {
       "name": "session_id",
       "type": "string"
      },
      {
       "name": "user_id",
       "type": "string"
      },
      {
       "name": "timestamp",
       "type": "datetime"
      }
     ]
    },
    "deduplication": {
     "enabled": true,
     "id_field": "session_id",
     "id_field_type": "string",
     "time_window": "12h"
    }
   },
   {
    "consumer_group_initial_offset": "earliest",
    "name": "orders",
    "schema": {
     "type": "json",
     "fields": [
      {
       "name": "user_id",
       "type": "string"
      },
      {
       "name": "order_id",
       "type": "string"
      },
      {
       "name": "timestamp",
       "type": "datetime"
      }
     ]
    },
    "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",
  "table_mapping": [
   {
    "source_id": "user_logins",
    "field_name": "session_id",
    "column_name": "session_id",
    "column_type": "UUID"
   },
   {
    "source_id": "user_logins",
    "field_name": "user_id",
    "column_name": "user_id",
    "column_type": "UUID"
   },
   {
    "source_id": "orders",
    "field_name": "order_id",
    "column_name": "order_id",
    "column_type": "UUID"
   },
   {
    "source_id": "user_logins",
    "field_name": "timestamp",
    "column_name": "login_at",
    "column_type": "DataTime"
   },
   {
    "source_id": "orders",
    "field_name": "timestamp",
    "column_name": "order_placed_at",
    "column_type": "DateTime"
   }
  ]
 }
}

💡 注意 : Web 界面会根据用户输入自动生成此配置，因此无需手动编辑。

贡献

欢迎贡献! 请参阅我们的 Contributing Guidelines 了解详情。

许可证

本项目采用 Apache License 2.0 授权。

关于

Real-time deduplication and temporal joins for streaming data glassflow.dev/

Releases 7

v0.0.1 Latest Apr 28, 2025 + 6 releases

Packages 0

No packages published

Show HN: GlassFlow – OSS streaming dedup and joins from Kafka to ClickHouse

Show HN: GlassFlow – 开源流式数据去重与 Kafka 到 ClickHouse 的 Join

License

glassflow/clickhouse-etl

Folders and files

Latest commit

History

Repository files navigation

GlassFlow for ClickHouse Streaming ETL

主要特性

快速开始

前提条件

本地开发设置

用法

创建管道

本地测试

架构

管道配置

根配置

源配置

连接参数

主题配置

模式配置

字段配置

去重配置

Sink 配置

表映射配置

Join 配置

Join 源配置

示例配置

贡献

许可证

关于

Resources

License

Code of conduct

Stars

Watchers

Forks

Releases 7

Packages 0

[Contributors 5](https://github