warehouse-init

初始化数据仓库架构

生成全面且可由用户编辑的数据仓库架构参考文件。

脚本： ../analyzing-data/scripts/ — 以下所有 CLI 命令均以 analyzing-data 技能目录为相对路径。运行任何 scripts/cli.py 命令前，请先相对于此文件执行 cd 到 ../analyzing-data/。

此操作的作用

1. 从数据仓库中发现所有数据库、架构、表和列
2. 结合代码库上下文丰富信息（dbt 模型、gusty SQL、架构文档）
3. 记录行数并识别大型表
4. 生成 .astro/warehouse.md——可进行版本控制、可在团队间共享的参考文件
5. 无需查询数据仓库，即可即时查找概念→表的对应关系

流程

第 1 步：读取数据仓库配置

cat ~/.astro/agents/warehouse.yml

获取要发现的数据库列表（例如，databases: [HQ, ANALYTICS, RAW]）。

第 2 步：搜索代码库以查找上下文（并行）

启动一个子 Agent，在代码中查找业务上下文：

Task(
    subagent_type="Explore",
    prompt="""
    Search for data model documentation in the codebase:

    1. dbt models: **/models/**/*.yml, **/schema.yml
       - Extract table descriptions, column descriptions
       - Note primary keys and tests

    2. Gusty/declarative SQL: **/dags/**/*.sql with YAML frontmatter
       - Parse frontmatter for: description, primary_key, tests
       - Note schema mappings

    3. AGENTS.md or CLAUDE.md files with data layer documentation

    Return a mapping of:
      table_name -> {description, primary_key, important_columns, layer}
    """
)

第 3 步：并行发现数据仓库信息

使用 Task 工具为每个数据库启动一个子 Agent：

For each database in configured_databases:
    Task(
        subagent_type="general-purpose",
        prompt="""
        Discover all metadata for database {DATABASE}.

        Use the CLI to run SQL queries:
        # Scripts are relative to ../analyzing-data/
        uv run scripts/cli.py exec "df = run_sql('...')"
        uv run scripts/cli.py exec "print(df)"

        1. Query schemas:
           SELECT SCHEMA_NAME FROM {DATABASE}.INFORMATION_SCHEMA.SCHEMATA

        2. Query tables with row counts:
           SELECT TABLE_SCHEMA, TABLE_NAME, ROW_COUNT, COMMENT
           FROM {DATABASE}.INFORMATION_SCHEMA.TABLES
           ORDER BY TABLE_SCHEMA, TABLE_NAME

        3. For important schemas (MODEL_*, METRICS_*, MART_*), query columns:
           SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, COMMENT
           FROM {DATABASE}.INFORMATION_SCHEMA.COLUMNS
           WHERE TABLE_SCHEMA = 'X'

        Return a structured summary:
        - Database name
        - List of schemas with table counts
        - For each table: name, row_count, key columns
        - Flag any tables with >100M rows as "large"
        """
    )

并行运行所有子 Agent（在一条消息中发起多个 Task 调用）。

第 4 步：发现分类值族

对于关键分类列（如 OPERATOR、STATUS、TYPE、FEATURE），发现其值族：

