data-lakehouse

data-lakehouse

用途

此技能可用于设计和实现数据湖仓架构，将数据湖与数据仓库功能相结合，以实现可扩展的大数据存储与分析。可在 Delta Lake 或 Iceberg 等平台上管理 PB 级数据，并支持 ACID 事务、架构演进和优化的查询性能。

适用场景

• 处理来自 S3 或 Kafka 等来源的大规模数据摄取，同时需要原始存储和结构化查询时。
• 面向需要实时更新的分析工作负载，例如电子商务中的 ETL 工作流或 IoT 数据处理。
• 将 Spark 或 Presto 集成用于非结构化数据的 SQL 分析时。

核心功能

• 架构设计：生成湖仓架构蓝图，包括分区策略和元数据管理（例如，使用 Iceberg 作为表格式）。
• 数据摄取：支持使用 Apache Spark 等工具进行批量和流式数据摄取，并处理 Parquet 或 ORC 等格式。
• 查询优化：实现缓存和索引以加速查询，例如创建带有 Z-order 聚类的 Delta Lake 表。
• 可扩展性：通过云 API 自动扩展存储和计算资源，例如使用 AWS Glue 执行 ETL 作业。
• 安全性：通过 AWS Lake Formation 授权等策略实施行级访问控制。

使用模式

• 模式 1：设置新的湖仓时，调用该技能生成配置文件，然后用它初始化存储。例如：从 CSV 数据创建 Delta Lake 表。
• 模式 2：在分析工作流中，使用该技能通过添加索引来优化查询，然后通过 Spark SQL 执行查询。
• 模式 3：维护时，定期检查现有表的架构演进和合并操作。

常用命令/API

使用 OpenClaw CLI 或 API 调用此技能。身份验证需要将 $DATA_LAKEHOUSE_API_KEY 设置为环境变量。

• CLI 命令：初始化湖仓项目：

  openclaw skill data-lakehouse init --project my-lakehouse --storage s3://my-bucket --engine delta
  

  这会创建包含 S3 存储桶和 Delta Lake 引擎的基本配置文件。

• API 端点：通过 POST 请求创建表：

  curl -H "Authorization: Bearer $DATA_LAKEHOUSE_API_KEY" \
       -d '{"table_name": "sales_data", "format": "parquet", "partition_by": ["date"]}' \
       https://api.openclaw.ai/data-lakehouse/tables
  

  响应包含可立即使用的表元数据。

• 代码片段：在 Python 中与 Spark 集成：

  from pyspark.sql import SparkSession
  spark = SparkSession.builder.appName("lakehouse").getOrCreate()
  df = spark.read.format("delta").load("s3://my-bucket/sales_data")
  df.write.format("delta").mode("append").save("s3://my-bucket/sales_data")
  

  这会将数据追加到 Delta 表；请确保 Spark 已配置 AWS 凭证。

• 配置格式：湖仓配置使用 JSON，例如：

  {
    "storage": "s3://my-bucket",
    "engine": "iceberg",
    "auth": {"key": "$DATA_LAKEHOUSE_API_KEY"}
  }
  

  使用 CLI 加载此配置：openclaw skill data-lakehouse apply --config path/to/config.json。

集成说明

• 与其他技能集成：传递输出以关联“big-data”技能，例如使用 Spark 流式处理，将 Kafka 摄取技能中的数据导入此技能。
• 外部工具：设置存储桶策略以与 AWS S3 集成；在 CLI 命令中使用 --aws-region us-west-2。对于 Spark，请确保环境中包含 spark.delta 等依赖项。
• API 集成：从其他服务调用时，针对速率限制处理重试；例如，在串联的 API 调用中使用相同的 $DATA_LAKEHOUSE_API_KEY。
• 依赖管理：始终指定版本，例如，Delta Lake 兼容性要求 Spark 3.0+。

错误处理

• 常见错误：通过检查是否已设置 $DATA_LAKEHOUSE_API_KEY 来处理身份验证失败；在脚本中使用 os.environ.get('DATA_LAKEHOUSE_API_KEY')。
• 处理步骤：对于存储访问错误（例如 S3 权限问题），添加 try-except 代码块：

  try:
      spark.read.format("delta").load("s3://my-bucket/data")
  except Exception as e:
      print(f"Error: {e}. Check bucket permissions and retry.")
  

  在 API 调用中使用指数退避重试网络问题等暂时性错误。
• 调试：使用 CLI 标志 --verbose 获取详细日志，例如 openclaw skill data-lakehouse init --verbose。使用 openclaw skill data-lakehouse validate --file config.json 验证配置。

具体使用示例

• 示例 1：构建销售分析湖仓： 首先，进行初始化：openclaw skill data-lakehouse init --project sales-ana --storage s3://sales-data。然后，导入数据：使用 API 创建表，并按照上述示例通过 Spark 追加数据。最后，使用以下命令查询：spark.sql("SELECT * FROM sales_data WHERE date > '2023-01-01'")。

• 示例 2：优化现有湖仓以处理 IoT 数据： 运行：openclaw skill data-lakehouse optimize --table iot_metrics --add-index date。此操作会添加索引；使用以下查询代码片段进行验证：df = spark.read.format("iceberg").load("s3://iot-bucket/metrics").filter(df.date > current_date())。监控优化后的性能。

图关系

• 相关内容："big-data"（共享数据处理流水线标签）、"data-engineering"（ETL 工作流的集群关联）。
• 关联方式：通过 "data-lakehouse" 等标签进行跨技能查询，并通过 "data-engineering" 集群执行顺序任务流。