Dominando a Governança de Dados e MDM em Sistemas ERP

Nota: Este artigo foi escrito para profissionais de TI, consultores de negócios e desenvolvedores que desejam elevar a qualidade dos dados em seus sistemas de gestão empresarial. Todo o conteúdo é original e traz exemplos de código prontos para uso.

Introdução

Os sistemas de gestão empresarial (ERP) são o coração operacional de empresas de todos os portes. Eles consolidam finanças, estoque, produção, recursos humanos e muito mais em uma única fonte de verdade. Contudo, a qualidade e a consistência dos dados são desafios recorrentes: informações duplicadas, campos desatualizados ou regras de negócio conflitantes podem gerar decisões equivocadas e custos desnecessários.

A resposta para esses problemas está na Governança de Dados – um conjunto de políticas, processos e tecnologias que asseguram que os dados sejam confiáveis, seguros e utilizáveis. Quando combinada ao Master Data Management (MDM), a governança cria um repositório centralizado de “dados mestres” (clientes, fornecedores, produtos, contas) que alimenta o ERP e todos os sistemas conectados.

Neste guia, vamos explorar:

Ao final, você terá um plano de ação para implementar uma camada de dados confiável que impulsiona a eficiência e a tomada de decisão.

1. Por que a Governança de Dados é crítica no ERP

A governança define propriedade, qualidade, segurança e acesso a cada elemento de dado. Sem ela, o ERP funciona como um “buraco negro” que absorve informações sem controle, comprometendo toda a cadeia de valor.

Principais pilares

2. Estrutura de Master Data Management (MDM) dentro do ERP

O MDM funciona como um hub central que mantém os registros mestres sincronizados entre o ERP e outros sistemas (CRM, e‑commerce, BI). A arquitetura típica inclui:

Modelo de Entidade‑Relacionamento simplificado

-- Tabela de Clientes (Master)

CREATE TABLE mdm_customer (
    customer_id   UUID PRIMARY KEY,
    legal_name    VARCHAR(200) NOT NULL,
    trade_name    VARCHAR(200),
    tax_id        VARCHAR(20)  UNIQUE NOT NULL,
    status        VARCHAR(20)  CHECK (status IN ('ACTIVE','INACTIVE')),
    created_at    TIMESTAMP    DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMP    DEFAULT CURRENT_TIMESTAMP
);


-- Tabela de Endereços (1:N)
CREATE TABLE mdm_address (
    address_id    UUID PRIMARY KEY,
    customer_id   UUID REFERENCES mdm_customer(customer_id),
    street        VARCHAR(200),
    city          VARCHAR(100),
    state         CHAR(2),
    zip_code      VARCHAR(10),
    is_primary    BOOLEAN DEFAULT FALSE,
    created_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Dica: Use UUIDs como chave primária para evitar colisões ao integrar dados de múltiplas origens.

Estratégia de deduplicação

Um algoritmo simples de deduplicação pode usar o CNPJ/CPF (tax_id) como chave natural, combinada com similaridade de nome (Levenshtein). Exemplo em Python:

# mdm_dedup.py

import uuid
from difflib import SequenceMatcher
from typing import List, Dict


def similar(a: str, b: str) -> float:
    return SequenceMatcher(None, a.lower(), b.lower()).ratio()


def deduplicate(customers: List[Dict]) -> List[Dict]:
    unique = {}
    for rec in customers:
        tax = rec['tax_id']
        if tax in unique:
            # já existe registro com mesmo tax_id → verifica nome
            if similar(rec['legal_name'], unique[tax]['legal_name']) < 0.85:
                # conflito grave → marcar para revisão manual
                rec['status'] = 'REVIEW'
                continue
        else:
            rec['customer_id'] = str(uuid.uuid4())
            unique[tax] = rec
    return list(unique.values())

Esse script pode ser inserido na pipeline de ingestão (por exemplo, um job do Apache Airflow) para garantir que somente registros únicos cheguem ao repositório MDM.

3. Estratégias de Integração de Dados: APIs, Barramento de Serviços e Eventos

A integração deve ser bidirecional: o ERP consome dados mestres e, simultaneamente, envia atualizações de volta ao MDM. As três abordagens mais comuns são:

Exemplo de API de negócios (FastAPI)

# api_mdm.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, constr
import asyncpg
import uuid


app = FastAPI(title="MDM Customer API")


class CustomerIn(BaseModel):
    legal_name: constr(min_length=1, max_length=200)
    trade_name: str = None
    tax_id: constr(regex=r'^\d{14}$')  # CNPJ simplificado
    status: str = "ACTIVE"


class CustomerOut(CustomerIn):
    customer_id: uuid.UUID


DATABASE_URL = "postgresql://user:pwd@db-host/mdm"


@app.post("/customers", response_model=CustomerOut)
async def create_customer(payload: CustomerIn):
    conn = await asyncpg.connect(DATABASE_URL)
    # Verifica duplicidade
    exists = await conn.fetchval(
        "SELECT 1 FROM mdm_customer WHERE tax_id=$1", payload.tax_id
    )
    if exists:
        await conn.close()
        raise HTTPException(status_code=409, detail="Cliente já cadastrado")
    cust_id = uuid.uuid4()
    await conn.execute(
        """
        INSERT INTO mdm_customer (customer_id, legal_name, trade_name, tax_id, status)
        VALUES ($1, $2, $3, $4, $5)
        """,
        cust_id, payload.legal_name, payload.trade_name, payload.tax_id, payload.status
    )
    await conn.close()
    return CustomerOut(customer_id=cust_id, payload.dict())

Observação: O endpoint acima usa asyncpg para conexão assíncrona ao PostgreSQL, garantindo alta performance em ambientes de alta concorrência.

# event_publisher.py

import json
import aio_pika


async def publish_customer_change(event_type: str, payload: dict):
    connection = await aio_pika.connect_robust("amqp://guest:guest@rabbitmq/")
    async with connection:
        channel = await connection.channel()
        exchange = await channel.declare_exchange("mdm.events", aio_pika.ExchangeType.FANOUT)
        message = aio_pika.Message(body=json.dumps({
            "type": event_type,
            "data": payload
        }).encode())
        await exchange.publish(message, routing_key="")

Um ERP costuma possuir workflows** (aprovação de compra, abertura de pedido, faturamento). Quando a qualidade dos dados é