Data Mesh e Architettura Decentralizzata dei Dati

+------------------------------------------------------------------+
|                    DATA DOMAIN: ORDINI                           |
+------------------------------------------------------------------+
|                                                                  |
|  +------------------+    +------------------+                    |
|  | Sorgenti Dati    |    | Servizi Applicat |                    |
|  | - PostgreSQL     |    | - Order API      |                    |
|  | - Event Stream   |    | - Order Worker   |                    |
|  | - Legacy ERP     |    | - Notification   |                    |
|  +--------+---------+    +--------+---------+                    |
|           |                       |                              |
|           v                       v                              |
|  +------------------------------------------------+             |
|  |         Pipeline di Trasformazione              |             |
|  |  - Ingestione (CDC / Event Streaming)           |             |
|  |  - Pulizia e Validazione                        |             |
|  |  - Aggregazione e Arricchimento                 |             |
|  +------------------------+-----------------------+              |
|                           |                                      |
|                           v                                      |
|  +------------------------------------------------+             |
|  |          DATA PRODUCTS (Output)                 |             |
|  |                                                 |             |
|  |  [1] ordini.fatto_ordini                        |             |
|  |      - Tabella Iceberg Gold                     |             |
|  |      - SLA: freshness 1h, completeness 99.5%    |             |
|  |                                                 |             |
|  |  [2] ordini.ordini_stream                       |             |
|  |      - Kafka topic (real-time)                  |             |
|  |      - SLA: latency <5s, availability 99.9%     |             |
|  |                                                 |             |
|  |  [3] ordini.metriche_giornaliere                |             |
|  |      - API REST / GraphQL                       |             |
|  |      - SLA: response time <200ms                |             |
|  +------------------------------------------------+             |
|                                                                  |
+------------------------------------------------------------------+

# data-contract.yaml - Contratto per il Data Product "Ordini"
# Formato basato su Data Contract Specification v0.9.3
# https://datacontract.com

dataContractSpecification: 0.9.3
id: urn:datacontract:ordini:fatto-ordini
info:
  title: Fatto Ordini
  version: 2.1.0
  description: |
    Tabella dei fatti contenente tutti gli ordini completati.
    Aggiornata ogni ora con CDC dal database operazionale.
  owner: team-ordini
  contact:
    name: Marco Rossi
    email: marco.rossi@azienda.it
    slack: "#team-ordini-data"

servers:
  production:
    type: iceberg
    catalog: lakehouse
    database: ordini
    table: fatto_ordini
    location: s3://data-lakehouse/ordini/fatto_ordini/

schema:
  type: table
  fields:
    - name: ordine_id
      type: bigint
      required: true
      unique: true
      description: Identificativo univoco dell'ordine
      pii: false
    - name: cliente_id
      type: integer
      required: true
      description: FK al dominio Customer
      references: urn:datacontract:customer:dim-clienti.cliente_id
    - name: data_ordine
      type: timestamp
      required: true
      description: Timestamp di creazione dell'ordine (UTC)
    - name: importo_totale
      type: decimal(12,2)
      required: true
      description: Importo totale dell'ordine in EUR
      checks:
        - type: range
          min: 0.01
          max: 999999.99
    - name: stato
      type: string
      required: true
      description: Stato corrente dell'ordine
      enum: [completato, annullato, rimborsato, in_lavorazione]
    - name: canale
      type: string
      required: true
      description: Canale di vendita
      enum: [web, mobile_app, marketplace, pos]
    - name: regione_cliente
      type: string
      required: false
      description: Regione di residenza del cliente

quality:
  type: SodaCL
  specification:
    checks for fatto_ordini:
      - row_count > 0
      - freshness(data_ordine) < 2h
      - missing_percent(ordine_id) = 0
      - missing_percent(importo_totale) = 0
      - invalid_percent(importo_totale) < 0.1%:
          valid min: 0.01
      - duplicate_percent(ordine_id) = 0
      - schema:
          name: fatto_ordini_schema
          warn:
            when schema changes: any

sla:
  freshness: 1h           # Dati aggiornati entro 1 ora
  availability: 99.5%      # Uptime della tabella
  completeness: 99.9%      # Percentuale record completi
  latency_p95: 5s          # Tempo query P95

terms:
  usage: |
    Dati disponibili per analytics, reporting e ML.
    Non utilizzare per comunicazioni dirette al cliente
    senza consenso esplicito del dominio Customer.
  retention: 7 anni (obbligo fiscale)
  classification: internal
  pii_fields: [cliente_id]

