Démarrer un Data Lakehouse en un après-midi

n’a pas besoin d’être que compliqué. Dans cet article, je vais vous montrer comment développer une table de base « de démarrage » qui utilise une table Iceberg sur le stockage AWS S3. Une fois la table enregistrée à l’aide d’AWS Glue, vous pourrez l’interroger et la muter depuis Amazon Athena, notamment en utilisant :

• Fusionner, mettre à jour et supprimer des données

• Optimiser et aspirer vos tables.

Je vais également vous montrer comment inspecter les mêmes tables localement depuis CanardDB, et nous verrons aussi comment utiliser Colle/Étincelle pour insérer plus de données de table.

Notre exemple est peut-être basique, mais il présentera la configuration, les différents outils et les processus que vous pouvez mettre en place pour créer un magasin de données plus étendu. Tous les fournisseurs de cloud modernes disposent d’équivalents aux services AWS dont je parle dans cet article. Il devrait donc être assez simple de reproduire ce dont je parle ici sur Azure, Google Cloud et autres.

Pour nous assurer que nous sommes tous sur la même longueur d’onde, voici une brève explication de certaines des technologies clés que nous utiliserons.

AWS Colle/Étincelle

AWS Glue est un service ETL sans serveur entièrement géré d’Amazon qui rationalise la préparation et l’intégration des données pour l’analyse et l’apprentissage automatique. Il détecte et catalogue automatiquement les métadonnées provenant de diverses sources, telles que S3, dans un magasin de données centralisé. De plus, il peut créer des scripts Spark ETL personnalisables basés sur Python pour exécuter ces tâches sur une plate-forme Apache Spark évolutive et sans serveur. Cela le rend idéal pour créer des lacs de données sur Amazon S3, charger des données dans des entrepôts de données comme Amazon Redshift et effectuer le nettoyage et la transformation des données. le tout sans gérer les infrastructures.

AWS Athéna

AWS Athena est un service de requête interactif qui simplifie l’analyse des données directement dans Amazon S3 à l’aide du SQL standard. En tant que plateforme sans serveur, il n’est pas nécessaire de gérer ou de provisionner des serveurs ; pointez simplement Athena vers vos données S3, définissez votre schéma (généralement avec AWS Glue) et commencez à exécuter des requêtes SQL. Il est fréquemment utilisé pour l’analyse, la création de rapports et l’exploration ad hoc d’ensembles de données volumineux dans des formats tels que CSV, JSON, ORC ou Parquet.

Tableaux d’icebergs

Les tables Iceberg sont un format de table ouvert pour les ensembles de données qui fournissent des fonctionnalités de type base de données pour les données stockées dans des lacs de données, tels que le stockage d’objets Amazon S3. Traditionnellement, sur S3, vous pouvez créer, lire et supprimer des objets (fichiers), mais leur mise à jour n’est pas possible. Le format Iceberg répond à cette limitation tout en offrant d’autres avantages, notamment les transactions ACID, l’évolution du schéma, le partitionnement caché et les fonctionnalités de voyage dans le temps.

CanardDB

DuckDB est une base de données analytique en mémoire écrite en C++ et conçue pour les charges de travail SQL analytiques. Depuis sa sortie il y a quelques années, il a gagné en popularité et constitue désormais l’un des principaux outils de traitement de données utilisés par les ingénieurs de données et les scientifiques, grâce à ses bases en SQL, ses performances et sa polyvalence.

Présentation du scénario

Disons que vous avez été chargé de créer une petite table d’analyse « entrepôt-lite » pour les événements de commande, mais que vous ne souhaitez pas encore adopter une plate-forme lourde. Il vous faut :

• Écritures sécurisées (pas de lecteurs cassés, pas de commits partiels)
• Au niveau de la ligne modifications (UPDATE/DELETE/MERGE, pas seulement ajouter)
• Lectures ponctuelles (pour les audits et le débogage)
• Analyses locales par rapport aux données de production précises pour des contrôles rapides

Ce que nous allons construire

