Remplissage des données historiques avec des pipelines

Dans l’ingénierie des données, le remplissage fait référence au processus de traitement rétroactif des données historiques par le biais d’un pipeline de données conçu pour traiter les données actuelles ou de diffusion en continu.

En règle générale, il s’agit d’un flux distinct qui envoie des données dans vos tables existantes. L’illustration suivante présente un flux de renvoi de données historiques vers les tables bronze de votre pipeline.

Certains scénarios qui peuvent nécessiter un remblai :

• Traitez les données historiques d’un système hérité pour entraîner un modèle Machine Learning (ML) ou créer un tableau de bord d’analyse de tendance historique.
• Retraitez un sous-ensemble de données en raison d’un problème de qualité des données liées aux sources de données en amont.
• Vos besoins métier ont changé et vous devez renvoyer des données pour une période différente qui n’a pas été couverte par le pipeline initial.
• Votre logique métier a changé et vous devez retraiter les données historiques et actuelles.

Le flux de rétroremplissage que vous utilisez dépend de la table cible et des données sources. Pour une cible de AUTO CDC dimension à variation lente (SCD) de type 1 avec un instantané faisant autorité, utilisez un flux AUTO CDC FROM SNAPSHOT à exécution unique. Pour une migration SCD qui rejoue des changements historiques, utilisez un flux AUTO CDC à usage unique.

Pour les remplissages rétroactifs par ajout uniquement : utilisez un flux d'ajout spécialisé avec l'option ONCE pour effectuer un remplissage rétroactif d'une table de streaming par ajout uniquement. Pour plus d’informations sur l’option, consultez append_flow ou ONCE

Considérations relatives au remplissage des données historiques dans une table de flux

• En règle générale, ajoutez les données à la table de streaming bronze. Les couches Silver et Gold en aval récupèrent les nouvelles données de la couche Bronze.
• Vérifiez que votre pipeline peut gérer correctement les données dupliquées au cas où les mêmes données sont ajoutées plusieurs fois.
• Vérifiez que le schéma de données historiques est compatible avec le schéma de données actuel.
• Considérez la taille du volume de données et l’accord de niveau de service (SLA) requis, et configurez en conséquence les tailles du cluster et du lot.

Exemple : ajouter un backfill à un pipeline existant

Dans cet exemple, supposons que vous disposez d’un pipeline qui ingère des données brutes d’inscription d’événements à partir d’une source de stockage cloud, à compter du 01 janvier 2025. Vous réalisez par la suite que vous souhaitez effectuer un renvoi des données historiques des trois années précédentes pour des cas d'usage de création de rapports et d'analyse en aval. Toutes les données se trouvent à un emplacement, partitionnés par année, mois et jour, au format JSON.

Pipeline initiale

Voici le code de pipeline de démarrage qui ingère de façon incrémentielle les données d’inscription d’événements brutes à partir du stockage cloud.

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
incremental_load_path = f"{source_root_path}/*/*/*"

# create a streaming table and the default flow to ingest streaming events
@dp.table(name="registration_events_raw", comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2025-01-01T00:00:00.000+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}") # safeguard to not process data before begin_year
    )

SQL

-- create a streaming table and the default flow to ingest streaming events
CREATE OR REFRESH STREAMING LIVE TABLE registration_events_raw AS
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025'; -- safeguard to not process data before begin_year

Ici, nous utilisons l’option modifiedAfter Chargeur automatique pour nous assurer que nous ne traitons pas toutes les données à partir du chemin de stockage cloud. Le traitement incrémentiel est arrêté à cette frontière.

Remplissage des données des 3 dernières années

Vous souhaitez maintenant ajouter un ou plusieurs flux pour recharger les données précédentes. Dans cet exemple, procédez comme suit :

• Utilisez le append once flux. Cela effectue un renvoi unique sans continuer à s’exécuter après ce premier renvoi. Le code reste dans votre pipeline et, si le pipeline est un jour entièrement actualisé, le renvoi est réexécuté.
• Créez trois flux de renvoi, un pour chaque année (dans ce cas, les données sont divisées par année dans le chemin). Pour Python, nous paramétrons la création des flux, mais dans SQL, nous répétons le code trois fois, une fois pour chaque flux.