history:
  - version: 2.1.0
    date: 2025-12-01
    changes: Aggiunto campo canale
  - version: 2.0.0
    date: 2025-06-15
    changes: Breaking change - rinominato amount in importo_totale

+============================================================================+
|                    SELF-SERVE DATA PLATFORM                                |
+============================================================================+
|                                                                            |
|  +---------------------------+  +---------------------------+              |
|  |   Data Product Builder    |  |   Data Product Discovery  |              |
|  |   - Template pipeline     |  |   - Data Catalog          |              |
|  |   - Schema registry       |  |   - Search & Browse       |              |
|  |   - Contract generator    |  |   - Lineage viewer        |              |
|  |   - CI/CD per dati        |  |   - Quality dashboard     |              |
|  +---------------------------+  +---------------------------+              |
|                                                                            |
|  +---------------------------+  +---------------------------+              |
|  |   Infrastructure Layer    |  |   Governance Layer        |              |
|  |   - Compute provisioning  |  |   - Policy engine         |              |
|  |   - Storage management    |  |   - Access control (RBAC) |              |
|  |   - Networking            |  |   - Audit logging         |              |
|  |   - Monitoring & Alerts   |  |   - Compliance checks     |              |
|  +---------------------------+  +---------------------------+              |
|                                                                            |
|  +---------------------------+  +---------------------------+              |
|  |   Mesh Connectivity       |  |   Observability           |              |
|  |   - Event bus (Kafka)     |  |   - Data quality metrics  |              |
|  |   - API gateway           |  |   - SLA monitoring        |              |
|  |   - Query federation      |  |   - Cost tracking         |              |
|  |   - Cross-domain joins    |  |   - Usage analytics       |              |
|  +---------------------------+  +---------------------------+              |
|                                                                            |
+============================================================================+

# terraform/modules/data-product/main.tf
# Modulo Terraform per il provisioning di un Data Product

variable "domain_name" {
  type        = string
  description = "Nome del dominio (es: ordini, catalogo)"
}

variable "product_name" {
  type        = string
  description = "Nome del data product"
}

variable "owner_team" {
  type        = string
  description = "Team responsabile"
}

variable "sla_tier" {
  type    = string
  default = "standard"  # standard | premium | critical
}

# Storage: bucket S3 dedicato per il dominio
resource "aws_s3_bucket" "data_product" {
  bucket = "data-mesh-${var.domain_name}-${var.product_name}"

  tags = {
    Domain    = var.domain_name
    Product   = var.product_name
    Owner     = var.owner_team
    ManagedBy = "data-platform"
    SLATier   = var.sla_tier
  }
}

# Iceberg: tabella registrata nel catalogo
resource "aws_glue_catalog_table" "iceberg_table" {
  database_name = var.domain_name
  name          = var.product_name

  table_type = "EXTERNAL_TABLE"
  parameters = {
    "table_type"          = "ICEBERG"
    "metadata_location"   = "s3://${aws_s3_bucket.data_product.id}/metadata/"
    "data_contract_url"   = "s3://data-contracts/${var.domain_name}/${var.product_name}.yaml"
  }
}

# IAM: ruolo per il team di dominio
resource "aws_iam_role" "domain_role" {
  name = "data-mesh-${var.domain_name}-${var.product_name}-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = {
        AWS = "arn:aws:iam::role/${var.owner_team}"
      }
    }]
  })
}

# Monitoring: dashboard CloudWatch automatica
resource "aws_cloudwatch_dashboard" "data_product" {
  dashboard_name = "data-product-${var.domain_name}-${var.product_name}"
  dashboard_body = templatefile("${path.module}/dashboard.json.tpl", {
    domain  = var.domain_name
    product = var.product_name
    sla     = var.sla_tier
  })
}

# Output: informazioni per il team
output "data_product_uri" {
  value = "s3://${aws_s3_bucket.data_product.id}/"
}

output "catalog_table" {
  value = "${var.domain_name}.${var.product_name}"
}

# governance/policies/data_product_policy.rego
# Policy OPA per la validazione dei Data Product

package datamesh.governance

# Regola 1: Ogni data product DEVE avere un data contract valido
deny[msg] {
    not input.data_contract
    msg := "Data product deve avere un data contract definito"
}

# Regola 2: Ogni data product DEVE avere un owner
deny[msg] {
    not input.data_contract.info.owner
    msg := "Data contract deve specificare il team owner"
}

# Regola 3: I campi PII devono essere dichiarati
deny[msg] {
    field := input.data_contract.schema.fields[_]
    contains(lower(field.name), "email")
    not field.pii
    msg := sprintf("Campo '%s' potrebbe contenere PII ma non e contrassegnato", [field.name])
}

