Как подключить ИИ к управлению метаданными: практический опыт внедрения DataHub + MCP

Проблема: почему метаданные остаются "мертвым грузом" в вашей организации?

Представьте типичную ситуацию в компании среднего размера: у вас есть DataHub с тысячами таблиц, сотнями пайплайнов и десятками дашбордов. Технически метаданные собраны, но на практике они используются только для ручного поиска. Новый аналитик тратит 2 часа на поиск нужной таблицы, data scientist не понимает, какие фичи уже существуют, а инженеры создают дублирующие пайплайны.

Традиционные дата-каталоги решают проблему хранения, но не проблему использования. Вот где на помощь приходит генеративный ИИ и Model Context Protocol (MCP) — протокол, который позволяет LLM взаимодействовать с внешними системами, включая каталоги метаданных.

Решение: DataHub + MCP = интеллектуальный ассистент для данных

Комбинация DataHub (как источник структурированных метаданных) и MCP (как мост к LLM) создает систему, где:

• Аналитик может спросить на естественном языке: "Какие таблицы содержат данные о продажах за последний квартал?"
• Data scientist получает рекомендации: "Для прогноза оттока клиентов используй эти 5 фичей из таблицы X"
• Инженер видит impact analysis: "Если я изменю эту колонку, какие дашборды сломаются?"

Архитектура решения: как все устроено под капотом

Прежде чем переходить к реализации, важно понять архитектурные компоненты:

Ключевое преимущество этой архитектуры — разделение ответственности. DataHub остается источником истины для метаданных, MCP Server обеспечивает безопасный доступ, а LLM Client предоставляет естественный интерфейс.

1 Подготовка DataHub: настройка и проверка доступа

Перед интеграцией с MCP убедитесь, что ваш DataHub правильно настроен:

# Проверяем доступность DataHub API
curl -X GET https://your-datahub.com/api/gms/health

# Получаем токен для аутентификации (если используется)
curl -X POST https://your-datahub.com/api/v2/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "mutation { createAccessToken(input: {type: PERSONAL, actorUrn: \"urn:li:corpuser:admin\", name: \"mcp-token\", duration: \"ONE_YEAR\"}) { accessToken } }"}'

2 Создание MCP Server для DataHub

Создаем Python-приложение, которое будет выступать в роли MCP Server:

# mcp_datahub_server.py
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
from typing import Any
import httpx
from dataclasses import dataclass
import json

@dataclass
class DataHubConfig:
    base_url: str
    token: str
    timeout: int = 30

