DALI · OpenLineage

Интеграция с OpenLineage

DALI принимает события OpenLineage от runtime-инструментов (dbt, Spark, Airflow, Flink) и сливает их со статическим графом HOUND в единую YGG-базу. Результат: один граф, в котором видно и что описано в SQL, и что реально выполнялось в продакшне.

Двойной источник: HOUND + OpenLineage

HOUND (статика)

·SQL-файлы из репозитория

·Полная колоночная линяжность

·DDL: таблицы, схемы, типы

·Покрытие: 6 диалектов

OpenLineage (рантайм)

·dbt / Spark / Airflow / Flink

·Реальные данные выполнения

·Статистика: строки, размер, время

·Иерархия: DAG → Task → Run

YGG (слияние)

·Узлы: static_observed + runtime_observed

·Рёбра DATA_FLOW из обоих источников

·Одна база, единый граф

·LOOM: твёрдые vs пунктирные

При слиянии каждый узел получает два флага: static_observed (HOUND разобрал DDL) и runtime_observed (OL-событие зафиксировало выполнение). Если оба true — узел «известен в полной мере».

Пайплайн обработки события

1Ingestion

Файловый watcher (30с) или HTTP POST /api/v1/lineage принимает JSON-событие. Сохраняет raw JSON в FRIGG для аудита (TTL 30 дней).

2Нормализация

OlIdentifierNormalizer приводит имена к каноническому виду с учётом диалекта: Oracle → UPPERCASE, PostgreSQL → lowercase, ClickHouse → preserve.

3Дедупликация

Повторный runId автоматически пропускается. Запросы, отличающиеся только значениями констант, считаются одним шаблоном. Дублирующиеся рёбра в графе устраняются.

4SQL-допарсинг

HoundJobParser извлекает SQL из facets события и отправляет его в HOUND для полноценного ANTLR4-разбора. Строит DaliStatement-узлы как из статических файлов.

5Запись в YGG

OlToDaliMapper конвертирует OL Dataset → DaliTable/DaliColumn. columnLineage.fields → DATA_FLOW рёбра. inputs[] → READS_FROM, outputs[] → WRITES_TO.

Дедупликация

Дедупликация по тексту

Сигнатура запроса, инвариантная к форматированию. Один и тот же SQL с разными отступами или регистром считается дублём и не создаёт лишних узлов в графе.

SQL-отпечаток

Дедупликация по шаблону

Запросы, отличающиеся только значениями констант в условиях, считаются одним шаблоном. Тысячи параметризованных запусков одного SELECT схлопываются в единый узел графа.

SQL-шаблон

Diamond dedup

При обходе графа один и тот же путь A→B может возникнуть по нескольким маршрутам. Дублирующиеся рёбра автоматически устраняются перед отправкой в LOOM.

Граф-дедупликация

Допарсинг SQL внутри OL-событий

OpenLineage-события от dbt и Spark часто содержат SQL-текст в SQLJobFacet. DALI извлекает этот SQL и передаёт его в HOUND для полноценного ANTLR4-разбора — точно так же, как если бы это был статический файл.

// OL-событие от dbt содержит SQL в facets:
{
  "job": { "facets": {
    "sql": { "query": "SELECT o.id, c.tier, o.amount * c.discount AS discounted_amount\n       FROM stg_orders o JOIN stg_customers c ON o.customer_id = c.id" }
  }},
  "outputs": [{ "name": "orders_enriched", "facets": {
    "columnLineage": {
      "fields": {
        "discounted_amount": {
          "inputFields": [
            { "name": "stg_orders",    "field": "amount" },
            { "name": "stg_customers", "field": "discount" }
          ]
        }
      }
    }
  }}]
}

// DALI: HoundJobParser.extractSql() → отправить в HOUND → DaliStatement-узел
// Результат: граф знает не только "discounted_amount ← amount + discount",
// но и полный AST, все промежуточные операции и связи с DDL-определениями таблиц.

Поддерживаемые источники (producers)

Инструмент	Транспорт	Колоночная линяжность	Особенности
dbt	файл / HTTP	полная	SQLJobFacet → допарсинг HOUND; тест-стенд на DuckDB; dbt v1.9.0+
Spark	файл / HTTP	полная	openlineage-spark_2.12 v1.33.0; DataFrame-операции → DATA_FLOW рёбра
Airflow	файл / HTTP	через задачи	DAGRun → TaskInstance иерархия через parentRunId; NominalTimeFacet для расписания
Flink	файл	базовая	Dataset-уровень без column lineage; roadmap: v1.6+
Trino	HTTP (план)	через SQL	SQL из QueryCompletedEvent → допарсинг HOUND Trino-диалектом

Разметка в LOOM

После слияния каждый узел и ребро в YGG получают метку источника. LOOM визуализирует их по-разному.

origin: "static"

Только HOUND

Сплошная граница, 100% прозрачность. SQL разобран, в продакшне не наблюдался.

Сплошное ребро

origin: "runtime"

Только OL

Пунктирная граница, 70% прозрачность. Реальное выполнение есть, DDL ещё не разобран.

Пунктирное/синее ребро

origin: "both"

HOUND + OL

Сплошная граница + ★ бейдж. Полная картина: и структура, и факт выполнения.

Сплошное + синее ребро

Транспорты

Файловый watcher

DALI сканирует директорию dali.ol.events-dir каждые 30 секунд. Обработанные → processed/, упавшие → failed/. Удобно для dbt/Spark с file transport.

Phase 1 · production-ready

HTTP endpoint

POST /api/v1/lineage — стандартный OL-совместимый эндпоинт. Марquez-совместимый URL. Producer пушит напрямую без промежуточного файла.

Phase 3 · production-ready

Kafka (roadmap)

Планируется при объёме >1К событий в день. Текущая архитектура рассчитана на файл/HTTP — достаточно для большинства on-premise инсталляций.

roadmap · v1.6+

Покрытие SQL-операций по диалектам

43 операции × 6 диалектов — что именно HOUND извлекает из каждого SQL

Открыть матрицу