AI-трекинг Data Lineage: автоматизация и внедрение

Реализация AI-трекинга линейки данных (Data Lineage)

Представьте: вы изменяете структуру таблицы в staging слое, и через день дашборды отдела продаж показывают нули. Причина — пропущенный reference в SQL-трансформации. Без Data Lineage найти её — часы ручного перебора. Когда пайплайнов десятки, а дашбордов сотни, ручной impact analysis становится узким горлышком. Мы автоматизируем построение графа, который покажет каждый источник и трансформацию, и сделает impact analysis за секунды.

Мы интегрируем AI-трекинг происхождения данных, который автоматически строит граф из вашего SQL, dbt-моделей и Python-кода. Без этого найти источник ошибки в колонке — часы ручного поиска. Наше решение строит граф на лету, без ручной документации, и покрывает 80-90% типовых ETL-паттернов. Остальные 10% — runtime-трекинг через OpenLineage.

Автоматизация Data Lineage: почему это критично

Потеря линейки — одна из частых причин простоев в дата-платформах. Изменение одной колонки в raw-слое может сломать 20+ дашбордов в BI. Без автоматического трекинга impact analysis занимает дни, а ошибки обнаруживаются постфактум. AI-трекинг даёт вам:

• граф зависимостей — обновляется после каждого деплоя
• impact analysis — за секунды: "что сломается, если я удалю таблицу X"
• аудит трансформаций — откуда данные пришли и как изменились

Ручной impact analysis занимает 2-3 дня, AI-трекинг — менее 100 мс на графе из 1000 узлов. Разница — в тысячи раз. Экономия бюджета на поддержку данных достигает 40%, а время на аудит сокращается с 2 дней до 2 часов.

Что даёт impact analysis?

Impact analysis — это ответ на вопрос "что сломается, если я изменю эту таблицу". Без него каждое изменение — риск. AI-трекинг позволяет за секунды получить полный список затронутых дашбордов, моделей и пайплайнов. Например, удаление колонки client_id в staging слое может затронуть 15 дашбордов, 3 dbt-модели и 2 API-энпойнта. Impact analysis покажет это до деплоя.

Как мы строим граф линейки данных?

Парсинг SQL и dbt-моделей

Для извлечения линейки из SQL используем sqlglot для парсинга и LLM (Claude 3.5, GPT-4o) для сложных случаев — динамический SQL, оконные функции, ORM. Для dbt — читаем manifest.json, строим граф на основе ref-зависимостей.

from anthropic import Anthropic import sqlparse import sqlglot import networkx as nx import json from dataclasses import dataclass @dataclass class LineageNode: node_id: str name: str node_type: str # table, view, query, model, api, file schema: dict = None metadata: dict = None @dataclass class LineageEdge: source: str target: str transform_type: str # select, join, aggregation, filter, union columns_mapped: dict = None # {source_col: target_col} class DataLineageTracker: def __init__(self): self.llm = Anthropic() self.graph = nx.DiGraph() self.nodes = {} def parse_sql_lineage(self, sql: str, output_table: str = None) -> dict: """Извлечение линейки из SQL запроса""" try: # Парсинг через sqlglot statements = sqlglot.parse(sql) lineage = {'sources': [], 'targets': [], 'columns': {}} for stmt in statements: # Таблицы в FROM и JOIN for table in stmt.find_all(sqlglot.expressions.Table): if table.name: lineage['sources'].append(table.name) # Целевая таблица (CREATE TABLE AS / INSERT INTO) if isinstance(stmt, sqlglot.expressions.Create): lineage['targets'].append(str(stmt.this)) elif isinstance(stmt, sqlglot.expressions.Insert): lineage['targets'].append(str(stmt.this)) if output_table: lineage['targets'].append(output_table) # Маппинг колонок через LLM для сложных случаев lineage['column_mapping'] = self._extract_column_mapping(sql) return lineage except Exception: # Fallback: LLM-парсинг return self._llm_parse_lineage(sql, output_table) def _llm_parse_lineage(self, sql: str, output_table: str = None) -> dict: """LLM-извлечение линейки для сложного SQL""" response = self.llm.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=400, messages=[{ "role": "user", "content": f"""Extract data lineage from this SQL. SQL: {sql[:1500]} Output table: {output_table or "unknown"} Return JSON: {{ "sources": ["table1", "table2"], "targets": ["output_table"], "transforms": ["aggregation", "join"], "column_mapping": {{"source.col1": "target.col_a"}} }}""" }] ) try: return json.loads(response.content[0].text) except Exception: return {'sources': [], 'targets': [], 'transforms': [], 'column_mapping': {}} def _extract_column_mapping(self, sql: str) -> dict: """Маппинг колонок источник → цель""" try: parsed = sqlglot.parse_one(sql) mapping = {} for col in parsed.find_all(sqlglot.expressions.Column): alias = col.find_ancestor(sqlglot.expressions.Alias) if alias: target_name = str(alias.alias) source_name = str(col) mapping[source_name] = target_name return mapping except Exception: return {} def add_dbt_lineage(self, manifest_path: str): """Импорт линейки из dbt manifest.json""" with open(manifest_path) as f: manifest = json.load(f) for node_id, node in manifest.get('nodes', {}).items(): if node.get('resource_type') == 'model': model_name = node['name'] # Добавление узла модели self.graph.add_node(model_name, **{ 'type': 'dbt_model', 'schema': node.get('database', '') + '.' + node.get('schema', ''), 'description': node.get('description', ''), 'tags': node.get('tags', []) }) # Зависимости (upstream) for dep in node.get('depends_on', {}).get('nodes', []): dep_name = dep.split('.')[-1] self.graph.add_edge(dep_name, model_name, transform_type='dbt_ref') def build_lineage_from_code(self, python_code: str, file_name: str = "transform.py") -> dict: """Извлечение линейки из Python кода трансформации""" response = self.llm.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=400, messages=[{ "role": "user", "content": f"""Extract data lineage from this Python ETL code. File: {file_name} Code: {python_code[:1500]} Return JSON: {{ "reads_from": ["table/file/api names"], "writes_to": ["output table/file names"], "transforms": ["description of transformations applied"], "column_transforms": ["human-readable descriptions of column transformations"] }}""" }] ) try: return json.loads(response.content[0].text) except Exception: return {} 