# Regola 4: SLA obbligatori per data product in produzione
deny[msg] {
    input.environment == "production"
    not input.data_contract.sla
    msg := "SLA obbligatori per data product in produzione"
}

# Regola 5: Freshness SLA deve essere definito
deny[msg] {
    input.environment == "production"
    not input.data_contract.sla.freshness
    msg := "SLA di freshness obbligatorio per data product in produzione"
}

# Regola 6: Naming convention per le tabelle
deny[msg] {
    table_name := input.data_contract.servers.production.table
    not re_match("^[a-z][a-z0-9_]*$", table_name)
    msg := sprintf("Nome tabella '%s' non conforme: usa solo lowercase, numeri e underscore", [table_name])
}

# Regola 7: Data classification obbligatoria
deny[msg] {
    not input.data_contract.terms.classification
    msg := "Classification obbligatoria (public, internal, confidential, restricted)"
}

# Regola 8: Retention policy obbligatoria per compliance GDPR
deny[msg] {
    not input.data_contract.terms.retention
    msg := "Retention policy obbligatoria per compliance GDPR"
}

# governance/standards/global_types.yaml
# Standard globali per il Data Mesh aziendale

naming_conventions:
  tables:
    pattern: "^[a-z][a-z0-9_]{2,60}$"
    prefixes:
      fact: "fatto_"       # Tabelle dei fatti (fact tables)
      dimension: "dim_"    # Dimensioni
      staging: "stg_"      # Staging temporaneo
      aggregate: "agg_"    # Aggregazioni pre-calcolate

  columns:
    pattern: "^[a-z][a-z0-9_]{1,60}$"
    reserved_suffixes:
      _id: "Identificativo (integer o bigint)"
      _ts: "Timestamp (UTC)"
      _dt: "Data senza orario"
      _amt: "Importo monetario (decimal)"
      _qty: "Quantità (integer)"
      _pct: "Percentuale (decimal 0-100)"
      _flag: "Booleano"

global_types:
  currency:
    type: decimal(12,2)
    description: "Importi monetari in EUR"
  identifier:
    type: bigint
    description: "Chiavi primarie e foreign key"
  timestamp:
    type: timestamp
    timezone: UTC
    description: "Tutti i timestamp in UTC"
  country_code:
    type: string
    format: ISO-3166-1-alpha-2
    description: "Codice paese ISO a 2 lettere"

interoperability:
  shared_dimensions:
    - name: dim_data
      owner: platform-team
      description: "Dimensione temporale condivisa (calendario)"
    - name: dim_geografia
      owner: platform-team
      description: "Gerarchia geografica (nazione, regione, provincia, comune)"
    - name: dim_valuta
      owner: platform-team
      description: "Conversioni valuta giornaliere"

compliance:
  gdpr:
    pii_fields_must_be_tagged: true
    retention_policy_required: true
    right_to_deletion_supported: true
  data_classification:
    levels: [public, internal, confidential, restricted]
    default: internal

+============================================================================+
|                        FEDERATED GOVERNANCE LAYER                          |
|   Policy Engine (OPA) | Schema Registry | Compliance Automation            |
+============================================================================+
        |                        |                        |
        v                        v                        v
+--------------------+  +--------------------+  +--------------------+
|  DOMAIN: ORDINI    |  |  DOMAIN: CATALOGO  |  |  DOMAIN: CUSTOMER  |
|                    |  |                    |  |                    |
| +----------------+ |  | +----------------+ |  | +----------------+ |
| | Data Products  | |  | | Data Products  | |  | | Data Products  | |
| | - fatto_ordini | |  | | - dim_prodotti | |  | | - dim_clienti  | |
| | - ordini_stream| |  | | - prezzi_storico| |  | | - segmentazione| |
| | - metriche_day | |  | | - inventory    | |  | | - comportamento| |
| +----------------+ |  | +----------------+ |  | +----------------+ |
|                    |  |                    |  |                    |
| +----------------+ |  | +----------------+ |  | +----------------+ |
| | Pipeline (dbt) | |  | | Pipeline (dbt) | |  | | Pipeline (dbt) | |
| | - bronze       | |  | | - bronze       | |  | | - bronze       | |
| | - silver       | |  | | - silver       | |  | | - silver       | |
| | - gold         | |  | | - gold         | |  | | - gold         | |
| +----------------+ |  | +----------------+ |  | +----------------+ |
|                    |  |                    |  |                    |
| [Team: 5+1 DE]    |  | [Team: 4+1 DE]    |  | [Team: 3+1 DE]    |
+--------------------+  +--------------------+  +--------------------+
        |                        |                        |
        v                        v                        v
