# Creer un topic avec 6 partitions et facteur de replication 3
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic commandes \
  --partitions 6 \
  --replication-factor 3

# Decrire un topic existant
kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic commandes

# Resultat typique :
# Topic: commandes  Partitions: 6  ReplicationFactor: 3
# Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3
# Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3,1
# ...

# Configuration KRaft (server.properties)
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@broker1:9093,2@broker2:9093,3@broker3:9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
log.dirs=/var/kafka-logs
metadata.log.dir=/var/kafka-metadata

// Configuration d'un producer idempotent
Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Idempotence : elimine les doublons en cas de retry
props.put("enable.idempotence", true);  // implicitement acks=all
props.put("max.in.flight.requests.per.connection", 5); // max 5 avec idempotence
props.put("retries", Integer.MAX_VALUE);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Envoi avec callback pour gestion d'erreur
producer.send(new ProducerRecord<>("commandes", orderId, orderJson),
    (metadata, exception) -> {
        if (exception != null) {
            log.error("Echec envoi commande {}: {}", orderId, exception.getMessage());
            // Logique de compensation
        } else {
            log.info("Commande {} envoyee - partition:{} offset:{}",
                orderId, metadata.partition(), metadata.offset());
        }
    });

// Producer transactionnel pour exactly-once
props.put("transactional.id", "payment-processor-1");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();

    // Lire les messages d'entree
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, String> record : records) {
        // Traitement metier
        String result = processPayment(record.value());

        // Ecriture du resultat dans le topic de sortie
        producer.send(new ProducerRecord<>("paiements-valides", record.key(), result));
    }

    // Commit des offsets ET de la production dans une seule transaction
    producer.sendOffsetsToTransaction(
        getOffsetsToCommit(records),
        consumer.groupMetadata()
    );

    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

// Consumer avec commit manuel - at-least-once semantics
Properties cProps = new Properties();
cProps.put("bootstrap.servers", "broker1:9092");
cProps.put("group.id", "payment-service");
cProps.put("enable.auto.commit", false);
cProps.put("auto.offset.reset", "earliest"); // latest | earliest | none
cProps.put("isolation.level", "read_committed"); // pour EOS

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(cProps);
consumer.subscribe(Arrays.asList("commandes"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processOrder(record);  // Traitement metier
    }
    consumer.commitSync();  // Commit apres traitement complet
}

{
  "name": "postgres-source-commandes",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "db-prod.internal",
    "database.port": "5432",
    "database.user": "cdc_reader",
    "database.password": "${vault:secret/kafka/db-password}",
    "database.dbname": "ecommerce",
    "table.include.list": "public.commandes,public.lignes_commande",
    "topic.prefix": "cdc.ecommerce",
    "plugin.name": "pgoutput",

    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",

    "transforms": "route,unwrap,addTimestamp",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "cdc\\.ecommerce\\.(.*)",
    "transforms.route.replacement": "streaming.$1",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.addTimestamp.timestamp.field": "processed_at",

    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq-postgres-commandes",
    "errors.deadletterqueue.topic.replication.factor": 3,
    "errors.deadletterqueue.context.headers.enable": true
  }
}

{
  "type": "record",
  "name": "Commande",
  "namespace": "com.ecommerce.events",
  "fields": [
    {"name": "commande_id", "type": "string"},
    {"name": "client_id", "type": "string"},
    {"name": "montant", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}},
    {"name": "devise", "type": {"type": "enum", "name": "Devise", "symbols": ["EUR", "USD", "GBP"]}},
    {"name": "statut", "type": "string"},
    {"name": "created_at", "type": {"type": "long", "logicalType": "timestamp-millis"}},
    {"name": "commentaire", "type": ["null", "string"], "default": null}
  ]
}

// Kafka Streams - Exemple complet de traitement de commandes
StreamsBuilder builder = new StreamsBuilder();

// KStream : flux de commandes (chaque commande est un evenement)
KStream<String, Commande> commandes = builder.stream(
    "commandes",
    Consumed.with(Serdes.String(), commandeSerde)
);

// KTable : table des clients (derniere info connue par client)
KTable<String, Client> clients = builder.table(
    "clients",
    Consumed.with(Serdes.String(), clientSerde)
);

// Enrichir chaque commande avec les infos client (stream-table join)
KStream<String, CommandeEnrichie> commandesEnrichies = commandes
    .selectKey((key, commande) -> commande.getClientId())
    .join(clients,
        (commande, client) -> new CommandeEnrichie(commande, client)
    );

// Aggregation : chiffre d'affaires par categorie sur fenetre de 1 heure
KTable<Windowed<String>, Double> caParCategorie = commandesEnrichies
    .groupBy((key, ce) -> ce.getCategorie(),
        Grouped.with(Serdes.String(), commandeEnrichieSerde))
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .aggregate(
        () -> 0.0,
        (categorie, ce, total) -> total + ce.getMontant(),
        Materialized.<String, Double, WindowStore<Bytes, byte[]>>as("ca-par-categorie")
            .withValueSerde(Serdes.Double())
    );

// Ecrire le resultat dans un topic de sortie
caParCategorie.toStream()
    .map((windowedKey, montant) -> KeyValue.pair(windowedKey.key(), montant))
    .to("ca-par-categorie-hourly", Produced.with(Serdes.String(), Serdes.Double()));

-- Creer un stream depuis un topic Kafka
CREATE STREAM commandes_stream (
  commande_id VARCHAR KEY,
  client_id VARCHAR,
  montant DECIMAL(10,2),
  categorie VARCHAR,
  created_at TIMESTAMP
) WITH (
  KAFKA_TOPIC = 'commandes',
  VALUE_FORMAT = 'AVRO',
  TIMESTAMP = 'created_at'
);

-- Vue materialisee : total par client (KTable)
CREATE TABLE total_par_client AS
  SELECT client_id,
         COUNT(*) AS nb_commandes,
         SUM(montant) AS total_depense,
         MAX(created_at) AS derniere_commande
  FROM commandes_stream
  GROUP BY client_id
  EMIT CHANGES;

