Meltano : Le Guide Complet de l'ELT

Analogie pour débutants

Si dbt est le chef cuisinier qui transforme les ingrédients en plats, Meltano est le livreur qui va chercher les ingrédients chez les fournisseurs (APIs, bases de données, fichiers CSV) et les range dans le frigo (data warehouse). Meltano est aussi le manager du restaurant qui coordonne les livraisons et la cuisine pour que tout se passe dans le bon ordre, au bon moment.

# Ajouter un tap pour extraire depuis PostgreSQL
$ meltano add extractor tap-postgres

# Ajouter un tap pour l'API GitHub
$ meltano add extractor tap-github

# Ajouter un tap pour des fichiers CSV
$ meltano add extractor tap-csv

# Ajouter un target pour charger dans Snowflake
$ meltano add loader target-snowflake

# Charger dans PostgreSQL
$ meltano add loader target-postgres

# Charger dans des fichiers JSON (debug)
$ meltano add loader target-jsonl

# meltano.yml
version: 1
project_id: mon-projet-data

plugins:
  extractors:
    - name: tap-postgres
      variant: meltanolabs
      pip_url: meltanolabs-tap-postgres
      config:
        host: localhost
        port: 5432
        database: app_production
        user: readonly_user
      select:
        - public-orders.*    # Sélectionner toutes les colonnes
        - public-customers.* # de ces tables
        - "!public-logs.*"   # Exclure cette table

  loaders:
    - name: target-snowflake
      variant: meltanolabs
      pip_url: meltanolabs-target-snowflake
      config:
        account: xy12345.eu-west-1
        database: RAW
        schema: POSTGRES
        warehouse: LOADING_WH

  transformers:
    - name: dbt-snowflake
      pip_url: dbt-core dbt-snowflake

jobs:
  - name: daily-sync
    tasks:
      - tap-postgres target-snowflake dbt-snowflake:run

schedules:
  - name: daily-sync-schedule
    job: daily-sync
    interval: "@daily"

# meltano.yml
environments:
  - name: dev
    config:
      plugins:
        extractors:
          - name: tap-postgres
            config:
              host: localhost
              database: app_dev

  - name: prod
    config:
      plugins:
        extractors:
          - name: tap-postgres
            config:
              host: prod-db.company.com
              database: app_production

# Exécution :
$ meltano --environment=dev run tap-postgres target-snowflake
$ meltano --environment=prod run tap-postgres target-snowflake

Première exécution : récupère tout depuis le début.
Exécutions suivantes : récupère seulement les données créées/modifiées depuis la dernière exécution.
Cet état est stocké localement ou dans un backend configurable (base de données, S3, etc.).

PostgreSQL → Snowflake

# 1. Initialiser le projet
$ meltano init mon-pipeline && cd mon-pipeline

# 2. Ajouter les plugins
$ meltano add extractor tap-postgres
$ meltano add loader target-snowflake

# 3. Configurer (ou via meltano.yml)
$ meltano config tap-postgres set host db.company.com
$ meltano config tap-postgres set database production
$ meltano config tap-postgres set user readonly

# 4. Sélectionner les tables
$ meltano select tap-postgres "public-users" "*"
$ meltano select tap-postgres "public-orders" "*"
$ meltano select tap-postgres "public-payments" "*"

# 5. Lancer la synchronisation
$ meltano run tap-postgres target-snowflake

GitHub → PostgreSQL

# Synchroniser les issues, PRs et commits
$ meltano add extractor tap-github
$ meltano config tap-github set repositories '["org/repo1", "org/repo2"]'
$ meltano config tap-github set auth_token $GITHUB_TOKEN
$ meltano select tap-github "issues" "*"
$ meltano select tap-github "pull_requests" "*"
$ meltano select tap-github "commits" "*"

$ meltano add loader target-postgres
$ meltano run tap-github target-postgres

Shopify + Stripe → Snowflake → dbt → Metabase

# meltano.yml - Jobs composés
jobs:
  - name: ecommerce-pipeline
    tasks:
      # Étape 1 : Extraire Shopify
      - tap-shopify target-snowflake
      # Étape 2 : Extraire Stripe
      - tap-stripe target-snowflake
      # Étape 3 : Transformer avec dbt
      - dbt-snowflake:run
      # Étape 4 : Tester avec dbt
      - dbt-snowflake:test

schedules:
  - name: ecommerce-daily
    job: ecommerce-pipeline
    interval: "@daily"  # Tous les jours à minuit

# Extraire depuis S3 + charger dans BigQuery
$ meltano add extractor tap-s3-csv
$ meltano config tap-s3-csv set bucket my-logs-bucket
$ meltano config tap-s3-csv set prefix "logs/2026/"

$ meltano add loader target-bigquery
$ meltano run tap-s3-csv target-bigquery

# MySQL production → PostgreSQL analytics
$ meltano add extractor tap-mysql
$ meltano add loader target-postgres

# Sync complet puis incrémental
$ meltano run tap-mysql target-postgres
# La 2e exécution ne prend que les deltas