Si vous travaillez sur votre propre projet et que vous n’utilisez pas de calcul sans serveur, vous pouvez ajuster le nombre maximum de travailleurs pour le pipeline. En augmentant le nombre maximal de workers, vous vous assurez de disposer des ressources nécessaires pour traiter les données historiques tout en continuant à traiter les données en streaming actuelles en respectant le SLA.

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
backfill_years = spark.conf.get("backfill_years") # e.g. "2024,2023,2022"
incremental_load_path = f"{source_root_path}/*/*/*"

# meta programming to create append once flow for a given year (called later)
def setup_backfill_flow(year):
    backfill_path = f"{source_root_path}/year={year}/*/*"
    @dp.append_flow(
        target="registration_events_raw",
        once=True,
        name=f"flow_registration_events_raw_backfill_{year}",
        comment=f"Backfill {year} Raw registration events")
    def backfill():
        return (
            spark
            .read
            .format("json")
            .option("inferSchema", "true")
            .load(backfill_path)
        )

# create the streaming table
dp.create_streaming_table(name="registration_events_raw", comment="Raw registration events")

# append the original incremental, streaming flow
@dp.append_flow(
        target="registration_events_raw",
        name="flow_registration_events_raw_incremental",
        comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2024-12-31T23:59:59.999+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}")
    )

# parallelize one time multi years backfill for faster processing
# split backfill_years into array
for year in backfill_years.split(","):
    setup_backfill_flow(year) # call the previously defined append_flow for each year

SQL

-- create the streaming table
CREATE OR REFRESH STREAMING TABLE registration_events_raw;

-- append the original incremental, streaming flow
CREATE FLOW
  registration_events_raw_incremental
AS INSERT INTO
  registration_events_raw BY NAME
SELECT * FROM STREAM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025';