uv run cli.py exec "df = run_sql('''
SELECT DISTINCT column_name, COUNT(*) as occurrences
FROM table
WHERE column_name IS NOT NULL
GROUP BY column_name
ORDER BY occurrences DESC
LIMIT 50
''')"
uv run cli.py exec "print(df)"

按共同前缀/后缀将相关值归为不同的族（例如，将 ExportCSV、ExportJSON、ExportParquet 归入 Export*）。

第 5 步：合并结果

合并数据仓库元数据和代码库上下文：

1. 快速参考表——概念 → 表的映射（若在代码中找到，则预先填充）
2. 分类列——关键筛选列的值族
3. 数据库章节——每个数据库各一节
4. 架构子章节——按架构对表进行分组
5. 表详情——列、行数、来自代码的描述、警告

第 6 步：生成 warehouse.md

将文件写入：

• .astro/warehouse.md（默认——特定于项目且可进行版本控制）
• ~/.astro/agents/warehouse.md（使用 --global 标志时）

输出格式

# Warehouse Schema

> Generated by `/astronomer-data:warehouse-init` on {DATE}. Edit freely to add business context.

## Quick Reference

| Concept | Table | Key Column | Date Column |
|---------|-------|------------|-------------|
| customers | HQ.MODEL_ASTRO.ORGANIZATIONS | ORG_ID | CREATED_AT |


## Categorical Columns

When filtering on these columns, explore value families first (values often have variants):

| Table | Column | Value Families |
|-------|--------|----------------|
| {TABLE} | {COLUMN} | `{PREFIX}*` ({VALUE1}, {VALUE2}, ...) |


## Data Layer Hierarchy

Query downstream first: `reporting` > `mart_*` > `metric_*` > `model_*` > `IN_*`

| Layer | Prefix | Purpose |
|-------|--------|---------|
| Reporting | `reporting.*` | Dashboard-optimized |
| Mart | `mart_*` | Combined analytics |
| Metric | `metric_*` | KPIs at various grains |
| Model | `model_*` | Cleansed sources of truth |
| Raw | `IN_*` | Source data - avoid |

## {DATABASE} Database

### {SCHEMA} Schema

#### {TABLE_NAME}
{DESCRIPTION from code if found}

| Column | Type | Description |
|--------|------|-------------|
| COL1 | VARCHAR | {from code or inferred} |

- **Rows:** {ROW_COUNT}
- **Key column:** {PRIMARY_KEY from code or inferred}
{IF ROW_COUNT > 100M: - **⚠️ WARNING:** Large table - always add date filters}

## Relationships

{根据 *_ID 等列名推断的关系}

命令选项

选项	效果
/astronomer-data:warehouse-init	生成 .astro/warehouse.md
/astronomer-data:warehouse-init --refresh	重新生成，同时保留用户编辑内容
/astronomer-data:warehouse-init --database HQ	仅发现指定数据库
/astronomer-data:warehouse-init --global	写入 ~/.astro/agents/

第 7 步：预先填充缓存

生成 warehouse.md 后，填充概念缓存：

# Scripts are relative to ../analyzing-data/
uv run cli.py concept import -p .astro/warehouse.md
uv run cli.py concept learn customers HQ.MART_CUST.CURRENT_ASTRO_CUSTS -k ACCT_ID

第 8 步：提供集成到 CLAUDE.md 的选项（询问用户）

询问用户：

是否要将快速参考表添加到 CLAUDE.md 文件中？

这样可以确保架构映射始终处于数据查询的上下文中，使复杂查询的准确率从约 25% 提升至约 100%。

选项：

1. 是，添加到 CLAUDE.md（推荐）——追加快速参考章节
2. 否，跳过——仅使用 warehouse.md 和缓存

如果用户选择“是”：

1. 检查 .claude/CLAUDE.md 或 CLAUDE.md 是否存在
2. 如果存在，追加快速参考章节（避免重复）
3. 如果不存在，创建仅包含快速参考的 .claude/CLAUDE.md

要添加的快速参考章节：

## Data Warehouse Quick Reference

When querying the warehouse, use these table mappings:

| Concept | Table | Key Column | Date Column |
|---------|-------|------------|-------------|
{rows from warehouse.md Quick Reference}

**Large tables (always filter by date):** {list tables with >100M rows}

> Auto-generated by `/astronomer-data:warehouse-init`. Run `/astronomer-data:warehouse-init --refresh` to update.

**如果用户选择“是”：**将快速参考章节追加到 .claude/CLAUDE.md 或 CLAUDE.md。

生成后

告知用户：

Generated .astro/warehouse.md

Summary:
  - {N} databases, {N} schemas, {N} tables
  - {N} tables enriched with code descriptions
  - {N} concepts cached for instant lookup

Next steps:
  1. Edit .astro/warehouse.md to add business context
  2. Commit to version control
  3. Run /astronomer-data:warehouse-init --refresh when schema changes

刷新行为

指定 --refresh 时：

1. 读取现有的 warehouse.md
2. 保留所有 HTML 注释（``）
3. 保留快速参考表条目（用户添加）
4. 保留用户添加的描述
5. 更新行数并添加新表
6. 使用 `` 注释标记已移除的表

缓存过期与架构漂移

默认情况下，运行时缓存的 TTL 为 7 天。7 天后，缓存条目将过期，并在下次使用时重新发现。

何时刷新

在以下情况下运行 /astronomer-data:warehouse-init --refresh：

• 架构变更：新增、重命名或删除表
• 列变更：新增列或更改列类型
• 部署后：如果数据管道部署了架构迁移
• 每周：作为良好实践，即使没有已知变更

缓存过期的迹象

留意以下迹象：

• 查询失败并出现“table not found”错误
• 结果似乎有误或已过时
• 未发现新表

手动重置缓存

如果怀疑缓存有问题：

# Scripts are relative to ../analyzing-data/
uv run scripts/cli.py cache status
uv run scripts/cli.py cache clear --stale-only
uv run scripts/cli.py cache clear

已识别的代码库模式

模式	来源	提取内容
**/models/**/*.yml	dbt	表和列的说明、测试
**/dags/**/*.sql	gusty	YAML frontmatter（description、primary_key）
AGENTS.md、CLAUDE.md	文档	数据层级结构、约定
**/docs/**/*.md	文档	业务背景

会话示例

User: /astronomer-data:warehouse-init

Agent:
→ Reading warehouse configuration...
→ Found 1 warehouse with databases: HQ, PRODUCT

→ Searching codebase for data documentation...
  Found: AGENTS.md with data layer hierarchy
  Found: 45 SQL files with YAML frontmatter in dags/declarative/

→ Launching parallel warehouse discovery...
  [Database: HQ] Discovering schemas...
  [Database: PRODUCT] Discovering schemas...

→ HQ: Found 29 schemas, 401 tables
→ PRODUCT: Found 1 schema, 0 tables

→ Merging warehouse metadata with code context...
  Enriched 45 tables with descriptions from code

→ Generated .astro/warehouse.md

Summary:
  - 2 databases
  - 30 schemas
  - 401 tables
  - 45 tables enriched with code descriptions
  - 8 large tables flagged (>100M rows)

Next steps:
  1. Review .astro/warehouse.md
  2. Add concept mappings to Quick Reference
  3. Commit to version control
  4. Run /astronomer-data:warehouse-init --refresh when schema changes