-- Push query : flux continu de mises a jour (long-running)
SELECT * FROM total_par_client EMIT CHANGES;

-- Pull query : etat actuel ponctuel (request/response)
SELECT * FROM total_par_client WHERE client_id = 'CLI-42';

-- Detection d'anomalie : clients avec plus de 10 commandes en 5 min
CREATE TABLE alertes_fraude AS
  SELECT client_id,
         COUNT(*) AS nb_commandes,
         SUM(montant) AS total
  FROM commandes_stream
  WINDOW TUMBLING (SIZE 5 MINUTES)
  GROUP BY client_id
  HAVING COUNT(*) > 10
  EMIT CHANGES;

// Flink DataStream - Detection de fraude en temps reel
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Configuration du checkpointing (exactly-once)
env.enableCheckpointing(60_000); // checkpoint toutes les 60s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setCheckpointTimeout(120_000);
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// Source Kafka
KafkaSource<Transaction> source = KafkaSource.<Transaction>builder()
    .setBootstrapServers("broker1:9092")
    .setTopics("transactions")
    .setGroupId("fraud-detector")
    .setStartingOffsets(OffsetsInitializer.committedOffsets())
    .setDeserializer(new TransactionDeserializer())
    .build();

DataStream<Transaction> transactions = env.fromSource(
    source,
    WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(10))
        .withTimestampAssigner((tx, ts) -> tx.getTimestamp()),
    "Kafka Source"
);

// Traitement : detecter les transactions suspectes
DataStream<Alerte> alertes = transactions
    .keyBy(Transaction::getCompteId)
    .window(SlidingEventTimeWindows.of(
        Time.minutes(5),  // taille de fenetre
        Time.minutes(1)   // slide
    ))
    .process(new DetectionFraudeFunction());

// Sink : ecrire les alertes
alertes.sinkTo(
    KafkaSink.<Alerte>builder()
        .setBootstrapServers("broker1:9092")
        .setRecordSerializer(new AlerteSerializer("alertes-fraude"))
        .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        .build()
);

env.execute("Fraud Detection Pipeline");

# Creer un savepoint avant une mise a jour
flink savepoint :jobId s3://checkpoints/savepoints/

# Arreter le job avec savepoint
flink stop --savepointPath s3://checkpoints/savepoints/ :jobId

# Redemarrer depuis le savepoint avec le nouveau code
flink run -s s3://checkpoints/savepoints/savepoint-abc123 \
  -c com.example.FraudDetection \
  fraud-detection-v2.jar

// CEP : detecter un pattern de fraude
// Pattern : 3 transactions > 500 EUR en moins de 10 minutes
Pattern<Transaction, ?> fraudPattern = Pattern
    .<Transaction>begin("premier")
        .where(new SimpleCondition<Transaction>() {
            public boolean filter(Transaction tx) {
                return tx.getMontant() > 500;
            }
        })
    .next("deuxieme")
        .where(new SimpleCondition<Transaction>() {
            public boolean filter(Transaction tx) {
                return tx.getMontant() > 500;
            }
        })
    .next("troisieme")
        .where(new SimpleCondition<Transaction>() {
            public boolean filter(Transaction tx) {
                return tx.getMontant() > 500;
            }
        })
    .within(Time.minutes(10));

PatternStream<Transaction> patternStream = CEP.pattern(
    transactions.keyBy(Transaction::getCompteId),
    fraudPattern
);

DataStream<Alerte> alertes = patternStream.select(
    (Map<String, List<Transaction>> pattern) -> {
        Transaction t1 = pattern.get("premier").get(0);
        Transaction t3 = pattern.get("troisieme").get(0);
        return new Alerte(t1.getCompteId(), "3 transactions > 500 EUR en 10min",
            t1.getMontant() + t3.getMontant());
    });

# Structured Streaming - Pipeline de traitement de commandes
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("Commandes Streaming") \
    .config("spark.sql.streaming.checkpointLocation", "s3://checkpoints/commandes") \
    .getOrCreate()

# Schema des commandes
schema_commande = StructType([
    StructField("commande_id", StringType()),
    StructField("client_id", StringType()),
    StructField("montant", DoubleType()),
    StructField("categorie", StringType()),
    StructField("timestamp", TimestampType())
])

# Source : lire depuis Kafka
commandes_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092") \
    .option("subscribe", "commandes") \
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .load() \
    .select(from_json(col("value").cast("string"), schema_commande).alias("data")) \
    .select("data.*")

# Transformation : CA par categorie avec fenetre de 10 minutes
ca_par_categorie = commandes_df \
    .withWatermark("timestamp", "5 minutes") \
    .groupBy(
        window("timestamp", "10 minutes", "5 minutes"),
        "categorie"
    ) \
    .agg(
        count("*").alias("nb_commandes"),
        sum("montant").alias("chiffre_affaires"),
        avg("montant").alias("panier_moyen")
    )

# Sink : ecrire dans Delta Lake
query = ca_par_categorie.writeStream \
    .format("delta") \
    .outputMode("update") \
    .option("checkpointLocation", "s3://checkpoints/ca-categorie") \
    .trigger(processingTime="30 seconds") \
    .start("s3://datalake/gold/ca_par_categorie")

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms.window import FixedWindows, SlidingWindows
from apache_beam.transforms.trigger import AfterWatermark, AfterProcessingTime, AccumulationMode

# Pipeline Beam - Analyse de logs en temps reel
options = PipelineOptions([
    '--runner=FlinkRunner',          # Peut etre DataflowRunner, SparkRunner...
    '--flink_master=localhost:8081',
    '--streaming'
])