# meltano.yml — Pipeline multi-sources
jobs:
  - name: marketing-domain
    tasks:
      - tap-google-ads target-snowflake
      - tap-facebook target-snowflake
      - tap-hubspot target-snowflake
  - name: product-domain
    tasks:
      - tap-segment target-snowflake
      - tap-mixpanel target-snowflake
  - name: finance-domain
    tasks:
      - tap-stripe target-snowflake
      - tap-quickbooks target-snowflake

# Initialiser un projet
meltano init mon-projet

# Lister les plugins disponibles
meltano discover extractors
meltano discover loaders

# Ajouter un plugin
meltano add extractor tap-postgres

# Configurer un plugin
meltano config tap-postgres set host localhost
meltano config tap-postgres list  # Voir la config

# Sélectionner des tables
meltano select tap-postgres --list --all  # Voir tout
meltano select tap-postgres "public-users" "*"

# Exécuter un pipeline
meltano run tap-postgres target-snowflake

# Exécuter avec dbt
meltano run tap-postgres target-snowflake dbt-snowflake:run

# Gérer l'état
meltano state list
meltano state get dev:tap-postgres-to-target-snowflake

# Orchestration
meltano schedule list
meltano invoke airflow scheduler

# Sélectionner des tables spécifiques
$ meltano select tap-postgres "public-orders" "*"

# Exclure des colonnes sensibles
$ meltano select tap-postgres "public-users" "*"
$ meltano select tap-postgres "public-users" "!password"
$ meltano select tap-postgres "public-users" "!ssn"

# Utiliser des wildcards
$ meltano select tap-postgres "public-*" "*"  # Toutes les tables du schéma public

# Option 1 : Variables d'environnement
$ export TAP_POSTGRES_PASSWORD=my_secret
# Dans meltano.yml : password: $TAP_POSTGRES_PASSWORD

# Option 2 : Fichier .env (non versionné)
# .env
TAP_POSTGRES_PASSWORD=my_secret
TARGET_SNOWFLAKE_PASSWORD=another_secret

# Option 3 : Vault (avancé)
# Intégration avec HashiCorp Vault, AWS Secrets Manager, etc.

# Ajouter Airflow comme orchestrateur
$ meltano add orchestrator airflow

# Créer un schedule
$ meltano schedule add daily-sync \
    --job daily-pipeline \
    --interval "@daily"

# Démarrer le scheduler
$ meltano invoke airflow scheduler

# L'UI Airflow est accessible sur http://localhost:8080

# meltano.yml
plugins:
  mappers:
    - name: meltano-map-transformer
      variant: meltano
      pip_url: meltano-map-transformer
      mappings:
        - name: hash-emails
          config:
            stream_maps:
              users:
                email: null              # Supprimer le champ email
                email_domain: "record['email'].split('@')[1]"
                __else__: null           # Garder les autres champs

# Exécution avec le mapper
$ meltano run tap-postgres hash-emails target-snowflake

# 1. Créer le squelette du tap
$ pip install cookiecutter
$ cookiecutter https://github.com/meltano/sdk \
    --directory="cookiecutter/tap-template"

# 2. Implémenter le stream
# tap_myapi/streams.py
from singer_sdk import typing as th
from singer_sdk.streams import RESTStream

class UsersStream(RESTStream):
    name = "users"
    path = "/api/v1/users"
    primary_keys = ["id"]
    replication_key = "updated_at"

    schema = th.PropertiesList(
        th.Property("id", th.IntegerType),
        th.Property("name", th.StringType),
        th.Property("email", th.StringType),
        th.Property("updated_at", th.DateTimeType),
    ).to_dict()

    def get_url_params(self, context, next_page_token):
        params = {"page_size": 100}
        if next_page_token:
            params["page"] = next_page_token
        return params

# Architecture des schémas dans le warehouse
RAW.SHOPIFY      ← Meltano charge ici (données brutes)
RAW.STRIPE       ← Meltano charge ici
ANALYTICS.STG    ← dbt staging (nettoyage)
ANALYTICS.INT    ← dbt intermediate (logique)
ANALYTICS.MARTS  ← dbt marts (tables finales)

# meltano.yml - Deux schedules
schedules:
  - name: daily-incremental
    job: daily-sync        # Incrémental
    interval: "@daily"
  - name: weekly-full
    job: full-refresh      # Full table
    interval: "0 2 * * 0"  # Dimanche 2h

jobs:
  - name: full-refresh
    tasks:
      - tap-postgres target-snowflake --full-refresh

L'incrémental quotidien est rapide mais peut accumuler des dérives. Le full refresh hebdomadaire corrige les écarts.

• Choisissez Meltano si : vous voulez le contrôle total, vous êtes à l'aise avec le CLI et le code, vous voulez éviter le vendor lock-in, et votre budget est limité.
• Choisissez Fivetran si : vous ne voulez rien gérer, votre équipe est non-technique, et le budget n'est pas un problème.
• Choisissez Airbyte si : vous voulez un compromis entre open-source et facilité d'utilisation, avec une UI web pour la configuration.

• Paralléliser les pipelines : exécuter tap-shopify et tap-stripe en parallèle (pas en séquence)
• Utiliser des workers Airflow : distribuer l'exécution sur plusieurs machines
• Optimiser la sélection : ne synchroniser que les tables/colonnes nécessaires
• Partitionner les grands extracts : utiliser des plages de dates pour les tables volumineuses