class DataHubMCPServer:
    def __init__(self, config: DataHubConfig):
        self.config = config
        self.client = httpx.AsyncClient(
            base_url=config.base_url,
            headers={"Authorization": f"Bearer {config.token}"},
            timeout=config.timeout
        )
        self.server = Server("datahub-mcp-server")
        
        # Регистрируем инструменты (tools)
        self.server.list_tools()(self.list_tools)
        self.server.call_tool()(self.call_tool)
        
    async def search_datasets(self, query: str, limit: int = 10) -> list[dict]:
        """Поиск датасетов в DataHub"""
        search_query = {
            "query": query,
            "count": limit,
            "filters": {"platform": ["bigquery", "snowflake", "postgres"]}
        }
        
        response = await self.client.post(
            "/api/v2/search",
            json=search_query
        )
        response.raise_for_status()
        return response.json().get("data", {}).get("searchAcrossEntities", {}).get("searchResults", [])
    
    async def get_dataset_lineage(self, dataset_urn: str) -> dict:
        """Получение lineage для датасета"""
        graphql_query = """
        query GetLineage($urn: String!) {
          dataset(urn: $urn) {
            downstreamLineage {
              entities {
                urn
                type
                ... on Dataset {
                  name
                  platform {
                    name
                  }
                }
              }
            }
          }
        }
        """
        
        response = await self.client.post(
            "/api/v2/graphql",
            json={
                "query": graphql_query,
                "variables": {"urn": dataset_urn}
            }
        )
        response.raise_for_status()
        return response.json()
    
    async def list_tools(self) -> list[dict]:
        """Список доступных инструментов"""
        return [
            {
                "name": "search_datasets",
                "description": "Поиск датасетов в DataHub по названию, описанию или тегам",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Поисковый запрос"},
                        "limit": {"type": "integer", "description": "Максимальное количество результатов", "default": 10}
                    },
                    "required": ["query"]
                }
            },
            {
                "name": "get_dataset_lineage",
                "description": "Получение lineage (зависимостей) для датасета",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "dataset_urn": {"type": "string", "description": "URN датасета (например, urn:li:dataset:(urn:li:dataPlatform:bigquery,project.dataset.table))"}
                    },
                    "required": ["dataset_urn"]
                }
            }
        ]
    
    async def call_tool(self, name: str, arguments: dict) -> dict:
        """Вызов конкретного инструмента"""
        if name == "search_datasets":
            results = await self.search_datasets(
                arguments["query"],
                arguments.get("limit", 10)
            )
            return {
                "content": [{
                    "type": "text",
                    "text": json.dumps(results, indent=2, ensure_ascii=False)
                }]
            }
        elif name == "get_dataset_lineage":
            lineage = await self.get_dataset_lineage(arguments["dataset_urn"])
            return {
                "content": [{
                    "type": "text",
                    "text": json.dumps(lineage, indent=2, ensure_ascii=False)
                }]
            }
        else:
            raise ValueError(f"Unknown tool: {name}")
    
    async def run(self):
        """Запуск MCP сервера"""
        async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
            await self.server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name="datahub-mcp-server",
                    server_version="0.1.0",
                    capabilities=self.server.get_capabilities(
                        notification_options=NotificationOptions(),
                        experimental_capabilities={}
                    )
                )
            )

if __name__ == "__main__":
    config = DataHubConfig(
        base_url="https://your-datahub.com",
        token="your-token-here"
    )
    server = DataHubMCPServer(config)
    asyncio.run(server.run())

3 Настройка клиента (Claude Desktop, Cursor, VSCode)

Конфигурация для Claude Desktop (config.json):

{
  "mcpServers": {
    "datahub": {
      "command": "python",
      "args": [
        "/path/to/mcp_datahub_server.py"
      ],
      "env": {
        "DATAHUB_URL": "https://your-datahub.com",
        "DATAHUB_TOKEN": "your-token-here"
      }
    }
  }
}

Для VSCode с расширением Continue:

// .continue/config.json
{
  "models": [
    {
      "title": "Claude 3.5 Sonnet",
      "provider": "openai",
      "model": "claude-3-5-sonnet",
      "apiBase": "https://api.anthropic.com",
      "apiKey": "your-anthropic-key"
    }
  ],
  "tabAutocompleteModel": {
    "title": "Starcoder",
    "provider": "openai",
    "model": "starcoder"
  },
  "experimental": {
    "mcpServers": {
      "datahub": {
        "command": "uv",
        "args": ["run", "mcp-datahub-server"],
        "cwd": "/path/to/your/mcp/server"
      }
    }
  }
}

4 Расширенные возможности: семантический поиск и RAG

Для улучшения поиска можно добавить векторное представление метаданных. Это особенно полезно, когда пользователи ищут данные на естественном языке, а не по точным названиям таблиц.

# Дополнение к MCP Server для семантического поиска
from sentence_transformers import SentenceTransformer
import numpy as np
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