1. Créez une table Iceberg dans Colle & S3 via Athéna
2. Charger et muter les lignes (INSÉRER/MISE À JOUR/SUPPRIMER/FUSIONNER)
3. Voyage dans le temps aux instantanés précédents (par horodatage et par ID d’instantané)
4. Continuez vite avec OPTIMISER et VIDE
5. Lire le même tableau localement de CanardDB (Accès S3 via DuckDB Secrets)
6. Découvrez comment ajouter de nouveaux enregistrements à notre table à l’aide du code Glue Spark

Donc, en un mot, nous utiliserons : –

• S3 pour le stockage des données
• Catalogue Glue pour les métadonnées/découverte des tables
• Athena pour les lectures SQL sans serveur et écrit
• DuckDB pour des analyses locales bon marché par rapport à la même table Iceberg
• Étincelle pour traiter le grognement

De notre point de vue, le principal point à retenir est qu’en utilisant les technologies ci-dessus, nous serons en mesure d’effectuer des requêtes de type base de données sur le stockage d’objets.

Mise en place de notre environnement de développement

Je préfère isoler les outils locaux dans un environnement séparé. Utilisez n’importe quel outil que vous aimez pour ce faire ; Je vais montrer en utilisant conda puisque c’est ce que je fais habituellement. À des fins de démonstration, j’exécuterai tout le code dans un environnement Jupyter Notebook.

# create and activate a local env
conda create -n iceberg-demo python=3.11 -y
conda activate iceberg-demo

# install duckdb CLI + Python package and awscli for quick tests
pip install duckdb awscli jupyter

Conditions préalables

Comme nous utiliserons les services AWS, vous aurez besoin d’un compte AWS. Aussi,

• Un compartiment S3 pour le lac de données (par exemple, s3://my-demo-lake/warehouse/)
• Une base de données Glue (nous allons en créer une)
• Moteur Athéna version 3 jedans votre groupe de travail
• Un rôle ou un utilisateur IAM pour Athena avec les autorisations S3 + Glue

1/ Configuration d’Athéna

Une fois connecté à AWS, ouvrez Athena dans la console et définissez votre groupe de travail, la version du moteur et l’emplacement de sortie S3 (pour les résultats de la requête). Pour ce faire, recherchez une icône de menu de style hamburger en haut à gauche de l’écran d’accueil d’Athena. Cliquez dessus pour faire apparaître un nouveau bloc de menu sur la gauche. Là, vous devriez voir un lien Administration-> Groupes de travail. Vous serez automatiquement affecté au groupe de travail principal. Vous pouvez vous en tenir à cela ou en créer un nouveau si vous le souhaitez. Quelle que soit l’option que vous choisissez, modifiez-la et assurez-vous que les options suivantes sont sélectionnées.

• Moteur d’analyse – Athena SQL. Réglez manuellement la version du moteur sur 3.0.
• Sélectionnez la configuration des résultats de requête gérée par le client et saisissez les informations requises sur le compartiment et le compte.

2/ Créer une table Iceberg dans Athena

Nous stockerons les événements de commande et laisserons Iceberg gérer le partitionnement de manière transparente. J’utiliserai une partition « cachée » le jour de l’horodatage pour répartir les écritures/lectures. Revenez à la page d’accueil d’Athena et lancez l’éditeur de requêtes Trino SQL. Votre écran devrait ressembler à ceci.

Tapez et exécutez le SQL suivant. Modifiez les noms de bucket/table en fonction.

-- This automatically creates a Glue database 
-- if you don't have one already
CREATE DATABASE IF NOT EXISTS analytics;

CREATE TABLE analytics.sales_iceberg (
  order_id    bigint,
  customer_id bigint,
  ts          timestamp,
  status      string,
  amount_usd  double
)
PARTITIONED BY (day(ts))
LOCATION 's3://your_bucket/warehouse/sales_iceberg/'
TBLPROPERTIES (
  'table_type' = 'ICEBERG',
  'format' = 'parquet',
  'write_compression' = 'snappy'
)

3) Charger et muter les données (INSERT / UPDATE / DELETE / MERGE)

