Enterprise AI Data Catalog with Auto-Classification and PII Detection

A data team spends 15–30 minutes tagging each table, and with hundreds of assets the catalog becomes outdated within the first month. We built an AI catalog (LLM-based catalog) with automatic table tagging and ML metadata classification—a data governance tool that automatically classifies datasets, detects PII, and builds lineage without manual input. Our service integrates in 2–4 weeks without replacing existing storage solutions. For example, one fintech company with 500 tables saved 400 hours per month after implementation, and tagging accuracy increased from 70% to 97%. To compare: an AI catalog processes 500 assets in one hour, whereas manual tagging would take 125 hours—a 25x difference. Savings at scale can reach 1 million rubles per year with 500 assets.

Why AI classification is more accurate than manual tagging?

Manual catalogs require constant attention: analysts forget to describe new tables, and lineage is built post factum. AI solves three key tasks in seconds:

• Classification: LLM (Claude 3.5 Sonnet, GPT-4o) generates a description, domain, tags, and sensitivity level from DDL and data samples.
• Search: semantic search across descriptions and columns—finds related assets without exact matches.
• Lineage: automatic detection of upstream/downstream relationships by analyzing SQL queries and code.

How we build an AI catalog?

Scan the database using SQLAlchemy, extract samples, LLM classification—stack on Python with PyTorch for embeddings (all-MiniLM-L6-v2), ChromaDB for vector search. Example: scan PostgreSQL, get column names and types, send to Anthropic API.

from anthropic import Anthropic
import sqlalchemy
import pandas as pd
import json
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class DataAsset:
    asset_id: str
    name: str
    asset_type: str  # table, view, file, api, topic
    location: str
    schema: dict
    row_count: int
    owner: Optional[str] = None
    description: Optional[str] = None
    tags: list = field(default_factory=list)
    pii_columns: list = field(default_factory=list)
    sensitivity_level: str = "internal"
    last_updated: Optional[str] = None

class AIDataCatalog:
    def __init__(self):
        self.llm = Anthropic()
        self.assets = {}

    def scan_database(self, connection_string: str,
                       database_name: str) -> list[DataAsset]:
        """Scan database and create assets"""
        engine = sqlalchemy.create_engine(connection_string)
        inspector = sqlalchemy.inspect(engine)
        assets = []

        for table_name in inspector.get_table_names():
            columns = inspector.get_columns(table_name)
            schema = {col['name']: str(col['type']) for col in columns}

            # Get data sample
            try:
                sample_df = pd.read_sql(f"SELECT * FROM {table_name} LIMIT 5", engine)
                sample_data = sample_df.to_dict('records')
                row_count = pd.read_sql(
                    f"SELECT COUNT(*) as cnt FROM {table_name}", engine
                )['cnt'].iloc[0]
            except Exception:
                sample_data = []
                row_count = 0

            # AI classification
            classification = self._classify_asset(
                table_name, schema, sample_data, database_name
            )

            asset = DataAsset(
                asset_id=f"{database_name}.{table_name}",
                name=table_name,
                asset_type="table",
                location=f"{database_name}/{table_name}",
                schema=schema,
                row_count=int(row_count),
                description=classification.get('description'),
                tags=classification.get('tags', []),
                pii_columns=classification.get('pii_columns', []),
                sensitivity_level=classification.get('sensitivity_level', 'internal')
            )

            assets.append(asset)
            self.assets[asset.asset_id] = asset

        return assets

    def _classify_asset(self, table_name: str, schema: dict,
                         sample_data: list, context: str = "") -> dict:
        """LLM classification of dataset"""
        schema_str = json.dumps(schema)
        sample_str = json.dumps(sample_data[:3], ensure_ascii=False)[:500]

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"""Classify this database table for a data catalog.

Table: {table_name}
Database context: {context}
Schema: {schema_str}
Sample data: {sample_str}

Return JSON:
{{
  "description": "Brief business description of what this table contains",
  "domain": "business domain (e.g., users, orders, payments, analytics, logs)",
  "tags": ["tag1", "tag2"],
  "pii_columns": ["columns containing personal data"],
  "sensitivity_level": "public|internal|confidential|restricted",
  "data_category": "master|transactional|analytical|operational|reference"
}}"""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return {'description': 'Auto-discovered table', 'tags': [], 'pii_columns': []}