with beam.Pipeline(options=options) as p:
    # Source : lire depuis Kafka
    logs = (p
        | 'Lire Kafka' >> beam.io.ReadFromKafka(
            consumer_config={'bootstrap.servers': 'broker1:9092'},
            topics=['application-logs']
        )
        | 'Parser JSON' >> beam.Map(parse_log_json)
        | 'Filtrer Erreurs' >> beam.Filter(lambda log: log['level'] == 'ERROR')
    )

    # Aggregation avec fenetre et trigger
    erreurs_par_service = (logs
        | 'Fenetre 5min' >> beam.WindowInto(
            FixedWindows(300),  # 5 minutes
            trigger=AfterWatermark(
                early=AfterProcessingTime(60)  # resultats partiels toutes les 60s
            ),
            accumulation_mode=AccumulationMode.DISCARDING
        )
        | 'Cle par Service' >> beam.Map(lambda log: (log['service'], 1))
        | 'Compter' >> beam.CombinePerKey(sum)
    )

    # Sinks multiples
    erreurs_par_service | 'Vers BigQuery' >> beam.io.WriteToBigQuery(
        'projet:dataset.erreurs_par_service',
        schema='service:STRING,count:INTEGER,window_start:TIMESTAMP'
    )

    (erreurs_par_service
        | 'Filtrer Alertes' >> beam.Filter(lambda kv: kv[1] > 100)
        | 'Vers PagerDuty' >> beam.ParDo(EnvoyerAlerteFn())
    )

// Exemple CQRS - Cote Command
@Service
public class CommandeCommandService {

    private final KafkaTemplate<String, DomainEvent> kafkaTemplate;

    public void creerCommande(CreerCommandeRequest request) {
        // 1. Validation des regles metier
        validerStock(request.getProduits());
        validerLimiteCredit(request.getClientId(), request.getMontant());

        // 2. Emettre l'evenement (source de verite)
        CommandeCreee event = CommandeCreee.builder()
            .commandeId(UUID.randomUUID().toString())
            .clientId(request.getClientId())
            .produits(request.getProduits())
            .montant(request.getMontant())
            .timestamp(Instant.now())
            .version(1)
            .build();

        kafkaTemplate.send("domain-events.commandes", event.getCommandeId(), event);
    }
}

// Cote Query - Projecteur qui construit la vue de lecture
@Component
public class CommandeProjecteur {

    private final CommandeReadRepository readRepo;

    @KafkaListener(topics = "domain-events.commandes", groupId = "commande-projector")
    public void onEvent(DomainEvent event) {
        if (event instanceof CommandeCreee e) {
            readRepo.save(new CommandeView(e.getCommandeId(), e.getClientId(),
                e.getMontant(), "CREEE", e.getTimestamp()));
        } else if (event instanceof CommandePayee e) {
            readRepo.updateStatut(e.getCommandeId(), "PAYEE");
        } else if (event instanceof CommandeExpediee e) {
            readRepo.updateStatut(e.getCommandeId(), "EXPEDIEE");
            readRepo.setTracking(e.getCommandeId(), e.getNumeroSuivi());
        }
    }
}

// Event Sourcing - Reconstitution de l'etat d'un compte bancaire
public class CompteBancaire {
    private String compteId;
    private BigDecimal solde = BigDecimal.ZERO;
    private String statut = "ACTIF";
    private List<DomainEvent> uncommittedEvents = new ArrayList<>();

    // Reconstituer l'etat depuis les evenements
    public static CompteBancaire fromEvents(List<DomainEvent> events) {
        CompteBancaire compte = new CompteBancaire();
        events.forEach(compte::apply);
        return compte;
    }

    // Commande : debiter le compte
    public void debiter(BigDecimal montant, String motif) {
        if (solde.compareTo(montant) < 0) {
            throw new SoldeInsuffisantException(compteId, solde, montant);
        }
        if (!"ACTIF".equals(statut)) {
            throw new CompteInactifException(compteId);
        }
        // Emettre l'evenement (pas de mutation directe !)
        emit(new CompteDebite(compteId, montant, motif, Instant.now()));
    }

    // Appliquer un evenement (mutation de l'etat)
    private void apply(DomainEvent event) {
        if (event instanceof CompteOuvert e) {
            this.compteId = e.getCompteId();
            this.statut = "ACTIF";
        } else if (event instanceof CompteCredite e) {
            this.solde = this.solde.add(e.getMontant());
        } else if (event instanceof CompteDebite e) {
            this.solde = this.solde.subtract(e.getMontant());
        } else if (event instanceof CompteFerme e) {
            this.statut = "FERME";
        }
    }

    // Snapshot pour eviter de rejouer des milliers d'evenements
    // Snapshot tous les 100 evenements
    public CompteSnapshot toSnapshot() {
        return new CompteSnapshot(compteId, solde, statut, version);
    }
}

{
  "type": "kafka",
  "spec": {
    "dataSchema": {
      "dataSource": "clickstream_events",
      "timestampSpec": {
        "column": "event_time",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": ["user_id", "page_url", "device", "country", "campaign"]
      },
      "metricsSpec": [
        {"type": "count", "name": "event_count"},
        {"type": "longSum", "name": "duration_ms", "fieldName": "duration"},
        {"type": "thetaSketch", "name": "unique_users", "fieldName": "user_id"}
      ],
      "granularitySpec": {
        "segmentGranularity": "HOUR",
        "queryGranularity": "MINUTE",
        "rollup": true
      }
    },
    "ioConfig": {
      "consumerProperties": {
        "bootstrap.servers": "kafka-broker:9092"
      },
      "topic": "clickstream",
      "useEarliestOffset": true
    }
  }
}

-- Table Kafka Engine : consomme directement depuis Kafka
CREATE TABLE clickstream_kafka (
    event_time DateTime64(3),
    user_id String,
    page_url String,
    device LowCardinality(String),
    country LowCardinality(String),
    duration_ms UInt32
) ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'kafka-broker:9092',
    kafka_topic_list = 'clickstream',
    kafka_group_name = 'clickhouse_consumer',
    kafka_format = 'JSONEachRow';

-- Table de destination : MergeTree pour les requetes analytiques
CREATE TABLE clickstream (
    event_time DateTime64(3),
    user_id String,
    page_url String,
    device LowCardinality(String),
    country LowCardinality(String),
    duration_ms UInt32
) ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(event_time)
ORDER BY (country, device, event_time)
TTL event_time + INTERVAL 90 DAY;