Athena prend en charge le véritable Iceberg DML, vous permettant d’insérer des lignes, de mettre à jour et de supprimer des enregistrements, ainsi que d’effectuer des insertions à l’aide de l’instruction MERGE. Sous le capot, Iceberg utilise ACID basé sur des instantanés avec suppression de fichiers ; les lecteurs restent cohérents tandis que les écrivains travaillent en parallèle.

Semez quelques rangs.

INSERT INTO analytics.sales_iceberg VALUES
  (101, 1, timestamp '2025-08-01 10:00:00', 'created', 120.00),
  (102, 2, timestamp '2025-08-01 10:05:00', 'created',  75.50),
  (103, 2, timestamp '2025-08-02 09:12:00', 'created',  49.99),
  (104, 3, timestamp '2025-08-02 11:47:00', 'created', 250.00);

Une vérification rapide de la santé mentale.

SELECT * FROM analytics.sales_iceberg ORDER BY order_id;

 order_id | customer_id |          ts           |  status  | amount_usd
----------+-------------+-----------------------+----------+-----------
  101     | 1           | 2025-08-01 10:00:00   | created  | 120.00
  102     | 2           | 2025-08-01 10:05:00   | created  |  75.50
  103     | 2           | 2025-08-02 09:12:00   | created  |  49.99
  104     | 3           | 2025-08-02 11:47:00   | created  | 250.00

Mettre à jour et supprimer.

UPDATE analytics.sales_iceberg
SET status = 'paid'
WHERE order_id IN (101, 102)

-- removes order 103
DELETE FROM analytics.sales_iceberg
WHERE status = 'created' AND amount_usd < 60

Upserts idempotents avec MERGE

Traitons la commande 104 comme remboursée et créons une nouvelle commande 105.

MERGE INTO analytics.sales_iceberg AS t
USING (
  VALUES
    (104, 3, timestamp '2025-08-02 11:47:00', 'refunded', 250.00),
    (105, 4, timestamp '2025-08-03 08:30:00', 'created',   35.00)
) AS s(order_id, customer_id, ts, status, amount_usd)
ON s.order_id = t.order_id
WHEN MATCHED THEN 
  UPDATE SET 
    customer_id = s.customer_id,
    ts = s.ts,
    status = s.status,
    amount_usd = s.amount_usd
WHEN NOT MATCHED THEN 
  INSERT (order_id, customer_id, ts, status, amount_usd)
  VALUES (s.order_id, s.customer_id, s.ts, s.status, s.amount_usd);

Vous pouvez maintenant réinterroger pour voir : 101/102 → payé103 supprimé, 104 → rembourséet 105 → créé. (Si vous exécutez cela dans un compte « réel », vous remarquerez que le nombre d’objets S3 augmente – nous en saurons plus sur la maintenance sous peu.)

SELECT * FROM analytics.sales_iceberg ORDER BY order_id

# order_id customer_id ts status amount_usd
1 101 1 2025-08-01 10:00:00.000000 paid 120.0
2 105 4 2025-08-03 08:30:00.000000 created 35.0
3 102 2 2025-08-01 10:05:00.000000 paid 75.5
4 104 3 2025-08-02 11:47:00.000000 refunded 250.0

4) Voyage dans le temps (et voyage en version)

C’est là que réside la véritable valeur de l’utilisation d’Iceberg. Vous pouvez interroger la table telle qu’elle était à un moment donné ou par un ID d’instantané spécifique. Dans Athena, utilisez cette syntaxe,

-- Time travel to noon on Aug 2 (UTC)
SELECT order_id, status, amount_usd
FROM analytics.sales_iceberg
FOR TIMESTAMP AS OF TIMESTAMP '2025-08-02 12:00:00 UTC'
ORDER BY order_id;

-- Or Version travel (replace the id with an actual snapshot id from your table)

SELECT *
FROM analytics.sales_iceberg
FOR VERSION AS OF 949530903748831860;

Pour obtenir les différents ID de version (instantané) associés à une table particulière, utilisez cette requête.