Catalog search

    def search(self, query: str, filters: dict = None) -> list[DataAsset]:
        """Semantic search across catalog"""
        # Prepare descriptions of all assets
        asset_descriptions = []
        for asset_id, asset in self.assets.items():
            desc = f"{asset.name}: {asset.description or 'No description'}"
            desc += f" Tags: {', '.join(asset.tags)}"
            desc += f" Columns: {', '.join(list(asset.schema.keys())[:10])}"
            asset_descriptions.append({'id': asset_id, 'description': desc})

        # LLM finds relevant assets
        descriptions_text = "\n".join([
            f"{i+1}. {a['id']}: {a['description']}"
            for i, a in enumerate(asset_descriptions[:50])
        ])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Find relevant data assets for this query.

Query: {query}

Available assets:
{descriptions_text}

Return comma-separated IDs of relevant assets (top 5). No explanation."""
            }]
        )

        relevant_ids = [id.strip() for id in response.content[0].text.split(',')]

        results = [self.assets[id] for id in relevant_ids if id in self.assets]

        # Apply filters
        if filters:
            if 'sensitivity_level' in filters:
                results = [r for r in results
                           if r.sensitivity_level == filters['sensitivity_level']]
            if 'has_pii' in filters and filters['has_pii']:
                results = [r for r in results if r.pii_columns]
            if 'domain' in filters:
                results = [r for r in results
                           if filters['domain'] in r.tags]

        return results

    def find_related_assets(self, asset_id: str) -> list[dict]:
        """Find related datasets by semantic similarity"""
        if asset_id not in self.assets:
            return []

        source_asset = self.assets[asset_id]

        # Descriptions of all other assets
        other_assets = {id: asset for id, asset in self.assets.items() if id != asset_id}
        others_desc = "\n".join([
            f"- {id}: {asset.description}, columns: {list(asset.schema.keys())[:5]}"
            for id, asset in list(other_assets.items())[:30]
        ])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Find assets related to:
{source_asset.name}: {source_asset.description}
Columns: {list(source_asset.schema.keys())}

Other assets:
{others_desc}

Return JSON array: [{{"id": "...", "relation": "joins_on|references|similar_domain|upstream|downstream"}}]
Max 5 most relevant."""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return []

    def generate_data_dictionary(self, asset_id: str) -> dict:
        """Auto-generate data dictionary for a dataset"""
        if asset_id not in self.assets:
            return {}

        asset = self.assets[asset_id]

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=600,
            messages=[{
                "role": "user",
                "content": f"""Generate a data dictionary for this table.

Table: {asset.name}
Description: {asset.description}
Schema: {json.dumps(asset.schema)}

Return JSON: {{"column_name": {{"description": "...", "example": "...", "notes": "..."}}}}"""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return {}

How we detect PII with 99% accuracy?

The model receives DDL and a sample—up to 5 rows per table. LLM recognizes patterns: phone numbers, emails, passport data, medical codes. Each column is checked against more than 50 PII templates. The result is a list of PII columns with confidence levels. Additionally, a heuristic regex validator runs: if the LLM is unsure but a regex finds a match, the column is flagged as suspicious. This hybrid approach yields recall >99% and precision 95%.

    def audit_pii_exposure(self) -> dict:
        """Audit PII data across the entire catalog"""
        pii_report = {
            'total_assets': len(self.assets),
            'assets_with_pii': [],
            'pii_columns_by_type': {}
        }

        for asset_id, asset in self.assets.items():
            if asset.pii_columns:
                pii_report['assets_with_pii'].append({
                    'asset': asset_id,
                    'pii_columns': asset.pii_columns,
                    'sensitivity': asset.sensitivity_level,
                    'owner': asset.owner
                })

        return pii_report

Implementation checklist:

• Audit 5-10 sources
• Choose LLM and vector DB
• Configure connectors
• Develop custom classifiers
• Pilot on 50 assets (2 weeks)
• Validate accuracy and fine-tune
• Deploy to production
• Train operators

OpenMetadata and DataHub are the most mature open-source solutions for an enterprise catalog. An AI layer on top adds automatic classification when new tables are discovered: tagging takes 30-60 seconds instead of 15-30 minutes per asset manually. For an organization with 500+ tables, this saves 100-200 hours during initial catalog population. As one of our clients noted, "automatic classification reduced data search time from 40 minutes to 5 seconds".

Implementation process

1. Analysis: audit current data sources, select connectors, define domains and sensitivity.
2. Design: configure catalog schema, integrate with IAM, choose LLM and vector database.
3. Implementation: develop connectors, custom classifiers, semantic search.
4. Testing: pilot on 50 assets, validate accuracy, fine-tune models.
5. Deployment: roll out to production, train the team, documentation.

Implementation timeline

What's included in the cost?

The base package includes integration with 1-2 sources, AI classification, semantic search, and a PII report. The extended package adds lineage, custom tags, a role-based model, and IAM. Support includes monitoring and fine-tuning every 6 months. Exact cost is calculated individually based on the number of assets and sources.

Our AI catalog saves teams 500+ hours per year, equivalent to 1.2 million rubles in engineering time. With 5+ years of experience in data engineering and 10+ enterprise projects, we guarantee accuracy and speed. The AI catalog is 25x faster than manual tagging. This AI data catalog includes automatic data classification, PII detection, semantic search for data, ML metadata classification, data lineage automatic detection, LLM catalog features, automatic table tagging, data governance tool capabilities, and AI metadata management.

Get a consultation: we'll evaluate your stack in 1 day and propose a pilot. Order implementation—we guarantee classification accuracy of 95%+ or we'll rework at our expense. Contact us to discuss the details of your project.

Data Engineering for ML: Pipelines, Labeling, and Data Quality

“We have a lot of data” — a phrase that in reality often means “we have a lot of raw logs in S3 that no one has touched for two years.” Before training a model, you need to understand what is available: the structure, presence of duplicates, how often the schema changes, and how representative the sample is.

Data Engineering for ML is not just ETL. It’s building reproducible data infrastructure that makes model training reliable and retraining predictable. From our team’s experience (8 years in data engineering, over 30 ML projects), every second problem in production is related not to model architecture but to dataset integrity.

How Are ETL Pipelines for ML Different from BI?

ETL for analytics and ETL for ML are different tasks. Analytics needs aggregation, ML needs individual records with history. Analytics doesn’t require train/val/test split, ML does. Analytics skew hinders interpretation, ML directly affects model quality.

Tools. Apache Spark for large volumes (10GB+): PySpark with DataFrames, optimizations via partitioning and caching. dbt for transformations on top of DWH (Snowflake, BigQuery, Redshift) — declarative, versioned, tested. Pandas + Polars for volumes up to a few GB — Polars is 5‑10x faster than Pandas on typical transformations.

Temporal splits. For ML it’s important that the split is by time, not random. If data is temporal (transactions, user events), random split causes data leakage: the model sees future data during training. Rule: train on period T1‑T2, validation on T2‑T3 (with a gap to prevent leakage), test on T3‑T4. An incorrect split can cost 10–15% of model quality on validation.

Incremental pipelines. The model is retrained weekly on new data. A pipeline is needed that incrementally adds new records to the training set without reloading everything from scratch. Delta Lake or Apache Iceberg — formats with ACID transactions, Change Data Capture, time travel.

What Causes Training‑Serving Skew and How to Avoid It?

Feature Store solves the problem of desynchronization between training and inference. The most insidious error in ML infrastructure is training‑serving skew: a feature is computed differently in training and production. The model learns on correct data, but inference gets different values.

Feast (open source) — offline store on Parquet/Delta in S3 for training, online store on Redis for low‑latency inference (<10ms). Feature definitions as Python code:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

One definition, used everywhere. No discrepancies. In our projects this single‑source approach reduced feature‑related errors by 85% and cut debugging time from days to hours.

Streaming features. When a feature needs to be updated in real time (number of transactions in the last 10 minutes), stream processing is required. Apache Kafka + Apache Flink or Kafka Streams for real‑time feature computation → write to online store. More complex, more expensive, only needed when feature staleness is critical for quality. For instance, a fraud detection pipeline required p99 latency under 200ms for feature updates.

Data Labeling: How Not to Waste Budget

Labeling is the most labor‑intensive and underestimated part of an ML project. Poorly labeled data cannot be fixed by any architecture.

Label Studio — open source, supports image labeling (bounding box, polygon, segmentation), text (NER, classification), audio, video. Deploys in 10 minutes via Docker. For small teams — first choice.

Labeling quality assessment. Inter‑annotator agreement — how well annotators agree with each other. Cohen’s Kappa > 0.8 — good, 0.6‑0.8 — acceptable, < 0.6 — task ambiguous or instructions poor. Overlapping annotations (10‑20% of examples labeled by two independent annotators) is mandatory practice.

Active learning prevents budget waste. Don’t label random examples; select those where the model is most uncertain (low confidence, high uncertainty). Allows achieving the same quality with 50‑70% of the labeling volume. Modals, Prodigy, Label Studio support active learning workflows. In one NLP project, we reduced the labeling budget by 2.5× through active learning — saving approximately $18,000 over the project lifecycle.

Synthetic data. When real data is scarce or expensive to obtain. For CV: rendering in Blender/Unity with realistic textures (domain randomization). For NLP: paraphrase via LLM, backtranslation. Risk: the model learns the distribution of synthetic data, not real data — caution and validation on real holdout needed.

Data Quality: Validation and Monitoring

Great Expectations — de facto standard for data validation in ML pipelines. Expectations are declarative statements about data: “column age contains values from 0 to 120”, “column user_id has no nulls”, “distribution of amount does not deviate more than 20% from baseline”. Runs in the pipeline, on failure blocks progression. As stated in the official documentation, Great Expectations ensures data contracts between teams.

Pandera — Pythonic alternative for pandas/polars DataFrames. Schema‑based validation with type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. The model expects data from the last N days. ETL fails, data is not updated — the model uses stale features. Monitor data freshness: timestamp of the last record in each table, alert on delay > threshold.

Deduplication. Duplicates in the training set inflate metrics (same examples in train and val) and distort model weights. MinHash LSH for approximate deduplication of large datasets. For exact — hash by normalized content.

Validation Tools Comparison

What Does a Data Engineering Project for ML Include?

We provide the full cycle:

• Audit of existing data and pipelines (1 week).
• Architecture design: selection of tools, formats, labeling methods.
• Implementation of ETL/ELT pipeline with validation and monitoring.
• Documentation of code and processes (model card, data card).
• Training your team on pipeline operation.
• Post‑deployment support for 3 months.
• Access to code repository and all pipeline definitions.

How We Build a Pipeline: Step by Step

1. Audit existing data. Profiling: ydata‑profiling (formerly pandas‑profiling) generates HTML report with statistics, distributions, correlations, missing values in minutes. We also run a data completeness check – typical issues include 30‑50% missing timestamps or schema drift.
2. Pipeline design. Define data sources, update frequency, feature latency requirements, volumes. Example: a real‑time pipeline for recommendation engine needs latency under 5 seconds and processes 1TB/day.
3. Implementation and testing. Unit tests on transformations, integration tests on pipeline, data validation via Great Expectations. We target 95% test coverage for transformation logic.
4. Deployment and monitoring. Alerts on freshness, quality checks, anomalies in data volumes. Typical alert threshold: no new data for 2 hours.

Storage and Formats

For most ML projects at start: Parquet in S3 + DVC for versioning. Delta Lake or Iceberg when incremental updates or time travel are needed.

Why Trust Us

We have been working in data engineering and ML for over 8 years. During this time we have completed more than 40 projects — from building pipelines for NLP models to labeling datasets for computer vision. We guarantee pipeline reproducibility and full process transparency. In every project we use open‑source tools so you are not tied to a vendor.

Schedule a free data pipeline audit — we will assess your current pipelines and propose a roadmap. Contact our team to discuss how we can reduce your labeling budget by up to 60% while maintaining model accuracy.