-- Vue materialisee : transfert automatique Kafka -> MergeTree
CREATE MATERIALIZED VIEW clickstream_mv TO clickstream AS
SELECT * FROM clickstream_kafka;

-- Requete analytique sub-seconde
SELECT
    toStartOfMinute(event_time) AS minute,
    country,
    device,
    count() AS events,
    uniqExact(user_id) AS unique_users,
    avg(duration_ms) AS avg_duration
FROM clickstream
WHERE event_time >= now() - INTERVAL 1 HOUR
GROUP BY minute, country, device
ORDER BY minute DESC;

-- Source : connexion directe a Kafka
CREATE SOURCE clickstream_source
FROM KAFKA CONNECTION kafka_conn (TOPIC 'clickstream')
FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn;

-- Vue materialisee incrementale : mise a jour en continu
CREATE MATERIALIZED VIEW dashboard_metrics AS
SELECT
    date_trunc('minute', event_time) AS minute,
    country,
    COUNT(*) AS total_events,
    COUNT(DISTINCT user_id) AS unique_users,
    AVG(duration_ms)::INT AS avg_duration
FROM clickstream_source
GROUP BY 1, 2;

-- Requete instantanee sur l'etat courant (pull)
SELECT * FROM dashboard_metrics
WHERE minute >= mz_now() - INTERVAL '1 hour';

-- Abonnement aux changements (push) via SUBSCRIBE
SUBSCRIBE TO dashboard_metrics WITH (SNAPSHOT = false);

-- Exactitude syntaxique : codes postaux francais valides
SELECT
    COUNT(*) AS total_clients,
    SUM(CASE WHEN code_postal ~ '^\d{5}$' THEN 1 ELSE 0 END) AS codes_valides,
    ROUND(
        100.0 * SUM(CASE WHEN code_postal ~ '^\d{5}$' THEN 1 ELSE 0 END)
        / COUNT(*), 2
    ) AS pct_accuracy
FROM clients;

-- Exactitude semantique : montants coherents
SELECT
    COUNT(*) AS total_commandes,
    SUM(CASE
        WHEN montant_total = (quantite * prix_unitaire * (1 - remise / 100.0))
        THEN 1 ELSE 0
    END) AS montants_exacts,
    ROUND(
        100.0 * SUM(CASE
            WHEN montant_total = (quantite * prix_unitaire * (1 - remise / 100.0))
            THEN 1 ELSE 0
        END) / COUNT(*), 2
    ) AS pct_accuracy
FROM lignes_commande;

-- Completude au niveau colonne
SELECT
    'email' AS colonne,
    COUNT(*) AS total,
    SUM(CASE WHEN email IS NOT NULL AND email != '' THEN 1 ELSE 0 END) AS non_vides,
    ROUND(100.0 * SUM(CASE WHEN email IS NOT NULL AND email != '' THEN 1 ELSE 0 END) / COUNT(*), 2)
        AS pct_completude
FROM clients
UNION ALL
SELECT
    'telephone', COUNT(*),
    SUM(CASE WHEN telephone IS NOT NULL AND telephone != '' THEN 1 ELSE 0 END),
    ROUND(100.0 * SUM(CASE WHEN telephone IS NOT NULL AND telephone != '' THEN 1 ELSE 0 END) / COUNT(*), 2)
FROM clients;

-- Completude au niveau entite : tous les magasins ont-ils remonte ?
SELECT
    m.magasin_id,
    m.nom,
    CASE WHEN v.magasin_id IS NOT NULL THEN 'PRESENT' ELSE 'MANQUANT' END AS statut
FROM magasins m
LEFT JOIN ventes_jour v ON m.magasin_id = v.magasin_id AND v.date_vente = CURRENT_DATE - 1
WHERE v.magasin_id IS NULL;

-- Coherence intra-table : dates logiques
SELECT COUNT(*) AS incoherences_dates
FROM contrats
WHERE date_fin IS NOT NULL AND date_fin < date_debut;

-- Coherence inter-tables : totaux correspondants
SELECT
    c.client_id,
    c.nb_commandes AS nb_dans_clients,
    COALESCE(cmd.nb_reel, 0) AS nb_reel_commandes,
    ABS(c.nb_commandes - COALESCE(cmd.nb_reel, 0)) AS ecart
FROM clients c
LEFT JOIN (
    SELECT client_id, COUNT(*) AS nb_reel
    FROM commandes
    GROUP BY client_id
) cmd ON c.client_id = cmd.client_id
WHERE c.nb_commandes != COALESCE(cmd.nb_reel, 0);

-- Fraicheur : age des donnees les plus recentes par source
SELECT
    source_system,
    MAX(updated_at) AS derniere_mise_a_jour,
    EXTRACT(EPOCH FROM (NOW() - MAX(updated_at))) / 3600 AS age_heures,
    CASE
        WHEN EXTRACT(EPOCH FROM (NOW() - MAX(updated_at))) / 3600 < 1 THEN 'FRESH'
        WHEN EXTRACT(EPOCH FROM (NOW() - MAX(updated_at))) / 3600 < 24 THEN 'ACCEPTABLE'
        ELSE 'STALE'
    END AS statut_fraicheur
FROM raw_data
GROUP BY source_system
ORDER BY age_heures DESC;