Граф линейки и impact analysis

 def get_downstream_impact(self, source_table: str) -> dict: """Что сломается если изменить source_table""" if source_table not in self.graph: return {'affected': [], 'count': 0} # BFS для получения всех downstream узлов affected = [] visited = set() queue = [source_table] while queue: current = queue.pop(0) if current in visited: continue visited.add(current) successors = list(self.graph.successors(current)) for succ in successors: if succ != source_table: affected.append({ 'node': succ, 'distance': nx.shortest_path_length(self.graph, source_table, succ), 'type': self.graph.nodes[succ].get('type', 'unknown') }) queue.extend(successors) affected.sort(key=lambda x: x['distance']) return { 'source': source_table, 'affected': affected, 'count': len(affected) } def get_upstream_sources(self, target_table: str) -> dict: """Откуда данные в target_table""" if target_table not in self.graph: return {'sources': [], 'path': []} # Все предки ancestors = list(nx.ancestors(self.graph, target_table)) paths = {} for source in ancestors: try: path = nx.shortest_path(self.graph, source, target_table) paths[source] = path except nx.NetworkXNoPath: pass # AI-объяснение линейки explanation = self._explain_lineage(target_table, ancestors, paths) return { 'target': target_table, 'sources': ancestors, 'paths': paths, 'explanation': explanation } def _explain_lineage(self, target: str, sources: list, paths: dict) -> str: """LLM-объяснение линейки данных""" paths_summary = json.dumps( {s: p for s, p in list(paths.items())[:5]}, ensure_ascii=False ) response = self.llm.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=300, messages=[{ "role": "user", "content": f"""Explain the data lineage for table "{target}". Source tables: {sources} Key paths: {paths_summary} Summarize: where data originates, what transformations occur, potential data quality risks. 2-4 sentences, non-technical language.""" }] ) return response.content[0].text def detect_lineage_breaks(self) -> list[dict]: """Обнаружение разрывов в линейке""" breaks = [] # Таблицы без источников (кроме raw) for node in self.graph.nodes(): if self.graph.in_degree(node) == 0: node_data = self.graph.nodes[node] if node_data.get('type') not in ['raw_table', 'external_source']: breaks.append({ 'node': node, 'issue': 'no_upstream_lineage', 'severity': 'warning' }) # Таблицы без потребителей if self.graph.out_degree(node) == 0: breaks.append({ 'node': node, 'issue': 'orphaned_dataset', 'severity': 'info' }) return breaks 

Трекинг линейки через парсинг SQL + дополнение LLM покрывает 80-90% типовых ETL-паттернов. Для сложных tricky трансформаций (динамический SQL, ORM) нужен runtime tracing. OpenLineage + Marquez — стандарт для автоматического сбора линейки из Airflow, Spark и dbt без написания кода. Мы интегрируем OpenLineage в ваш существующий пайплайн за 1-2 дня, добавляя прозрачность без изменения кода.

Пошаговая инструкция внедрения AI-трекинга

1. Аудит — собираем SQL, dbt, Python-код всех пайплайнов.
2. Парсинг — sqlglot + LLM извлекают зависимости и маппинг колонок.
3. Построение графа — NetworkX создаёт DAG с узлами и рёбрами.
4. Impact analysis — реализуем API для запросов downstream/upstream.
5. Интеграция в CI/CD — автоматическое обновление графа при каждом деплое.
6. Тестирование — проверка покрытия и точности на реальных данных.

Что входит в результат работы

• Автоматически обновляемый граф линейки данных (дашборд на основе NetworkX + D3.js).
• API для impact analysis с поддержкой upstream/downstream запросов.
• Детекция разрывов в линейке (таблицы без источников или потребителей).
• Интеграция с CI/CD (GitHub Actions / GitLab CI) для автообновления.
• Документация архитектуры и инструкции для команды.
• Обучение команды (2 часа воркшоп).
• Поддержка 3 месяца после внедрения.

Этапы реализации AI-трекинга

Сравнение методов: ручной vs AI-трекинг

AI-трекинг в 50 раз быстрее ручного impact analysis, а экономия на поддержке данных превышает 40%. Оценим ваш проект — пишите. Наш опыт: 5+ лет в MLOps, более 20 реализованных проектов по Data Lineage. Сертифицированы по OpenLineage. Получите консультацию — свяжитесь с нами.