SELECT * FROM "analytics"."sales_iceberg$snapshots"
ORDER BY committed_at DESC;

5) Garder vos données en bonne santé : OPTIMISER et VIDE

Les écritures au niveau des lignes (UPDATE/DELETE/MERGE) créent de nombreuses supprimer des fichiers et peut fragmenter les données. Deux affirmations garantissent la rapidité et la facilité de stockage :

• OPTIMISER… RÉÉCRIRE LES DONNÉES À L’AIDE DE BIN_PACK — compacte les fichiers petits/fragmentés et replie les suppressions en données
• VIDE – expire les anciens instantanés + nettoie orphelin fichiers

-- compact "hot" data (yesterday) and merge deletes
OPTIMIZE analytics.sales_iceberg
REWRITE DATA USING BIN_PACK
WHERE ts >= date_trunc('day', current_timestamp - interval '1' day);

-- expire old snapshots and remove orphan files
VACUUM analytics.sales_iceberg;

6) Analyse locale avec DuckDB (lecture seule)

C’est formidable de pouvoir vérifier l’intégrité des tables de production à partir d’un ordinateur portable sans avoir à exécuter un cluster. DuckDB httpfs + iceberg les extensions simplifient les choses.

6.1 Installer et charger des extensions

Ouvrez votre bloc-notes Jupyter et saisissez ce qui suit.

# httpfs gives S3 support; iceberg adds Iceberg readers.

import duckdb as db
db.sql("install httpfs; load httpfs;")
db.sql("install iceberg; load iceberg;")

6.2 Fournir les informations d’identification S3 à DuckDB de la « bonne » manière (Secrets)

DuckDB dispose d’un gestionnaire de secrets petit mais puissant. La configuration la plus robuste dans AWS est la chaîne d’identificationfournisseur, qui réutilise tout ce que le SDK AWS peut trouver (variables d’environnement, rôle IAM, etc.). Par conséquent, vous devrez vous assurer que, par exemple, vos informations d’identification AWS CLI sont configurées.

db.sql("""CREATE SECRET ( TYPE s3, PROVIDER credential_chain )""")

Après cela, n’importe quel s3://… les lectures dans cette session DuckDB utiliseront les données secrètes.

6.3 Pointez DuckDB sur les métadonnées de la table Iceberg

La manière la plus explicite consiste à référencer un fichier de métadonnées concret (par exemple, le dernier fichier de votre table). metadata/ dossier:)

Pour obtenir une liste de ceux-ci, utilisez cette requête

result = db.sql("""
SELECT *
FROM glob('s3://your_bucket/warehouse/**')
ORDER BY file
""")
print(result)

...
...
s3://your_bucket_name/warehouse/sales_iceberg/metadata/00000-942a25ce-24e5-45f8-ae86-b70d8239e3bb.metadata.json                                      │
s3://your_bucket_name/warehouse/sales_iceberg/metadata/00001-fa2d9997-590e-4231-93ab-642c0da83f19.metadata.json                                      │
s3://your_bucket_name/warehouse/sales_iceberg/metadata/00002-0da3a4af-64af-4e46-bea2-0ac450bf1786.metadata.json                                      │
s3://your_bucket_name/warehouse/sales_iceberg/metadata/00003-eae21a3d-1bf3-4ed1-b64e-1562faa445d0.metadata.json                                      │
s3://your_bucket_name/warehouse/sales_iceberg/metadata/00004-4a2cff23-2bf6-4c69-8edc-6d74c02f4c0e.metadata.json    
...
...
...

Cherchez le fichier metadata.json avec le début du nom de fichier ayant le numéro le plus élevé 00004 dans mon cas. Ensuite, vous pouvez l’utiliser dans une requête comme celle-ci pour récupérer la dernière position de votre table sous-jacente.

# Use the highest numbered metadata file (00004 appears to be the latest in my case)
result = db.sql("""
SELECT *
FROM iceberg_scan('s3://your_bucket/warehouse/sales_iceberg/metadata/00004-4a2cff23-2bf6-4c69-8edc-6d74c02f4c0e.metadata.json')
LIMIT 10
""")
print(result)