class SemanticSearchExtension:
    def __init__(self, qdrant_url: str = "localhost:6333"):
        self.model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
        self.qdrant = QdrantClient(qdrant_url)
        
        # Создаем коллекцию если не существует
        try:
            self.qdrant.get_collection("datahub_metadata")
        except:
            self.qdrant.create_collection(
                collection_name="datahub_metadata",
                vectors_config=VectorParams(
                    size=384,  # Размерность all-MiniLM-L6-v2
                    distance=Distance.COSINE
                )
            )
    
    async def semantic_search(self, query: str, limit: int = 5) -> list[dict]:
        """Семантический поиск по метаданным"""
        query_embedding = self.model.encode(query).tolist()
        
        search_result = self.qdrant.search(
            collection_name="datahub_metadata",
            query_vector=query_embedding,
            limit=limit
        )
        
        return [
            {
                "score": hit.score,
                "metadata": hit.payload
            }
            for hit in search_result
        ]

Этот подход превращает ваш MCP Server в полноценную RAG-систему для метаданных. Если вы хотите глубже погрузиться в тему RAG, рекомендую нашу статью "RAG за 15 минут: создаем свою систему на Python с нуля".

Реальные use cases из нашего опыта внедрения

1. Ускорение onboarding новых сотрудников

Новый data scientist приходит в компанию. Вместо недели изучения документации, он задает вопросы в Claude Desktop:

• "Какие данные у нас есть для прогнозирования оттока клиентов?"
• "Покажи примеры feature engineering для нашей продуктовой аналитики"
• "Кто отвечает за данные о транзакциях?"

2. Impact analysis для миграций

При миграции с Redshift на BigQuery инженеры используют MCP для анализа:

-- Вместо ручного анализа lineage через UI
-- Инженер спрашивает у ИИ:
"Какие дашборды и пайплайны зависят от таблицы
prod_analytics.user_sessions?"

3. Автоматическая документация

MCP Server можно расширить для генерации документации на основе метаданных:

# Пример инструмента для генерации документации
async def generate_dataset_docs(self, dataset_urn: str) -> str:
    """Генерация документации для датасета"""
    metadata = await self.get_dataset_metadata(dataset_urn)
    
    prompt = f"""
    На основе следующих метаданных создай подробную документацию:
    Название: {metadata['name']}
    Описание: {metadata.get('description', 'Нет описания')}
    Колонки: {json.dumps(metadata.get('columns', []), ensure_ascii=False)}
    Владелец: {metadata.get('owners', [])}
    
    Включи:
    1. Назначение таблицы
    2. Ключевые метрики
    3. Примеры использования
    4. Ограничения и предостережения
    """
    
    # Используем LLM для генерации
    return await self.llm_client.generate(prompt)

Типичные ошибки и как их избежать

Roadmap для production-внедрения

1. Месяц 1: Пилот с 5-10 power users, сбор feedback
2. Месяц 2: Добавление аутентификации и аудита запросов
3. Месяц 3: Интеграция с корпоративными чат-ботами (Slack, Teams)
4. Месяц 4: Автоматическое обогащение метаданных через ИИ (например, генерация описаний для колонок)
5. Месяц 6: Predictive features: "Какие данные тебе понадобятся для этого анализа?"

Заключение: от каталога к интеллектуальному ассистенту

Интеграция DataHub с MCP — это не просто технический эксперимент, а стратегическое улучшение data infrastructure. Вы превращаете пассивный каталог метаданных в активного помощника, который:

• Сокращает time-to-insight для аналитиков
• Уменьшает operational risks при изменениях
• Повышает discoverability данных на 60-80% по нашим измерениям
• Созет foundation для более сложных AI-агентов в будущем

Если вы только начинаете путь с AI-агентами, рекомендую изучить наш гайд "Production-ready AI-агенты: как превратить хайп в работающую систему для бизнеса". А для тех, кто хочет глубже погрузиться в ML-инфраструктуру, полезной будет статья "Как интегрировать свои ML/DL модели в продакшн-приложения".

Начните с простого MCP Server для 2-3 самых востребованных use cases, соберите feedback от пользователей, и постепенно расширяйте функциональность. Уже через месяц вы увидите, как меняется культура работы с данными в вашей организации.