+============================================================================+
|                     SELF-SERVE DATA PLATFORM                               |
|                                                                            |
|  +-------------------+ +-------------------+ +-------------------+         |
|  | Compute Layer     | | Storage Layer     | | Connectivity      |         |
|  | - Spark / dbt     | | - S3 / ADLS      | | - Kafka           |         |
|  | - Airflow         | | - Iceberg tables  | | - API Gateway     |         |
|  | - Kubernetes      | | - Schema Registry | | - Query Federation|         |
|  +-------------------+ +-------------------+ +-------------------+         |
|                                                                            |
|  +-------------------+ +-------------------+ +-------------------+         |
|  | Data Catalog      | | Observability     | | Security          |         |
|  | - OpenMetadata    | | - Great Expect.   | | - RBAC / ABAC     |         |
|  | - Lineage         | | - SLA Monitoring  | | - Encryption      |         |
|  | - Search          | | - Cost Tracking   | | - Audit Trail     |         |
|  +-------------------+ +-------------------+ +-------------------+         |
+============================================================================+

-- dbt/models/ordini/gold/fatto_ordini.sql
-- Modello dbt per il Data Product "fatto_ordini"
-- Dominio: Ordini | Layer: Gold | Owner: team-ordini

{{ config(
    materialized='incremental',
    unique_key='ordine_id',
    partition_by={
        'field': 'data_ordine',
        'data_type': 'timestamp',
        'granularity': 'day'
    },
    tags=['gold', 'ordini', 'data-product'],
    meta={
        'owner': 'team-ordini',
        'sla_freshness': '1h',
        'sla_availability': '99.5%',
        'data_contract': 'urn:datacontract:ordini:fatto-ordini'
    }
) }}

WITH ordini_silver AS (
    SELECT * FROM {{ ref('stg_ordini_clean') }}
    {% if is_incremental() %}
    WHERE data_aggiornamento > (
        SELECT MAX(data_aggiornamento) FROM {{ this }}
    )
    {% endif %}
),

clienti AS (
    -- Cross-domain reference: consuma dal dominio Customer
    SELECT * FROM {{ source('customer_domain', 'dim_clienti') }}
),

arricchiti AS (
    SELECT
        o.ordine_id,
        o.cliente_id,
        o.data_ordine,
        o.importo_totale,
        o.stato,
        o.canale,
        c.regione AS regione_cliente,
        c.segmento AS segmento_cliente,
        o.data_aggiornamento
    FROM ordini_silver o
    LEFT JOIN clienti c ON o.cliente_id = c.cliente_id
)

SELECT * FROM arricchiti

# dbt/models/ordini/gold/schema.yml
# Test e documentazione per il data product fatto_ordini

version: 2

models:
  - name: fatto_ordini
    description: >
      Tabella dei fatti degli ordini completati.
      Data Product del dominio Ordini, aggiornato ogni ora.
    meta:
      owner: team-ordini
      data_contract: urn:datacontract:ordini:fatto-ordini
    columns:
      - name: ordine_id
        description: Identificativo univoco dell'ordine
        tests:
          - unique
          - not_null
      - name: cliente_id
        description: FK al dominio Customer
        tests:
          - not_null
          - relationships:
              to: source('customer_domain', 'dim_clienti')
              field: cliente_id
      - name: importo_totale
        description: Importo totale in EUR
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0.01
              max_value: 999999.99
      - name: stato
        description: Stato dell'ordine
        tests:
          - accepted_values:
              values: ['completato', 'annullato', 'rimborsato', 'in_lavorazione']
      - name: data_ordine
        description: Timestamp di creazione (UTC)
        tests:
          - not_null
          - dbt_utils.recency:
              datepart: hour
              field: data_ordine
              interval: 2

# api/ordini_data_product.py
# API per il Data Product "Ordini" - Dominio Ordini

from fastapi import FastAPI, Query, HTTPException
from pydantic import BaseModel
from typing import Optional
from datetime import date, datetime
import duckdb

app = FastAPI(
    title="Ordini Data Product API",
    version="2.1.0",
    description="API per il consumo del data product Ordini"
)

class MetricheOrdini(BaseModel):
    regione: str
    data: date
    num_ordini: int
    fatturato: float
    ordine_medio: float
    clienti_unici: int