┌──────────┬─────────────┬─────────────────────┬──────────┬────────────┐
│ order_id │ customer_id │         ts          │  status  │ amount_usd │
│  int64   │    int64    │      timestamp      │ varchar  │   double   │
├──────────┼─────────────┼─────────────────────┼──────────┼────────────┤
│      105 │           4 │ 2025-08-03 08:30:00 │ created  │       35.0 │
│      104 │           3 │ 2025-08-02 11:47:00 │ refunded │      250.0 │
│      101 │           1 │ 2025-08-01 10:00:00 │ paid     │      120.0 │
│      102 │           2 │ 2025-08-01 10:05:00 │ paid     │       75.5 │
└──────────┴─────────────┴─────────────────────┴──────────┴────────────┘

Vous voulez un instantané spécifique ? Utilisez-le pour obtenir une liste.

result = db.sql("""
SELECT *
FROM iceberg_snapshots('s3://your_bucket/warehouse/sales_iceberg/metadata/00004-4a2cff23-2bf6-4c69-8edc-6d74c02f4c0e.metadata.json')
""")
print("Available Snapshots:")
print(result)

Available Snapshots:
┌─────────────────┬─────────────────────┬─────────────────────────┬──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ sequence_number │     snapshot_id     │      timestamp_ms       │                                                          manifest_list                                                           │
│     uint64      │       uint64        │        timestamp        │                                                             varchar                                                              │
├─────────────────┼─────────────────────┼─────────────────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│               1 │ 5665457382547658217 │ 2025-09-09 10:58:44.225 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-5665457382547658217-1-bb7d0497-0f97-4483-98e2-8bd26ddcf879.avro │
│               3 │ 8808557756756599285 │ 2025-09-09 11:19:24.422 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-8808557756756599285-1-f83d407d-ec31-49d6-900e-25bc8d19049c.avro │
│               2 │   31637314992569797 │ 2025-09-09 11:08:08.805 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-31637314992569797-1-000a2e8f-b016-4d91-9942-72fe9ddadccc.avro   │
│               4 │ 4009826928128589775 │ 2025-09-09 11:43:18.117 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-4009826928128589775-1-cd184303-38ab-4736-90da-52e0cf102abf.avro │
└─────────────────┴─────────────────────┴─────────────────────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

7) Supplément optionnel : Écriture à partir de Spark/Glue

Si vous préférez Spark pour les écritures par lots plus importantes, Glue peut lire/écrire les tables Iceberg enregistrées dans le catalogue Glue. Vous souhaiterez probablement toujours utiliser Athena pour le SQL ad hoc, les voyages dans le temps et la maintenance, mais de gros CTAS/ETL peuvent provenir de tâches Glue. (Sachez simplement que la compatibilité des versions et les autorisations AWS LakeFormation peuvent avoir un impact, car Glue et Athena peuvent être légèrement en retard sur les versions Iceberg.)