-- one time backfill 2024
CREATE FLOW
  registration_events_raw_backfill_2024
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2024/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2023
CREATE FLOW
  registration_events_raw_backfill_2023
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2023/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2022
CREATE FLOW
  registration_events_raw_backfill_2022
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2022/*/*",
  format => "json",
  inferColumnTypes => true
);

Cette implémentation met en évidence plusieurs modèles importants.

Séparation des responsabilités

• Le traitement incrémentiel est indépendant des opérations de remplissage.
• Chaque flux a ses propres paramètres de configuration et d’optimisation.
• Il existe une distinction claire entre les opérations incrémentielles et de remplissage.

Exécution contrôlée

• L’utilisation de l’option ONCE garantit que chaque remplissage s’exécute exactement une fois.
• Le flux de renvoi reste dans le graphique de pipeline, mais devient inactif une fois terminé. Il est prêt à être utilisé lors de l’actualisation complète, automatiquement.
• Il existe une piste d’audit claire des opérations de renvoi dans la définition du pipeline.

Optimisation du traitement

• Vous pouvez diviser le large lot de données à remplir en plusieurs lots plus petits pour un traitement plus rapide ou pour mieux contrôler le processus.
• L'utilisation de l'autoscaling amélioré adapte dynamiquement la taille du cluster en fonction de sa charge actuelle.

Évolution du schéma

• L'utilisation de schemaEvolutionMode="addNewColumns" gère les changements de schéma sans problème.
• Vous disposez d’une inférence de schéma cohérente entre les données historiques et actuelles.
• Il existe une gestion sécurisée des nouvelles colonnes dans les données plus récentes.

Ajoutez un backfill à une table AUTO CDC SCD Type 1

Utilisez un flux AUTO CDC FROM SNAPSHOT à exécution unique pour ajouter un instantané faisant autorité à une cible SCD de type 1 qui reçoit également un flux continu de capture des changements de données (CDC). La version de l'instantané et la colonne de séquencement CDC appartiennent au même domaine d'ordonnancement. Un nouvel événement CDC prend le pas sur un ancien instantané, tandis qu’un nouvel instantané prime sur un ancien événement CDC.

Requirements

Avant d’ajouter le remblai, assurez-vous que les débits respectent les exigences suivantes :

• La cible utilise le SCD Type 1.
• La cible possède exactement un AUTO CDC FROM SNAPSHOT flux et un ou plusieurs AUTO CDC flux nommés de manière unique.
• Tous les flux utilisent le même nombre de clés, dans le même ordre. Les noms des clés du flux d'instantané sont comparés sans tenir compte de la casse aux noms des clés AUTO CDC. Les flux multiples AUTO CDC doivent utiliser des noms de clés et une casse identiques.
• La version de snapshot et chaque colonne de séquençage CDC ont exactement le même type de données.
• Le AUTO CDC FROM SNAPSHOT flow ne définit pas les attentes.
• Les AUTO CDC flux n’utilisent pas IGNORE NULL UPDATES. En Python, ne définissez ignore_null_updatespas , ignore_null_updates_column_list, ni ignore_null_updates_except_column_list.
• Le pipeline utilise le mode déclenché. Ce modèle ne prend pas en charge les pipelines continus.

Les deux types de flux peuvent utiliser l’interface de pipeline SQL ou Python. Vous pouvez combiner des flux SQL et Python dans une même cible.

L'instantané doit représenter l'état complet de la source à la version correspondante. Si une clé cible est absente de l’instantané, AUTO CDC FROM SNAPSHOT considère cette absence comme une suppression à la version de l’instantané. Un événement CDC avec une version plus récente préserve ou restaure la clé.

Ajoutez le backfill

Pour ajouter un backfill d’instantané ponctuel et continuer à traiter les événements CDC, utilisez les étapes suivantes :

1. Conservez la table cible existante et ses flux continus AUTO CDC dans la définition du pipeline.
2. Définissez l’instantané de référence et sa version. Pour un callback Python, la première invocation doit retourner un instantané et une version. Renvoyez None seulement après qu’au moins un instantané a été traité.
3. Ajoutez un AUTO CDC FROM SNAPSHOT flux avec once=True en Python ou ONCE en SQL. Pour un renvoi SQL dans une cible existante, incluez une requête WITH VERSION. Un flux d’instantané SQL sans WITH VERSION ne supporte qu’un chargement initial sur une cible vide.
4. Effectuez une mise à jour déclenchée du pipeline pour traiter le rattrapage et les événements CDC en cours.

L’exemple suivant commence avec un pipeline Python existant qui traite de manière incrémentale les changements issus de customers_cdc. Supposons que vous ayez déjà exécuté ce pipeline et alimenté la cible customers :

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def customers_cdc():
    return (
        spark.readStream.table("main.bronze.customers_cdc")
        .withColumn("change_timestamp", col("change_timestamp").cast("timestamp"))
    )


dp.create_streaming_table("customers")

dp.create_auto_cdc_flow(
    name="customers_incremental_cdc",
    target="customers",
    source="customers_cdc",
    keys=["customer_id"],
    sequence_by=col("change_timestamp"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "change_timestamp"],
    stored_as_scd_type=1,
)

Pour alimenter rétroactivement cette cible existante avec l’état de customers_snapshot au 1er janvier 2025, ajoutez le code suivant à la même définition de pipeline. Conservez la table cible et AUTO CDC le flux existants :

from datetime import datetime, timezone
from typing import Optional, Tuple

from pyspark.sql import DataFrame

backfill_version = datetime(2025, 1, 1, tzinfo=timezone.utc)


def backfill_snapshot_and_version(
    latest_snapshot_version: Optional[datetime],
) -> Optional[Tuple[DataFrame, datetime]]:
    if latest_snapshot_version is None:
        return (spark.read.table("main.legacy.customers_snapshot"), backfill_version)
    return None


dp.create_auto_cdc_from_snapshot_flow(
    target="customers",
    source=backfill_snapshot_and_version,
    keys=["customer_id"],
    stored_as_scd_type=1,
    once=True,
)

Le rappel doit retourner un instantané et une version lors de sa première invocation. Si elle renvoie None avant qu’un instantané ne soit traité, la mise à jour du pipeline échoue. Après le traitement de l’instantané, le renvoi de None indique qu’aucun instantané supplémentaire n’est disponible.

La version snapshot est un datetime Python, qui correspond au type Spark SQL TIMESTAMP. Le flux existant AUTO CDC convertit change_timestamp en TIMESTAMP de façon à ce que les deux types de séquençage correspondent exactement. L’exemple utilise Python pour les deux flows, mais vous pouvez définir l’un ou l’autre flow en SQL et mélanger SQL et Python flows dans la même destination. Pour la syntaxe SQL, y compris la requête requise WITH VERSION pour une cible non vide, voir CREATE FLOW (pipelines).

Après que le flux d’instantané a réussi son commit, les mises à jour incrémentales suivantes le sautent tandis que le AUTO CDC flux continue de traiter de nouveaux événements.

Ce motif de remplissage unifié ne prend pas en charge les cibles SCD Type 2 ni bitemporelles.

Exemple : Remblai une cible SCD lors d’une migration

Un scénario de migration courant est une table de dimension à changement lent (SCD) qui existe déjà dans un système hérité avec des années d’histoire accumulée, mais dont le flux de modifications original n’est plus disponible. Comme les événements de changement originaux ont disparu, vous rejouez à la place l’historique de la table héritée dans la nouvelle cible AUTO CDC, en une seule fois, puis vous attachez un nouveau flux CDC pour la suite. Pour en savoir plus sur AUTO CDC et les types de SCD, voir Les API AUTO CDC : Simplifier la capture des données de modifications avec pipelines.

Le schéma consiste en un flux ponctuel AUTO CDC vers la même table de streaming que celle ciblée par le flux continu AUTO CDC. Une cible AUTO CDC n’accepte que les flux AUTO CDC ; la graine doit donc aussi être un flux AUTO CDC. Un flux d’ajout simple INSERT INTO ONCE dans la même table ne passe pas la validation :

1. Créez la table de streaming cible dans laquelle votre flux AUTO CDC écrit.
2. Amorcez l'historique existant une seule fois à l'aide d'un flux AUTO CDC ONCE qui lit la table SCD existante en tant que flux, séquencé selon la colonne de début de validité de l'ancienne table. Relisez les lignes héritées en tant qu’événements de modification au lieu de les mettre en forme vous-même. AUTO CDC génère les colonnes d’historique __START_AT et __END_AT pour une cible SCD de type 2, n’écrivez donc pas ces colonnes directement.
3. Attachez le flux AUTO CDC continu en lisant le nouveau flux de modification. AUTO CDC résout le classement par clé, donc le basculement doit être respecté individuellement pour chaque clé d’entreprise : la première modification en direct de chaque clé doit être séquencée après la dernière modification amorcée pour cette même clé. Une valeur de séquence légèrement supérieure au maximum global hérité peut rester obsolète pour une clé individuelle, et le premier changement en temps réel de cette clé est alors ignoré ou mal ordonné.

Le code suivant crée une table de streaming qui suit les étapes ci-dessus :

CREATE OR REFRESH STREAMING TABLE customers_history;

-- One-time seed: replay the legacy history as change events
CREATE FLOW customers_history_seed
AS AUTO CDC ONCE INTO customers_history
FROM stream(legacy.customers_scd2)
KEYS (customer_id)
SEQUENCE BY valid_from
STORED AS SCD TYPE 2;

-- Ongoing live CDC into the same target
CREATE FLOW customers_history_cdc
AS AUTO CDC INTO customers_history
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY change_timestamp
STORED AS SCD TYPE 2;

Les deux flux doivent s’accorder sur leurs clés, leur type SCD et le type de données de leur colonne de séquençage. Dans l’exemple précédent, les deux flux sont séquencés selon un horodatage, qui utilise un seul instant de bascule pour séparer l’historique amorcé du flux en direct. Si la table héritée est séquencée selon une valeur d’un type différent de celui du flux en direct, effectuer un cast de l’un des deux afin que les types correspondent.

La même structure fonctionne pour une cible SCD de type 1 : remplacer STORED AS SCD TYPE 2 par STORED AS SCD TYPE 1 dans les deux flux, et la cible ne conserve que la ligne actuelle par clé. Avant de vous appuyer sur l’une ou l’autre configuration, validez, sur un échantillon de clés, que la première modification en production d’une clé initialisée produit exactement une nouvelle version et clôture correctement la version précédente. Un écart de séquençage par clé apparaît généralement à cette étape.

Ressources supplémentaires

• Charger et traiter des données de manière incrémentielle avec des flux de pipeline Lakeflow
• Utiliser des flux dans les pipelines Lakeflow
• Les API AUTO CDC : simplifiez la capture de changement de données avec des pipelines
• append_flow
• create_auto_cdc_from_snapshot_flow
• create_auto_cdc_flow
• CREATE FLOW (pipelines)