-- Validite : champs respectant les regles metier
SELECT
    'email_format' AS regle,
    SUM(CASE WHEN email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$' THEN 1 ELSE 0 END) AS valides,
    COUNT(*) AS total,
    ROUND(100.0 * SUM(CASE WHEN email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$' THEN 1 ELSE 0 END) / COUNT(*), 2) AS pct
FROM clients
UNION ALL
SELECT
    'age_range',
    SUM(CASE WHEN age BETWEEN 0 AND 150 THEN 1 ELSE 0 END),
    COUNT(*),
    ROUND(100.0 * SUM(CASE WHEN age BETWEEN 0 AND 150 THEN 1 ELSE 0 END) / COUNT(*), 2)
FROM clients
UNION ALL
SELECT
    'statut_enum',
    SUM(CASE WHEN statut IN ('ACTIF', 'INACTIF', 'SUSPENDU', 'FERME') THEN 1 ELSE 0 END),
    COUNT(*),
    ROUND(100.0 * SUM(CASE WHEN statut IN ('ACTIF', 'INACTIF', 'SUSPENDU', 'FERME') THEN 1 ELSE 0 END) / COUNT(*), 2)
FROM clients;

-- Detection de doublons exacts
SELECT
    email, COUNT(*) AS nb_occurrences
FROM clients
GROUP BY email
HAVING COUNT(*) > 1
ORDER BY nb_occurrences DESC;

-- Taux de doublons global
SELECT
    COUNT(*) AS total_lignes,
    COUNT(DISTINCT client_id) AS clients_uniques,
    COUNT(*) - COUNT(DISTINCT client_id) AS doublons,
    ROUND(100.0 * (COUNT(*) - COUNT(DISTINCT client_id)) / COUNT(*), 2) AS pct_doublons
FROM clients;

-- Doublons fuzzy : noms similaires (distance de Levenshtein)
SELECT
    a.client_id AS id_a,
    b.client_id AS id_b,
    a.nom, b.nom,
    a.email, b.email,
    levenshtein(LOWER(a.nom), LOWER(b.nom)) AS distance_nom
FROM clients a
JOIN clients b ON a.client_id < b.client_id
WHERE levenshtein(LOWER(a.nom), LOWER(b.nom)) <= 2
   AND (a.email = b.email OR a.telephone = b.telephone);

import great_expectations as gx

# Initialiser le contexte GE
context = gx.get_context()

# Connecter une source de donnees
datasource = context.sources.add_or_update_pandas(name="source_ventes")
data_asset = datasource.add_csv_asset(
    name="ventes_quotidiennes",
    filepath_or_buffer="data/ventes_2024.csv"
)
batch_request = data_asset.build_batch_request()

# Creer une Expectation Suite
suite = context.add_or_update_expectation_suite("ventes_quality_suite")

# Obtenir un validator pour definir les expectations
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="ventes_quality_suite"
)

# Definir les expectations
# 1. Completude : pas de nulls sur les champs critiques
validator.expect_column_values_to_not_be_null("commande_id")
validator.expect_column_values_to_not_be_null("montant")
validator.expect_column_values_to_not_be_null("date_vente")

# 2. Validite : types et plages
validator.expect_column_values_to_be_between(
    "montant", min_value=0.01, max_value=100000,
    mostly=0.99  # tolerer 1% d'anomalies
)
validator.expect_column_values_to_be_in_set(
    "statut", ["CONFIRMEE", "EXPEDIEE", "LIVREE", "ANNULEE"]
)
validator.expect_column_values_to_match_regex(
    "email", r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
)

# 3. Unicite
validator.expect_column_values_to_be_unique("commande_id")

# 4. Coherence
validator.expect_column_pair_values_a_to_be_greater_than_b(
    "date_livraison", "date_vente"
)

# 5. Volume : nombre de lignes attendu
validator.expect_table_row_count_to_be_between(
    min_value=1000, max_value=50000
)

# Sauvegarder la suite
validator.save_expectation_suite(discard_failed_expectations=False)

# Creer un checkpoint pour orchestrer la validation
checkpoint = context.add_or_update_checkpoint(
    name="ventes_daily_checkpoint",
    validations=[
        {
            "batch_request": batch_request,
            "expectation_suite_name": "ventes_quality_suite",
        }
    ],
    action_list=[
        # Stocker le resultat
        {
            "name": "store_validation_result",
            "action": {"class_name": "StoreValidationResultAction"}
        },
        # Mettre a jour les Data Docs
        {
            "name": "update_data_docs",
            "action": {"class_name": "UpdateDataDocsAction"}
        },
        # Envoyer une alerte Slack en cas d'echec
        {
            "name": "send_slack_notification",
            "action": {
                "class_name": "SlackNotificationAction",
                "slack_webhook": "${SLACK_WEBHOOK_URL}",
                "notify_on": "failure",
                "renderer": {
                    "class_name": "SlackRenderer"
                }
            }
        }
    ]
)

# Executer le checkpoint
result = checkpoint.run()

# Verifier le resultat
if not result.success:
    print("ECHEC de la validation !")
    for r in result.run_results.values():
        for expectation_result in r["validation_result"].results:
            if not expectation_result.success:
                print(f"  FAILED: {expectation_result.expectation_config.expectation_type}")
                print(f"  Details: {expectation_result.result}")
else:
    print("Validation reussie - pipeline peut continuer")

# checks/ventes_checks.yml
checks for ventes:
  # Completude
  - missing_count(commande_id) = 0:
      name: "Commande ID ne doit pas etre null"
  - missing_percent(email) < 5:
      name: "Email manquant tolere a 5%"

  # Volume
  - row_count between 1000 and 50000:
      name: "Volume de ventes quotidien attendu"
  - row_count > 0:
      name: "La table ne doit pas etre vide"

  # Fraicheur
  - freshness(date_vente) < 1d:
      name: "Les donnees doivent avoir moins de 24h"

  # Validite
  - invalid_percent(montant) < 1:
      valid min: 0.01
      valid max: 100000
      name: "Montants dans la plage valide"
  - invalid_count(statut) = 0:
      valid values: ["CONFIRMEE", "EXPEDIEE", "LIVREE", "ANNULEE"]

  # Unicite
  - duplicate_count(commande_id) = 0:
      name: "Pas de doublons sur commande_id"

  # Coherence inter-tables
  - values in (client_id) must exist in clients (client_id):
      name: "Integrite referentielle clients"

  # Custom SQL
  - failed rows:
      fail condition: date_livraison < date_vente
      name: "Date livraison doit etre apres date vente"

  # Schema
  - schema:
      fail:
        when required column missing:
          [commande_id, client_id, montant, date_vente, statut]
        when wrong type:
          montant: decimal
          date_vente: date

# configuration.yml
data_source ma_base:
  type: postgres
  host: ${POSTGRES_HOST}
  port: 5432
  username: ${POSTGRES_USER}
  password: ${POSTGRES_PASSWORD}
  database: production
  schema: public

soda_cloud:
  host: cloud.soda.io
  api_key_id: ${SODA_API_KEY_ID}
  api_key_secret: ${SODA_API_KEY_SECRET}

# Installation
pip install soda-core-postgres

# Executer les checks
soda scan -d ma_base -c configuration.yml checks/ventes_checks.yml

# Resultat typique :
# Scan summary:
# 10/10 checks PASSED
# 0 checks WARNED
# 0 checks FAILED
# All is good. No failures. No warnings. No errors.

-- Monitoring de fraicheur pour toutes les tables critiques
WITH freshness_check AS (
    SELECT
        'commandes' AS table_name,
        MAX(updated_at) AS last_update,
        INTERVAL '1 hour' AS sla_interval
    FROM commandes
    UNION ALL
    SELECT 'paiements', MAX(updated_at), INTERVAL '2 hours'
    FROM paiements
    UNION ALL
    SELECT 'inventory', MAX(updated_at), INTERVAL '30 minutes'
    FROM inventory
)
SELECT
    table_name,
    last_update,
    NOW() - last_update AS age,
    sla_interval,
    CASE
        WHEN NOW() - last_update > sla_interval THEN 'BREACH'
        WHEN NOW() - last_update > sla_interval * 0.8 THEN 'WARNING'
        ELSE 'OK'
    END AS status
FROM freshness_check
ORDER BY
    CASE WHEN NOW() - last_update > sla_interval THEN 0 ELSE 1 END,
    age DESC;

import pandas as pd
from scipy import stats

def detect_volume_anomaly(table_name, connection):
    """Detecter les anomalies de volume avec Z-score."""
    # Historique des 30 derniers jours
    query = f"""
        SELECT date_trunc('day', created_at) AS jour, COUNT(*) AS volume
        FROM {table_name}
        WHERE created_at >= NOW() - INTERVAL '30 days'
        GROUP BY 1
        ORDER BY 1
    """
    df = pd.read_sql(query, connection)

    if len(df) < 7:
        return {"status": "INSUFFICIENT_DATA"}

    # Calcul du Z-score pour le jour le plus recent
    volumes = df['volume'].values
    current = volumes[-1]
    historical = volumes[:-1]

    z_score = (current - historical.mean()) / historical.std()

    status = "OK"
    if abs(z_score) > 3:
        status = "CRITICAL"  # anomalie forte
    elif abs(z_score) > 2:
        status = "WARNING"   # anomalie moderee

    return {
        "table": table_name,
        "current_volume": int(current),
        "expected_range": f"{int(historical.mean() - 2*historical.std())} - "
                         f"{int(historical.mean() + 2*historical.std())}",
        "z_score": round(z_score, 2),
        "status": status
    }

def detect_schema_changes(table_name, connection, expected_schema):
    """Comparer le schema actuel avec le schema attendu."""
    query = f"""
        SELECT column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_name = '{table_name}'
        ORDER BY ordinal_position
    """
    current_schema = pd.read_sql(query, connection)
    current_cols = set(current_schema['column_name'])
    expected_cols = set(expected_schema.keys())

    changes = []

    # Colonnes ajoutees
    for col in current_cols - expected_cols:
        changes.append({"type": "ADDED", "column": col, "severity": "WARNING"})

    # Colonnes supprimees
    for col in expected_cols - current_cols:
        changes.append({"type": "REMOVED", "column": col, "severity": "CRITICAL"})

    # Changements de type
    for _, row in current_schema.iterrows():
        col = row['column_name']
        if col in expected_schema:
            if row['data_type'] != expected_schema[col]['type']:
                changes.append({
                    "type": "TYPE_CHANGED",
                    "column": col,
                    "from": expected_schema[col]['type'],
                    "to": row['data_type'],
                    "severity": "CRITICAL"
                })

    return changes

# dataops-pipeline.yml
name: DataOps Pipeline
triggers:
  - schedule: "0 */2 * * *"   # Toutes les 2 heures
  - on_push:
      branches: [main, develop]
  - on_data_change:
      sources: [raw_layer]

stages:
  validate:
    - data_quality_checks:
        tool: great_expectations
        suite: production_suite
    - schema_validation:
        tool: soda
        checks: schema_changes

  transform:
    - dbt_run:
        target: production
        select: "tag:incremental"
        fail_fast: true

  test:
    - dbt_test:
        severity: warn
    - custom_tests:
        script: tests/business_rules.py

  deploy:
    - blue_green_swap:
        strategy: zero_downtime
    - notify:
        channels: [slack, email]
        on: [success, failure]

  monitor:
    - freshness_check:
        max_delay: 30m
    - anomaly_detection:
        model: isolation_forest
        sensitivity: 0.95

name: dbt CI/CD Pipeline

on:
  pull_request:
    branches: [main]
    paths:
      - 'models/**'
      - 'macros/**'
      - 'tests/**'
      - 'dbt_project.yml'
      - 'packages.yml'

env:
  DBT_PROFILES_DIR: ./ci
  DBT_TARGET: ci
  SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
  SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
  SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

jobs:
  lint:
    name: "Lint & Format Check"
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install sqlfluff dbt-core dbt-snowflake
      - name: SQLFluff Lint
        run: sqlfluff lint models/ --dialect snowflake

  slim-ci:
    name: "Slim CI - Modified Models Only"
    needs: lint
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install dbt-core dbt-snowflake

      - name: Get production manifest
        run: |
          aws s3 cp s3://dbt-artifacts/prod/manifest.json \
            ./target/manifest.json

      - name: dbt deps
        run: dbt deps

      - name: dbt build (modified only)
        run: |
          dbt build \
            --select state:modified+ \
            --defer \
            --state ./target \
            --target ci \
            --fail-fast

      - name: Generate docs
        run: dbt docs generate

      - name: Comment PR with results
        uses: actions/github-script@v7
        with:
          script: |
            const results = require('./target/run_results.json');
            const summary = results.results.map(r =>
              `| ${r.unique_id} | ${r.status} | ${r.execution_time}s |`
            ).join('\n');
            github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: `## dbt CI Results\n| Model | Status | Time |\n|---|---|---|\n${summary}`
            });