Voici un exemple de code Glue Spark qui insère quelques nouvelles lignes de données, commençant à order_id = 110, dans notre table existante. Avant d’exécuter ceci, vous devez ajouter le paramètre de tâche Glue suivant (sous Détails du travail Glue-> Paramètres avancés-> Paramètres du travail.

Key: --conf
Value: spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

import sys
import random
from datetime import datetime
from pyspark.context import SparkContext
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql import Row

# --------------------------------------------------------
# Init Glue job
# --------------------------------------------------------
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# --------------------------------------------------------
# Force Iceberg + Glue catalog configs (dynamic only)
# --------------------------------------------------------
spark.conf.set("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.warehouse", "s3://your_bucket/warehouse/")
spark.conf.set("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
spark.conf.set("spark.sql.defaultCatalog", "glue_catalog")

# --------------------------------------------------------
# Debug: list catalogs to confirm glue_catalog is registered
# --------------------------------------------------------
print("Current catalogs available:")
spark.sql("SHOW CATALOGS").show(truncate=False)

# --------------------------------------------------------
# Read existing Iceberg table (optional)
# --------------------------------------------------------
existing_table_df = glueContext.create_data_frame.from_catalog(
    database="analytics",
    table_name="sales_iceberg"
)
print("Existing table schema:")
existing_table_df.printSchema()

# --------------------------------------------------------
# Create 5 new records
# --------------------------------------------------------
new_records_data = []
for i in range(5):
    order_id = 110 + i
    record = {
        "order_id": order_id,
        "customer_id": 1000 + (i % 10),
        "price": round(random.uniform(10.0, 500.0), 2),
        "created_at": datetime.now(),
        "status": "completed"
    }
    new_records_data.append(record)

new_records_df = spark.createDataFrame([Row(**r) for r in new_records_data])
print(f"Creating {new_records_df.count()} new records:")
new_records_df.show()

# Register temp view for SQL insert
new_records_df.createOrReplaceTempView("new_records_temp")

# --------------------------------------------------------
# Insert into Iceberg table (alias columns as needed)
# --------------------------------------------------------
spark.sql("""
    INSERT INTO analytics.sales_iceberg (order_id, customer_id, ts, status, amount_usd)
    SELECT order_id,
           customer_id,
           created_at AS ts,
           status,
           price AS amount_usd
    FROM new_records_temp
""")

print(" Sccessfully added 5 new records to analytics.sales_iceberg")

# --------------------------------------------------------
# Commit Glue job
# --------------------------------------------------------
job.commit()

Vérifiez à nouveau auprès d’Athéna.

select * from analytics.sales_iceberg
order by order_id

# order_id customer_id ts status amount_usd
1 101 1 2025-08-01 10:00:00.000000 paid 120.0
2 102 2 2025-08-01 10:05:00.000000 paid 75.5
3 104 3 2025-08-02 11:47:00.000000 refunded 250.0
4 105 4 2025-08-03 08:30:00.000000 created 35.0
5 110 1000 2025-09-10 16:06:45.505935 completed 248.64
6 111 1001 2025-09-10 16:06:45.505947 completed 453.76
7 112 1002 2025-09-10 16:06:45.505955 completed 467.79
8 113 1003 2025-09-10 16:06:45.505963 completed 359.9
9 114 1004 2025-09-10 16:06:45.506059 completed 398.52

Étapes futures

À partir de là, vous pouvez :

• Créez plus de tableaux avec des données.

• Expérimentez l’évolution de la partition (par exemple, changez la partition de la table du jour à l’heure à mesure que les volumes augmentent),

• Ajoutez une maintenance planifiée. Par exemple, EventBridgeStep et Lambdas pourraient être utilisés pour exécuter OPTIMISER/VIDE à un horaire programmé cadence.

Résumé

Dans cet article, j’ai essayé de fournir une voie claire pour créer un Lac de données Icebergsur AWS. Il devrait servir de guide aux ingénieurs de données qui souhaitent connecter un stockage d’objets simple à des entrepôts de données d’entreprise complexes.

J’espère avoir montré que la construction d’un Data Lakehouse (un système combinant le faible coût des lacs de données avec l’intégrité transactionnelle des entrepôts) ne nécessite pas nécessairement de longues extensions.déploiement d’infrastructures ive. Et même si la création d’une maison au bord d’un lac complète est quelque chose qui évolue sur une longue période, j’espère vous avoir convaincu que vous pouvez vraiment en fabriquer une en un après-midi.

En tirant parti Apache Icebergsur un système de stockage cloud comme Amazone S3j’ai démontré comment transformer des fichiers statiques en tables dynamiques et gérées capables d’effectuer des transactions ACID, des mutations au niveau des lignes (MERGE, UPDATE, DELETE) et des voyages dans le temps, le tout sans provisionner un seul serveur.

J’ai également montré qu’en utilisant de nouveaux outils d’analyse tels que CanardDBil est possible de lire localement des lacs de données petits à moyens. Et lorsque vos volumes de données augmentent et deviennent trop importants pour un traitement local, j’ai montré à quel point il était facile de passer à une plate-forme de traitement de données d’entreprise telle que Étincelle.