de-agent-skills: готовые PySpark и Spark SQL skills для AI-агента

de-agent-skills/
├── skills/
│   ├── pyspark_etl/SKILL.md    ← skill для агента: DataFrame API
│   └── spark_sql/SKILL.md      ← skill для агента: Spark SQL
├── docs/specs/
│   ├── pyspark_enterprise.md   ← enterprise-стандарты PySpark (TB-PB)
│   ├── spark_sql_enterprise.md ← enterprise-стандарты Spark SQL
│   ├── hdfs_hive_parquet_datalake.md
│   └── spark_sql_hdfs_hive_operations.md
└── guides/
    ├── spark_sql_hdfs_hive_tutorial_101.md
    └── hdfs_hive_partitioning_spark_tutorial_101.md

# Стандартные алиасы (агент всегда использует эти имена)
from pyspark.sql import functions as F
from pyspark.sql import types as T
from pyspark.sql.window import Window as W

# Explicit schema - не inferSchema
schema = T.StructType([
    T.StructField("order_id",   T.LongType(),      False),
    T.StructField("status",     T.StringType(),    True),
    T.StructField("amount",     T.DecimalType(18, 2), True),
    T.StructField("event_date", T.DateType(),      False),
])

df = spark.read.schema(schema).parquet("s3a://bucket/orders/")

# select() как явный contract схемы - не df.withColumn цепочки
result = (
    df
    .filter(F.col("event_date") >= "2024-01-01")   # partition pruning первым
    .select("order_id", "status", "amount")         # column pruning сразу
    .withColumn("amount_rub", F.col("amount") * 90)
)

# ❌ collect() на больших данных
data = df.collect()  # OutOfMemoryError при > нескольких GB

# ❌ Python loop вместо vectorized операций
for row in df.toLocalIterator():
    process(row)

# ❌ Цепочка withColumn (создаёт промежуточные планы)
df = df.withColumn("a", ...)
df = df.withColumn("b", ...)  # вместо одного select()

# ❌ inferSchema на production данных
df = spark.read.json(path)  # схема угадывается, нестабильно

-- Агент всегда структурирует запросы через CTEs
WITH
source_filtered AS (
    -- Ранняя фильтрация с partition pruning
    SELECT
        order_id,
        user_id,
        status,
        amount,
        event_date
    FROM nessie.staging.orders
    WHERE
        event_date >= '2024-01-01'          -- partition pruning
        AND event_date <  '2024-02-01'      -- bounded range (обязательно)
        AND status IS NOT NULL              -- явная null-фильтрация
),

deduped AS (
    -- Дедупликация перед join - не после
    SELECT *
    FROM (
        SELECT *,
               ROW_NUMBER() OVER (
                   PARTITION BY order_id
                   ORDER BY updated_at DESC  -- детерминированный порядок
               ) AS rn
        FROM source_filtered
    )
    WHERE rn = 1
),

enriched AS (
    -- JOIN только с дедуплицированными данными
    SELECT
        o.order_id,
        o.status,
        o.amount,
        u.region
    FROM deduped o
    LEFT JOIN nessie.dw.dim_users u
        ON o.user_id = u.user_id
        AND u.is_current = true             -- фильтр на dimension
),

final AS (
    SELECT
        region,
        COUNT(*)                            AS order_count,
        SUM(amount)                         AS total_amount,
        AVG(amount)                         AS avg_amount
    FROM enriched
    GROUP BY region
)

SELECT * FROM final
ORDER BY total_amount DESC

-- ❌ SELECT * в финальном выводе
SELECT * FROM orders JOIN users ON ...

-- ❌ DISTINCT после JOIN (сигнал неверного grain)
SELECT DISTINCT order_id FROM orders JOIN events ON ...

-- ❌ ORDER BY без LIMIT на большой таблице
SELECT * FROM orders ORDER BY amount DESC

-- ❌ COUNT(DISTINCT high_cardinality_col) - запускает shuffle
SELECT COUNT(DISTINCT user_id) FROM billion_row_table

-- ✓ Альтернатива: approx_count_distinct
SELECT approx_count_distinct(user_id) FROM billion_row_table

# Клонируем рядом с проектом
git clone https://github.com/ivanshamaev/de-agent-skills ~/.agent-skills

# Добавляем в CLAUDE.md своего проекта
cat >> CLAUDE.md << 'EOF'

## Agent Skills Reference

For PySpark code: follow ~/.agent-skills/skills/pyspark_etl/SKILL.md
For Spark SQL: follow ~/.agent-skills/skills/spark_sql/SKILL.md
Production standards: ~/.agent-skills/docs/specs/
EOF

cp -r ~/.agent-skills/docs/specs/ ./docs/spark-standards/
# Агент автоматически читает файлы в docs/ при анализе репозитория

# .opencode/skills/pyspark-review.yaml
name: pyspark-review
description: Проверяет PySpark код по production-стандартам
trigger: "review|проверь|code review"
steps:
  - name: load_standards
    tool: read
    path: "docs/spark-standards/pyspark_enterprise.md"
  - name: review
    llm: true
    prompt: |
      Проверь код по стандартам:
      {load_standards.output}

      Код для проверки:
      {selection}

      Проверь: explicit schema, partition pruning, join типы,
      anti-patterns (collect, loops, inferSchema), порядок оптимизации.