repos:
  - repo: https://github.com/sqlfluff/sqlfluff
    rev: 3.0.0
    hooks:
      - id: sqlfluff-lint
        args: [--dialect, snowflake]
        files: ^models/
      - id: sqlfluff-fix
        args: [--dialect, snowflake, --force]
        files: ^models/

  - repo: https://github.com/dbt-checkpoint/dbt-checkpoint
    rev: 2.0.0
    hooks:
      - id: check-model-has-description
      - id: check-model-has-tests
        args: [--test-cnt, "2"]
      - id: check-model-has-properties-file
      - id: check-model-name-contract
        args: [--pattern, "(staging|intermediate|marts)_.*"]
      - id: check-source-has-freshness
      - id: check-script-semicolon

  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.6.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-json

data_warehouse:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      database: ANALYTICS_DEV
      schema: "dbt_{{ env_var('USER', 'default') }}"
      warehouse: DEV_WH_XS
      threads: 4

    ci:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      database: ANALYTICS_CI
      schema: "pr_{{ env_var('PR_NUMBER', '0') }}"
      warehouse: CI_WH_S
      threads: 8

    staging:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      database: ANALYTICS_STAGING
      schema: public
      warehouse: STAGING_WH_M
      threads: 16

    prod:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      database: ANALYTICS_PROD
      schema: public
      warehouse: PROD_WH_L
      threads: 32