class HealthCheck(BaseModel):
    status: str
    freshness_minutes: int
    record_count: int
    last_update: datetime

# Connessione a DuckDB/Iceberg
con = duckdb.connect()
con.execute("""
    INSTALL iceberg;
    LOAD iceberg;
""")

@app.get("/health", response_model=HealthCheck)
async def health_check():
    """Verifica SLA del data product"""
    result = con.execute("""
        SELECT
            COUNT(*) as record_count,
            MAX(data_aggiornamento) as last_update,
            DATEDIFF('minute', MAX(data_aggiornamento), NOW())
                AS freshness_minutes
        FROM iceberg_scan('s3://data-mesh/ordini/fatto_ordini/')
    """).fetchone()

    freshness = result[2]
    status = "healthy" if freshness < 60 else "degraded"

    return HealthCheck(
        status=status,
        freshness_minutes=freshness,
        record_count=result[0],
        last_update=result[1]
    )

@app.get("/metriche", response_model=list[MetricheOrdini])
async def get_metriche(
    data_inizio: date = Query(..., description="Data inizio (YYYY-MM-DD)"),
    data_fine: date = Query(..., description="Data fine (YYYY-MM-DD)"),
    regione: Optional[str] = Query(None, description="Filtro regione"),
    limit: int = Query(100, ge=1, le=1000)
):
    """Metriche aggregate degli ordini per regione e giorno"""
    query = """
        SELECT
            regione_cliente AS regione,
            CAST(data_ordine AS DATE) AS data,
            COUNT(*) AS num_ordini,
            ROUND(SUM(importo_totale), 2) AS fatturato,
            ROUND(AVG(importo_totale), 2) AS ordine_medio,
            COUNT(DISTINCT cliente_id) AS clienti_unici
        FROM iceberg_scan('s3://data-mesh/ordini/fatto_ordini/')
        WHERE data_ordine BETWEEN ? AND ?
    """
    params = [data_inizio, data_fine]

    if regione:
        query += " AND regione_cliente = ?"
        params.append(regione)

    query += """
        GROUP BY regione_cliente, CAST(data_ordine AS DATE)
        ORDER BY fatturato DESC
        LIMIT ?
    """
    params.append(limit)

    rows = con.execute(query, params).fetchall()
    if not rows:
        raise HTTPException(status_code=404, detail="Nessun dato trovato")

    return [
        MetricheOrdini(
            regione=r[0], data=r[1], num_ordini=r[2],
            fatturato=r[3], ordine_medio=r[4], clienti_unici=r[5]
        )
        for r in rows
    ]

# .github/workflows/data-product-ci.yml
name: Data Product CI/CD - Ordini

on:
  push:
    paths:
      - 'domains/ordini/**'
  pull_request:
    paths:
      - 'domains/ordini/**'

jobs:
  validate-contract:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Validate Data Contract Schema
        run: |
          pip install datacontract-cli
          datacontract test domains/ordini/data-contract.yaml

      - name: Check Breaking Changes
        run: |
          datacontract breaking \
            domains/ordini/data-contract.yaml \
            --with-previous-version

  test-transformations:
    runs-on: ubuntu-latest
    needs: validate-contract
    steps:
      - uses: actions/checkout@v4

      - name: Install dbt
        run: pip install dbt-core dbt-duckdb

      - name: Run dbt Tests
        run: |
          cd domains/ordini/dbt
          dbt deps
          dbt seed --target test
          dbt run --target test
          dbt test --target test

      - name: Check Data Quality (Soda)
        run: |
          pip install soda-core-duckdb
          soda scan -d ordini_test \
            -c domains/ordini/soda/configuration.yml \
            domains/ordini/soda/checks.yml

  governance-check:
    runs-on: ubuntu-latest
    needs: validate-contract
    steps:
      - uses: actions/checkout@v4

      - name: Policy Validation (OPA)
        run: |
          opa eval \
            --data governance/policies/ \
            --input domains/ordini/data-contract.yaml \
            "data.datamesh.governance.deny"

      - name: Schema Registry Compatibility
        run: |
          curl -X POST \
            "$SCHEMA_REGISTRY_URL/compatibility/subjects/ordini-fatto_ordini/versions/latest" \
            -H "Content-Type: application/json" \
            -d "$(cat domains/ordini/schema.json)"

  deploy:
    runs-on: ubuntu-latest
    needs: [test-transformations, governance-check]
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Deploy Data Product
        run: |
          dbt run --target production
          datacontract publish domains/ordini/data-contract.yaml \
            --catalog-url $CATALOG_URL