Comment créer un Data Lake scalable en 2026

version: '3.8'

services:
  minio:
    image: minio/minio:latest
    container_name: minio-server
    ports:
      - "9000:9000"  # API S3
      - "9001:9001"  # Console UI
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    volumes:
      - minio_data:/data
    command: server /data --console-address ":9001"

  volumes:
    minio_data:

networks:
  default:
    name: datalake-net

#!/bin/bash

# Télécharger mc (Linux/macOS, adaptez pour Windows)
curl -O https://dl.min.io/client/mc/release/linux-amd64/mc
chmod +x mc
sudo mv mc /usr/local/bin/mc

# Configurer alias pour notre MinIO local
mc alias set myminio http://localhost:9000 minioadmin minioadmin

# Vérifier connexion
echo "Buckets disponibles :"
mc ls myminio

# Nettoyage optionnel : rm mc si pas sudo

#!/bin/bash

BUCKET=datalake

# Créer bucket si inexistant
mc mb myminio/$BUCKET

# Policy JSON pour read public (local dev only)
cat > public-read.json << EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": "*",
      "Action": ["s3:GetObject"],
      "Resource": ["arn:aws:s3:::$BUCKET/*"]
    }
  ]
}
EOF

mc policy add public-read myminio/$BUCKET public-read.json

# Vérifier
echo "Policy :"
mc policy get-json myminio/$BUCKET
mc ls myminio

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from minio import Minio
from io import BytesIO
import numpy as np
from datetime import datetime, timedelta

client = Minio(
    "localhost:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    secure=False
)

# Générer data sample
np.random.seed(42)
dates = pd.date_range('2023-01-01', '2023-12-31', freq='H')[:1000]
data = pd.DataFrame({
    'id': range(1000),
    'date': dates,
    'product': np.random.choice(['Laptop', 'Phone', 'Tablet'], 1000),
    'amount': np.random.uniform(100, 1000, 1000),
    'region': np.random.choice(['EU', 'US', 'ASIA'], 1000)
})

# Partitionner par year/month
data['year'] = data['date'].dt.year.astype(str)
data['month'] = data['date'].dt.month.astype(str).str.zfill(2)

for (year, month), group in data.groupby(['year', 'month']):
    table = pa.Table.from_pandas(group)
    buffer = BytesIO()
    pq.write_table(table, buffer)
    buffer.seek(0)
    client.put_object(
        "datalake",
        f"sales/year={year}/month={month}/sales.parquet",
        io.BytesIO(buffer.read()),
        length=buffer.tell(),
        content_type="application/parquet"
    )
    print(f"Uploadé : year={year}/month={month}")

print("Ingestion terminée. Vérifiez UI MinIO.")

import duckdb

con = duckdb.connect()

# Installer/charger extension S3
con.execute("INSTALL httpfs; LOAD httpfs;")

# Config MinIO comme S3 (no secure)
con.execute """
SET s3_endpoint='localhost:9000';
SET s3_access_key='minioadmin';
SET s3_secret_key='minioadmin';
SET s3_region='us-east-1';
"""

# Query partitionnée : ventes EU 2023
result = con.execute(
    """
    SELECT 
        product, 
        AVG(amount) as avg_amount,
        COUNT(*) as sales_count
    FROM 's3://datalake/sales/year=2023/*/*.parquet'
    WHERE region = 'EU'
    GROUP BY product
    ORDER BY avg_amount DESC
    """
).fetchdf()

print(result)

# Export optionnel
result.to_csv('query-result.csv', index=False)
print("Résultat exporté en CSV.")

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from minio import Minio
from io import BytesIO
import numpy as np
from datetime import datetime

client = Minio("localhost:9000", access_key="minioadmin", secret_key="minioadmin", secure=False)

# Nouvelles data 2024 (évolue : ajoute 'category')
np.random.seed(123)
dates = pd.date_range('2024-01-01', '2024-03-31', freq='H')[:500]
data_new = pd.DataFrame({
    'id': range(1000, 1500),
    'date': dates,
    'product': np.random.choice(['Laptop Pro', 'Phone Ultra'], 500),
    'amount': np.random.uniform(500, 2000, 500),
    'region': np.random.choice(['EU', 'US'], 500),
    'category': np.random.choice(['Electronics', 'Mobile'], 500)  # Schema evolution
})

data_new['year'] = data_new['date'].dt.year.astype(str)
data_new['month'] = data_new['date'].dt.month.astype(str).str.zfill(2)

for (year, month), group in data_new.groupby(['year', 'month']):
    table = pa.Table.from_pandas(group)
    buffer = BytesIO()
    pq.write_table(table, buffer)
    buffer.seek(0)
    client.put_object(
        "datalake",
        f"sales/year={year}/month={month}/sales.parquet",
        io.BytesIO(buffer.read()),
        length=buffer.tell(),
        content_type="application/parquet"
    )
    print(f"Appended : year={year}/month={month}")

print("Append terminé. Re-query pour vérifier.")