-- Etape 1: Build dans le schema "green" (secondaire)
-- dbt build --target prod --vars '{schema_suffix: _green}'

-- Etape 2: Verification de sante (health check)
SELECT
    'stg_orders' AS model_name,
    COUNT(*) AS row_count,
    MAX(updated_at) AS max_updated
FROM analytics_prod.public_green.stg_orders

UNION ALL

SELECT
    'fct_revenue' AS model_name,
    COUNT(*) AS row_count,
    MAX(updated_at) AS max_updated
FROM analytics_prod.public_green.fct_revenue;

-- Etape 3: Swap atomique des schemas
ALTER SCHEMA analytics_prod.public
  RENAME TO analytics_prod.public_old;
ALTER SCHEMA analytics_prod.public_green
  RENAME TO analytics_prod.public;

-- Etape 4: Cleanup (apres validation)
DROP SCHEMA IF EXISTS analytics_prod.public_old CASCADE;

unit_tests:
  - name: test_calcul_revenue_net
    description: "Verifie le calcul du revenu net avec remises et taxes"
    model: fct_revenue
    given:
      - input: ref('stg_orders')
        rows:
          - { order_id: 1, amount: 100.00, discount: 10.00, tax_rate: 0.20 }
          - { order_id: 2, amount: 200.00, discount: 0.00, tax_rate: 0.20 }
          - { order_id: 3, amount: 50.00, discount: 50.00, tax_rate: 0.20 }
      - input: ref('stg_refunds')
        rows:
          - { order_id: 3, refund_amount: 0.00 }
    expect:
      rows:
        - { order_id: 1, net_revenue: 108.00 }  # (100-10) * 1.20
        - { order_id: 2, net_revenue: 240.00 }  # (200-0) * 1.20
        - { order_id: 3, net_revenue: 0.00 }    # (50-50) * 1.20

  - name: test_classification_client
    description: "Verifie la segmentation client par montant d'achat"
    model: dim_customers
    given:
      - input: ref('stg_customers')
        rows:
          - { customer_id: 1, total_purchases: 15000 }
          - { customer_id: 2, total_purchases: 5000 }
          - { customer_id: 3, total_purchases: 500 }
    expect:
      rows:
        - { customer_id: 1, segment: 'premium' }
        - { customer_id: 2, segment: 'standard' }
        - { customer_id: 3, segment: 'basic' }

models:
  - name: fct_orders
    description: "Table de faits des commandes"
    tests:
      - dbt_utils.fewer_rows_than:
          compare_model: ref('stg_orders')
          # Apres deduplication, fct doit avoir moins de lignes
      - dbt_utils.expression_is_true:
          expression: "total_amount >= 0"
          config:
            severity: error

    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: order_date
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: "'2020-01-01'"
              max_value: "current_date()"
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']

models:
  - name: fct_revenue
    config:
      contract:
        enforced: true
    columns:
      - name: revenue_id
        data_type: varchar(36)
        constraints:
          - type: not_null
          - type: primary_key
      - name: customer_id
        data_type: varchar(36)
        constraints:
          - type: not_null
          - type: foreign_key
            to: ref('dim_customers')
            to_columns: [customer_id]
      - name: revenue_date
        data_type: date
        constraints:
          - type: not_null
      - name: net_amount
        data_type: number(18,2)
        constraints:
          - type: not_null
          - type: check
            expression: "net_amount >= 0"

# macros/test_regression_revenue.sql
{% test regression_revenue(model, compare_model) %}

  {% set audit_query = dbt_utils.audit_helper.compare_relations(
      a_relation=model,
      b_relation=compare_model,
      primary_key="revenue_id",
      columns=["customer_id", "revenue_date", "net_amount"]
  ) %}

  WITH comparison AS (
      {{ audit_query }}
  )

  SELECT *
  FROM comparison
  WHERE in_a != in_b
    -- Tolerance de 1% sur les montants
    AND ABS(a_net_amount - b_net_amount) / NULLIF(a_net_amount, 0) > 0.01

{% endtest %}

# Comparaison entre staging et production
# pip install data-diff

from data_diff import connect, diff_tables

