mcpField
E
etl-pipelines

etl-pipelines

github.com/alphaonedev数据与数据库44 次安装
代码仓库

关于 etl-pipelines

etl-pipelines 是面向 OpenClaw 的 ETL 工作流设计与实现技能，适用于结构化数据源。它涵盖提取、转换、加载、调度、增量更新及错误处理。

主要功能

  1. 01从文件、数据库和 API 提取数据
  2. 02使用筛选、聚合或 SQL 转换
  3. 03加载时校验模式并记录错误
  4. 04调度管道并连接编排工具
  5. 05支持增量加载和 CDC
  6. 06校验、重试并监控管道

适用场景

  1. 01将筛选后的 CSV 数据载入 PostgreSQL
  2. 02将数据库增量更新写入 BigQuery
  3. 03为 BI 仪表板构建定时数据摄取
  4. 04在系统间迁移结构化数据

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/alphaonedev/openclaw-graph --skill etl-pipelines

安装后在 Claude Code 中输入/etl-pipelines确认已加载。

详细介绍

etl-pipelines

用途

此技能可帮助 OpenClaw 在数据工程工作流中设计和实现用于数据提取、转换和加载的 ETL 流水线。它侧重于处理数据库、文件和 API 等结构化数据源，确保为分析和报告提供高效的数据流。

适用场景

构建用于批处理、实时数据摄取或数据迁移的数据流水线时，可使用此技能。适用于涉及大型数据集（例如 >1TB）、与 Apache Spark 或 AWS Glue 等工具集成，或为 BI 仪表板自动执行 ETL 的场景。

核心能力
  • 使用连接器（例如用于数据库的 JDBC）从 CSV、JSON 文件、SQL 数据库或 API 等来源提取数据。
  • 通过筛选、聚合或 SQL 查询等操作转换数据（例如使用 Pandas 或 Spark DataFrames）。
  • 将数据加载到 PostgreSQL、BigQuery 或 S3 存储桶等目标，并进行架构验证和错误日志记录。
  • 支持使用类似 cron 的表达式调度流水线，或与 Airflow 等编排工具集成。
  • 通过跟踪上次处理的时间戳或变更数据捕获（CDC）来处理增量加载。
使用模式

要使用此技能，请使用特定的 ETL 命令调用 OpenClaw。首先定义 JSON 格式的流水线配置，然后通过 CLI 或 API 执行。例如，传入如下配置文件：

{
  "source": {"type": "file", "path": "data/input.csv"},
  "transform": {"operations": ["filter column='id' > 100"]},
  "destination": {"type": "postgres", "table": "processed_data"}
}

以模块化方式构建流水线：先提取数据，然后在内存或分布式环境中进行转换，最后通过重试机制加载数据。务必设置用于身份验证的环境变量，例如 export OPENCLAW_API_KEY=your_key。

常用命令/API

使用 OpenClaw CLI 执行 ETL 任务。例如：

  • 创建流水线：openclaw etl create --config path/to/config.json --env $OPENCLAW_API_KEY （标志：--config 用于指定 JSON 文件，--env 用于身份验证；输出流水线 ID）。
  • 运行流水线：openclaw etl run <pipeline-id> --params '{"batch_size": 1000}' （标志：--params 用于运行时覆盖；通过 stdout 监控进度）。
  • API 端点：使用 POST /v1/etl/pipelines 创建流水线，请求正文为 JSON 配置；使用 GET /v1/etl/pipelines/{id} 获取状态。 进行代码集成时，使用 OpenClaw 的 Python SDK：
import openclaw
client = openclaw.Client(api_key=os.environ['OPENCLAW_API_KEY'])
pipeline = client.etl.create(config={'source': 'file.csv', 'transform': 'sql_query'})
client.etl.run(pipeline.id)

执行前，务必使用 openclaw etl validate --file path/to/config.json 验证配置。

集成说明

通过在配置中引用依赖项，将此技能与数据工具集成，例如指定 "engine": "spark" 以进行分布式处理。对于 AWS，请设置 $AWS_ACCESS_KEY_ID 等环境变量，并将 S3 等连接器用作数据源。传递输出结果，将此技能与其他 OpenClaw 技能串联使用，例如将 ETL 结果传递给机器学习技能。确保数据格式兼容（例如，大数据场景使用 Parquet）。在多工具设置中，可使用 Webhook：配置 POST /v1/etl/webhook，以便在外部事件发生时触发。

错误处理

通过使用 try-catch 块包装命令，或使用 --retry 3 等内置标志，在发生暂时性故障（例如网络问题）时自动重试。检查 API 响应中的错误代码（例如，配置错误为 400，服务端错误为 500），并记录详细信息。在代码中使用：

try:
    client.etl.run(pipeline.id)
except openclaw.EtlError as e:
    print(f"Error: {e.code} - {e.message}; Retrying...")

使用 openclaw etl logs <pipeline-id> 监控日志，并设置失败阈值，例如验证失败的记录超过 10% 时中止。使用环境变量配置自定义错误处理程序，例如 $ETL_ERROR_WEBHOOK_URL。

具体用法示例

  1. 从 CSV 文件提取数据，使用 SQL 转换，然后加载到 PostgreSQL：
    首先，创建配置：{"source": {"type": "file", "path": "sales.csv"}, "transform": {"sql": "SELECT * FROM data WHERE amount > 100"}, "destination": {"type": "postgres", "table": "sales_filtered", "conn_str": "dbname=mydb"}}
    然后运行：openclaw etl create --config sales_config.json; openclaw etl run <id> --env $OPENCLAW_API_KEY
    在标准配置下，这会在 5 分钟内处理 100 万行数据。

  2. 将数据库源的增量 ETL 数据加载到 BigQuery：
    配置：{"source": {"type": "mysql", "query": "SELECT * FROM orders WHERE updated_at > '2023-01-01'"}, "transform": {"operations": ["add_column": "processed_at=NOW()"]}, "destination": {"type": "bigquery", "dataset": "my_dataset", "table": "orders_incremental"}}
    执行：openclaw etl run <pipeline-id> --params '{"incremental": true}'
    这会处理每日更新，仅追加新记录。

图关系

  • 相关领域：数据工程集群（例如，数据仓库、大数据处理等技能）
  • 通过标签关联：etl（关联到数据管道）、data-engineering（关联到分析工具）
  • 依赖项：需要使用 $OPENCLAW_API_KEY 进行身份验证；可与 Spark 或 Airflow 等外部工具集成以进行编排。

评论

支持 Markdown