数据管道工具箱

数据管道工具箱

快速构建ETL数据管道：提取 → 转换 → 加载 → 调度

核心功能 多源提取 — REST APIs、GraphQL、SQL数据库、CSV/JSON/Parquet文件、S3/云存储、Kafka/SQS 数据转换 — 清洗、过滤、聚合、关联、跨表Join 多目标加载 — PostgreSQL/MySQL、Snowflake/Big查询、S3、数据仓库 定时调度 — Cron任务或事件触发 监控告警 — 失败自动通知，可视化运行状态 快速开始 # 创建数据管道 ./流水线.sh 创建 my-流水线

# 添加数据源 ./流水线.sh 提取 my-流水线 API --url https://API.example.com/data

# 添加转换规则 ./流水线.sh 转换 my-流水线 过滤器 "状态 == 'active'" ./流水线.sh 转换 my-流水线 聚合 "group by category, sum(amount)"

# 添加目标存储 ./流水线.sh load my-流水线 postgres --connection $DATABASE_URL

# 运行管道 ./流水线.sh 运行 my-流水线

支持的数据源 类型 具体来源 APIs REST API, GraphQL, 内部服务 数据库 PostgreSQL, MySQL, MongoDB, SQL Server 文件 CSV, JSON, Parquet, Excel 云存储 AWS S3, Google Cloud Storage 消息队列 Kafka, AWS SQS 支持的目标存储 类型 具体目标 数据库 PostgreSQL, MySQL, Big查询, Snowflake 数据仓库 命令行工具ckHouse, DuckDB, Time扩展DB 文件存储 S3, GCS, 本地文件 API 第三方API回传 典型使用场景 场景1：每日销售数据汇总 # 从CRM API提取昨日销售数据 ./流水线.sh 提取 dAIly-sales API \ --url "https://crm.example.com/API/orders?date=yesterday"

# 转换：按产品分类汇总 ./流水线.sh 转换 dAIly-sales 聚合 \ --group-by "product_category" \ --sum "quantity,amount"

# 加载到数据仓库 ./流水线.sh load dAIly-sales big查询 \ --project "my-project" --data设置 "sales" --table "dAIly_summary"

# 设置每日定时任务 ./流水线.sh schedule dAIly-sales "0 6 "

场景2：用户行为数据同步 # 从日志文件提取 ./流水线.sh 提取 user-记录s file --path "/var/记录s/应用/.json"

# 清洗和转换 ./流水线.sh 转换 user-记录s 过滤器 "event_type != 'heartbeat'" ./流水线.sh 转换 user-记录s 添加-column "timestamp:解析_timestamp(time)"

# 加载到命令行工具ckHouse ./流水线.sh load user-记录s 命令行工具ckhouse --connection $CH_URL

监控与告警 查看运行状态 ./流水线.sh 状态 my-流水线 # 输出： # 状态: ✅ 运行ning # Last 运行: 2026-05-05 06:00:00 # Duration: 45s # Records Processed: 12,847 # Errors: 0

配置告警 # 失败时发送邮件 ./流水线.sh alert my-流水线 emAIl --to admin@example.com

# 失败时发送飞书消息 ./流水线.sh alert my-流水线 网页hook --url "https://open.feishu.cn/..."

推荐资源 ShadowAI API（数据管道配套）: https://referer.shadowAI.xyz/r/1056448

由 AI智造工坊 (http://AI.qnitgroup.com) 整理发布 | 安装源: ClawHub