db_prod = connect("snowflake://user:pass@account/ANALYTICS_PROD/PUBLIC")
db_staging = connect("snowflake://user:pass@account/ANALYTICS_STAGING/PUBLIC")

diff_result = diff_tables(
    table1=db_prod.get_table("fct_revenue"),
    table2=db_staging.get_table("fct_revenue"),
    key_columns=["revenue_id"],
    extra_columns=["net_amount", "customer_id"],
    where="revenue_date >= '2024-01-01'"
)

for row_diff in diff_result:
    print(f"Difference trouvee: {row_diff}")

# Resultat:
# + row in staging only: revenue_id=X, net_amount=150.00
# ~ changed: revenue_id=Y, net_amount 100->120
# - row in prod only: revenue_id=Z (supprime en staging)

# Initialiser DVC dans un projet Git existant
git init && dvc init

# Configurer le stockage distant (S3)
dvc remote add -d myremote s3://my-bucket/dvc-store
dvc remote modify myremote region eu-west-1

# Tracker un dataset
dvc add data/raw/transactions_2024.parquet
# Cree: data/raw/transactions_2024.parquet.dvc (metadonnees)
# Ajoute au .gitignore: data/raw/transactions_2024.parquet

# Commiter les metadonnees dans Git
git add data/raw/transactions_2024.parquet.dvc data/raw/.gitignore
git commit -m "feat: add transactions dataset v1"

# Pousser les donnees vers le stockage distant
dvc push

# --- Plus tard: mettre a jour le dataset ---
# (apres modification du fichier parquet)
dvc add data/raw/transactions_2024.parquet
git add data/raw/transactions_2024.parquet.dvc
git commit -m "feat: update transactions dataset v2"
dvc push

# --- Rollback vers la version precedente ---
git checkout HEAD~1 -- data/raw/transactions_2024.parquet.dvc
dvc checkout
# Les donnees v1 sont restaurees !

stages:
  prepare:
    cmd: python src/prepare.py
    deps:
      - data/raw/transactions_2024.parquet
      - src/prepare.py
    params:
      - prepare.min_amount
      - prepare.date_range
    outs:
      - data/prepared/transactions_clean.parquet

  features:
    cmd: python src/featurize.py
    deps:
      - data/prepared/transactions_clean.parquet
      - src/featurize.py
    params:
      - features.window_size
      - features.aggregations
    outs:
      - data/features/feature_store.parquet

  train:
    cmd: python src/train.py
    deps:
      - data/features/feature_store.parquet
      - src/train.py
    params:
      - train.model_type
      - train.learning_rate
    outs:
      - models/fraud_detector.pkl
    metrics:
      - metrics/scores.json:
          cache: false

import lakefs_sdk
from lakefs_sdk.client import LakeFSClient

client = LakeFSClient(
    configuration=lakefs_sdk.Configuration(
        host="http://localhost:8000/api/v1",
        username="AKIAIOSFODNN7EXAMPLE",
        password="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
    )
)

# Creer un repository
client.repositories_api.create_repository(
    repository_creation={
        "name": "analytics-lake",
        "storage_namespace": "s3://my-bucket/analytics/",
        "default_branch": "main"
    }
)

# Creer une branche pour une experimentation
client.branches_api.create_branch(
    repository="analytics-lake",
    branch_creation={
        "name": "experiment/new-fraud-features",
        "source": "main"
    }
)

# Uploader des donnees sur la branche
with open("data/fraud_features_v2.parquet", "rb") as f:
    client.objects_api.upload_object(
        repository="analytics-lake",
        branch="experiment/new-fraud-features",
        path="features/fraud_features.parquet",
        content=f
    )

# Commiter
client.commits_api.commit(
    repository="analytics-lake",
    branch="experiment/new-fraud-features",
    commit_creation={
        "message": "Add new fraud detection features v2",
        "metadata": {
            "author": "data-team",
            "ticket": "DATA-1234"
        }
    }
)

# Merger vers main apres validation
client.refs_api.merge_into_branch(
    repository="analytics-lake",
    source_ref="experiment/new-fraud-features",
    destination_branch="main"
)

-- Creer une branche pour les modifications
CREATE BRANCH experiment_branch IN nessie FROM main;

-- Travailler sur la branche
USE REFERENCE experiment_branch IN nessie;

-- Modifier les tables sur la branche
ALTER TABLE analytics.fct_revenue
  ADD COLUMN new_metric DOUBLE;

INSERT INTO analytics.fct_revenue
SELECT *, calculate_new_metric(*)
FROM analytics.fct_revenue@main;  -- Reference croisee

-- Valider les resultats
SELECT COUNT(*), AVG(new_metric)
FROM analytics.fct_revenue;

-- Merger vers main
MERGE BRANCH experiment_branch INTO main IN nessie;

-- Snowflake Time Travel (jusqu'a 90 jours)
SELECT * FROM fct_revenue
AT(TIMESTAMP => '2024-03-15 10:00:00'::TIMESTAMP);

-- Comparer deux versions
SELECT
    COALESCE(a.revenue_id, b.revenue_id) AS revenue_id,
    a.net_amount AS amount_before,
    b.net_amount AS amount_after,
    b.net_amount - a.net_amount AS delta
FROM fct_revenue AT(OFFSET => -3600) a  -- Il y a 1 heure
FULL OUTER JOIN fct_revenue b
    ON a.revenue_id = b.revenue_id
WHERE a.net_amount != b.net_amount
   OR a.revenue_id IS NULL
   OR b.revenue_id IS NULL;

-- Delta Lake Time Travel (versions illimitees)
SELECT * FROM delta.`/data/fct_revenue`
VERSION AS OF 42;

SELECT * FROM delta.`/data/fct_revenue`
TIMESTAMP AS OF '2024-03-15 10:00:00';

-- Apache Iceberg Time Travel
SELECT * FROM analytics.fct_revenue
FOR SYSTEM_TIME AS OF TIMESTAMP '2024-03-15 10:00:00';

-- Iceberg: lister les snapshots disponibles
SELECT * FROM analytics.fct_revenue.snapshots;