Cloud & Object Storage for Data Engineers

# Exemple de metadata sur un objet S3
metadata_example = {
    # Metadata système (automatiques)
    "Content-Type": "application/octet-stream",
    "Content-Length": 1048576,
    "Last-Modified": "2024-01-15T10:30:00Z",
    "ETag": "d41d8cd98f00b204e9800998ecf8427e",
    
    # Metadata custom (x-amz-meta-*)
    "x-amz-meta-source": "kafka-topic-orders",
    "x-amz-meta-pipeline": "etl-daily",
    "x-amz-meta-schema-version": "2.1",
}

# Tags (pour billing, governance)
tags = {
    "Environment": "production",
    "Team": "data-engineering",
    "CostCenter": "DE-001",
}

print("Metadata et Tags permettent de :")
print("- Tracer l'origine des données")
print("- Filtrer pour la gouvernance")
print("- Allouer les coûts par équipe")

s3_cli_commands = """
# Lister les buckets
aws s3 ls

# Lister le contenu d'un bucket
aws s3 ls s3://my-bucket/bronze/

# Copier un fichier local vers S3
aws s3 cp data.csv s3://my-bucket/bronze/data.csv

# Copier un fichier S3 vers local
aws s3 cp s3://my-bucket/bronze/data.csv ./data.csv

# Synchroniser un dossier
aws s3 sync ./local-folder/ s3://my-bucket/bronze/

# Supprimer un fichier
aws s3 rm s3://my-bucket/bronze/old-data.csv

# Supprimer récursivement
aws s3 rm s3://my-bucket/temp/ --recursive
"""
print(s3_cli_commands)

# Installation : pip install boto3

import boto3
from botocore.exceptions import ClientError

# Créer un client S3
s3 = boto3.client('s3')

# --- Upload ---
def upload_file(file_path, bucket, key):
    """Upload un fichier vers S3."""
    s3.upload_file(file_path, bucket, key)
    print(f"✅ Uploaded {file_path} to s3://{bucket}/{key}")

# --- Download ---
def download_file(bucket, key, file_path):
    """Download un fichier depuis S3."""
    s3.download_file(bucket, key, file_path)
    print(f"✅ Downloaded s3://{bucket}/{key} to {file_path}")

# --- List objects ---
def list_objects(bucket, prefix=""):
    """Liste les objets dans un bucket."""
    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    if 'Contents' in response:
        for obj in response['Contents']:
            print(f"  {obj['Key']} ({obj['Size']} bytes)")
    return response.get('Contents', [])

# --- Presigned URL ---
def generate_presigned_url(bucket, key, expiration=3600):
    """Génère une URL temporaire pour accès direct."""
    url = s3.generate_presigned_url(
        'get_object',
        Params={'Bucket': bucket, 'Key': key},
        ExpiresIn=expiration
    )
    return url

# Exemple d'utilisation (commenté car pas de credentials)
# upload_file('data.csv', 'my-bucket', 'bronze/data.csv')
# list_objects('my-bucket', 'bronze/')
print("📝 Fonctions boto3 définies (upload, download, list, presigned URL)")

# Configuration Spark pour S3
spark_s3_config = """
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("S3 Access") \
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("spark.hadoop.fs.s3a.access.key", "YOUR_ACCESS_KEY") \
    .config("spark.hadoop.fs.s3a.secret.key", "YOUR_SECRET_KEY") \
    .config("spark.hadoop.fs.s3a.endpoint", "s3.amazonaws.com") \
    .getOrCreate()

# Lire depuis S3
df = spark.read.parquet("s3a://my-bucket/silver/sales/")

# Écrire vers S3
df.write.mode("overwrite").parquet("s3a://my-bucket/gold/aggregates/")
"""

# Optimisations S3A
s3a_optimizations = """
# Performance optimizations
spark.hadoop.fs.s3a.connection.maximum=200
spark.hadoop.fs.s3a.threads.max=64
spark.hadoop.fs.s3a.fast.upload=true
spark.hadoop.fs.s3a.fast.upload.buffer=bytebuffer
spark.hadoop.fs.s3a.multipart.size=104857600  # 100MB
"""
print(spark_s3_config)
print("\n--- Optimisations S3A ---")
print(s3a_optimizations)

# Installation : pip install azure-storage-blob

azure_blob_example = """
from azure.storage.blob import BlobServiceClient, BlobClient

# Connection string (dev only !)
connection_string = "DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;EndpointSuffix=core.windows.net"

# Créer le client
blob_service = BlobServiceClient.from_connection_string(connection_string)

# Upload
blob_client = blob_service.get_blob_client(container="bronze", blob="data.csv")
with open("data.csv", "rb") as f:
    blob_client.upload_blob(f, overwrite=True)

# Download
with open("downloaded.csv", "wb") as f:
    f.write(blob_client.download_blob().readall())

# List blobs
container_client = blob_service.get_container_client("bronze")
for blob in container_client.list_blobs():
    print(blob.name)
"""
print(azure_blob_example)

spark_azure_config = """
# Configuration Spark pour Azure Blob (ADLS Gen2)
storage_account = "mystorageaccount"
container = "bronze"

# Méthode 1 : Access Key (dev only)
spark.conf.set(
    f"fs.azure.account.key.{storage_account}.dfs.core.windows.net",
    "YOUR_ACCESS_KEY"
)

# Méthode 2 : Service Principal (production)
spark.conf.set(f"fs.azure.account.auth.type.{storage_account}.dfs.core.windows.net", "OAuth")
spark.conf.set(f"fs.azure.account.oauth.provider.type.{storage_account}.dfs.core.windows.net",
               "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set(f"fs.azure.account.oauth2.client.id.{storage_account}.dfs.core.windows.net", "CLIENT_ID")
spark.conf.set(f"fs.azure.account.oauth2.client.secret.{storage_account}.dfs.core.windows.net", "CLIENT_SECRET")
spark.conf.set(f"fs.azure.account.oauth2.client.endpoint.{storage_account}.dfs.core.windows.net",
               "https://login.microsoftonline.com/TENANT_ID/oauth2/token")

# Lire
df = spark.read.parquet(f"abfss://{container}@{storage_account}.dfs.core.windows.net/data/")
"""
print(spark_azure_config)

gcs_cli_commands = """
# gsutil - CLI pour GCS

# Lister les buckets
gsutil ls

# Lister le contenu d'un bucket
gsutil ls gs://my-bucket/bronze/

# Copier
gsutil cp data.csv gs://my-bucket/bronze/
gsutil cp gs://my-bucket/bronze/data.csv ./

# Synchroniser (comme rsync)
gsutil rsync -r ./local/ gs://my-bucket/bronze/

# Copie parallèle (gros fichiers)
gsutil -m cp -r ./data/ gs://my-bucket/bronze/
"""
print(gcs_cli_commands)

# Installation : pip install google-cloud-storage

gcs_python_example = """
from google.cloud import storage

# Créer le client (utilise GOOGLE_APPLICATION_CREDENTIALS)
client = storage.Client()

# Accéder au bucket
bucket = client.bucket("my-bucket")

# Upload
blob = bucket.blob("bronze/data.csv")
blob.upload_from_filename("data.csv")

# Download
blob.download_to_filename("downloaded.csv")

# List blobs
blobs = bucket.list_blobs(prefix="bronze/")
for blob in blobs:
    print(blob.name)

# Signed URL (temporaire)
url = blob.generate_signed_url(expiration=3600)  # 1 heure
"""
print(gcs_python_example)

spark_gcs_config = """
# Configuration Spark pour GCS

# Option 1 : Service Account Key (dev)
spark.conf.set("spark.hadoop.google.cloud.auth.service.account.enable", "true")
spark.conf.set("spark.hadoop.google.cloud.auth.service.account.json.keyfile", "/path/to/keyfile.json")

# Option 2 : Application Default Credentials (GKE, Cloud Functions)
# Pas de config nécessaire si ADC est configuré

# Lire depuis GCS
df = spark.read.parquet("gs://my-bucket/silver/data/")

# Écrire vers GCS
df.write.mode("overwrite").parquet("gs://my-bucket/gold/aggregates/")
"""
print(spark_gcs_config)

version: '3.8'

services:
  minio:
    image: minio/minio:latest
    container_name: minio
    ports:
      - "9000:9000"   # API S3
      - "9001:9001"   # Console Web
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: server /data --console-address ":9001"
    volumes:
      - minio-data:/data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 10s
      retries: 3

volumes:
  minio-data:

minio_commands = """
# Démarrer MinIO
docker-compose up -d

# Accéder à la console web
# http://localhost:9001
# Login: minioadmin / minioadmin

# Installer mc (MinIO Client)
# Linux
wget https://dl.min.io/client/mc/release/linux-amd64/mc
chmod +x mc
sudo mv mc /usr/local/bin/

# Mac
brew install minio/stable/mc

# Configurer mc
mc alias set myminio http://localhost:9000 minioadmin minioadmin

# Créer des buckets
mc mb myminio/bronze
mc mb myminio/silver
mc mb myminio/gold

# Lister
mc ls myminio/

# Upload
mc cp data.csv myminio/bronze/
"""
print(minio_commands)

# Le même code boto3 fonctionne avec MinIO !
# Il suffit de spécifier endpoint_url

minio_boto3_example = """
import boto3
from botocore.client import Config

# Configuration pour MinIO
s3 = boto3.client(
    's3',
    endpoint_url='http://localhost:9000',  # ← La seule différence !
    aws_access_key_id='minioadmin',
    aws_secret_access_key='minioadmin',
    config=Config(signature_version='s3v4')
)

# Créer un bucket
s3.create_bucket(Bucket='bronze')

# Upload (identique à S3)
s3.upload_file('data.csv', 'bronze', 'raw/data.csv')

# List (identique à S3)
response = s3.list_objects_v2(Bucket='bronze')
for obj in response.get('Contents', []):
    print(obj['Key'])

# Download (identique à S3)
s3.download_file('bronze', 'raw/data.csv', 'downloaded.csv')
"""
print(minio_boto3_example)
print("\n💡 Ce code fonctionne sur S3 en enlevant juste endpoint_url !")

spark_minio_config = """
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MinIO Access") \
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("spark.hadoop.fs.s3a.endpoint", "http://localhost:9000") \
    .config("spark.hadoop.fs.s3a.access.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.secret.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .getOrCreate()

# Lire depuis MinIO (même syntaxe que S3 !)
df = spark.read.csv("s3a://bronze/raw/data.csv", header=True)

# Écrire vers MinIO
df.write.mode("overwrite").parquet("s3a://silver/clean/data/")
"""
print(spark_minio_config)

# Éviter les petits fichiers avec Spark
small_files_solution = """
# Lecture de beaucoup de petits fichiers
df = spark.read.parquet("s3a://bronze/data/")  # 10,000 fichiers

# ❌ Écriture directe = même nombre de fichiers
# df.write.parquet("s3a://silver/data/")

# ✅ Repartitionner avant d'écrire
df.repartition(100) \  # 100 fichiers de ~100 MB
  .write.mode("overwrite") \
  .parquet("s3a://silver/data/")

# ✅ Ou coalesce (moins de shuffle)
df.coalesce(100) \
  .write.mode("overwrite") \
  .parquet("s3a://silver/data/")
"""
print(small_files_solution)

# Partitionnement avec Spark
partitioning_example = """
# Écriture partitionnée
df.write \
  .partitionBy("year", "month") \
  .mode("overwrite") \
  .parquet("s3a://silver/sales/")

# Lecture avec partition pruning
df = spark.read.parquet("s3a://silver/sales/")

# Cette requête ne lit QUE year=2024/month=01
df.filter("year = 2024 AND month = 1").show()
"""
print(partitioning_example)

# Exercice : Comparer la taille et le temps de lecture

format_comparison = """
import time
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Format Comparison").getOrCreate()

# Créer un DataFrame de test (1M lignes)
df = spark.range(1000000).toDF("id") \
    .withColumn("name", lit("test_name")) \
    .withColumn("amount", rand() * 1000)

# Écrire en CSV
start = time.time()
df.write.mode("overwrite").csv("s3a://test/csv/")
csv_write_time = time.time() - start

# Écrire en Parquet
start = time.time()
df.write.mode("overwrite").parquet("s3a://test/parquet/")
parquet_write_time = time.time() - start

# Lire CSV
start = time.time()
df_csv = spark.read.csv("s3a://test/csv/").count()
csv_read_time = time.time() - start

# Lire Parquet
start = time.time()
df_parquet = spark.read.parquet("s3a://test/parquet/").count()
parquet_read_time = time.time() - start

print(f"CSV Write: {csv_write_time:.2f}s, Read: {csv_read_time:.2f}s")
print(f"Parquet Write: {parquet_write_time:.2f}s, Read: {parquet_read_time:.2f}s")
"""
print(format_comparison)
print("\n💡 Résultat attendu : Parquet 5-10x plus rapide et 5-10x plus petit")

# Exemple de bucket policy S3
bucket_policy_example = """
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowDataTeamRead",
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::123456789:role/DataEngineerRole"
            },
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::my-datalake",
                "arn:aws:s3:::my-datalake/*"
            ]
        },
        {
            "Sid": "DenyPublicAccess",
            "Effect": "Deny",
            "Principal": "*",
            "Action": "s3:*",
            "Resource": "arn:aws:s3:::my-datalake/*",
            "Condition": {
                "Bool": {
                    "aws:SecureTransport": "false"
                }
            }
        }
    ]
}
"""
print(bucket_policy_example)

# Exemple de lifecycle policy S3
lifecycle_policy = """
{
    "Rules": [
        {
            "ID": "ArchiveBronzeData",
            "Status": "Enabled",
            "Filter": {
                "Prefix": "bronze/"
            },
            "Transitions": [
                {
                    "Days": 30,
                    "StorageClass": "STANDARD_IA"
                },
                {
                    "Days": 90,
                    "StorageClass": "GLACIER"
                }
            ],
            "Expiration": {
                "Days": 365
            }
        }
    ]
}
"""
print(lifecycle_policy)

# Étape 1 : Démarrer MinIO
print("""
# Terminal 1 : Démarrer MinIO
docker run -d --name minio \
  -p 9000:9000 -p 9001:9001 \
  -e MINIO_ROOT_USER=minioadmin \
  -e MINIO_ROOT_PASSWORD=minioadmin \
  minio/minio server /data --console-address ":9001"

# Vérifier que MinIO tourne
curl http://localhost:9000/minio/health/live
""")

# Étape 2 : Créer les buckets et uploader les données

step2_code = """
import boto3
from botocore.client import Config
import csv

# Connexion à MinIO
s3 = boto3.client(
    's3',
    endpoint_url='http://localhost:9000',
    aws_access_key_id='minioadmin',
    aws_secret_access_key='minioadmin',
    config=Config(signature_version='s3v4')
)

# Créer les buckets
for bucket in ['bronze', 'silver', 'gold']:
    try:
        s3.create_bucket(Bucket=bucket)
        print(f"✅ Bucket '{bucket}' créé")
    except Exception as e:
        print(f"⚠️ Bucket '{bucket}' existe déjà")

# Créer des données de test
with open('sales_data.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['date', 'product', 'category', 'amount', 'quantity'])
    writer.writerow(['2024-01-01', 'Laptop', 'Electronics', 1200, 1])
    writer.writerow(['2024-01-01', 'Mouse', 'Electronics', 25, 3])
    writer.writerow(['2024-01-02', 'Desk', 'Furniture', 350, 1])
    writer.writerow(['2024-01-02', 'Chair', 'Furniture', 150, 2])
    writer.writerow(['2024-01-03', 'Laptop', 'Electronics', 1200, 2])
    writer.writerow(['2024-01-03', 'Monitor', 'Electronics', 400, 1])

# Upload vers bronze
s3.upload_file('sales_data.csv', 'bronze', 'raw/sales_data.csv')
print("✅ Données uploadées vers bronze/raw/sales_data.csv")
"""
print(step2_code)

# Étape 3 : Transformer avec PySpark (Bronze → Silver)

step3_code = """
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, upper

# Créer SparkSession avec config MinIO
spark = SparkSession.builder \
    .appName("Bronze to Silver") \
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("spark.hadoop.fs.s3a.endpoint", "http://localhost:9000") \
    .config("spark.hadoop.fs.s3a.access.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.secret.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .getOrCreate()

# Lire depuis Bronze
df_bronze = spark.read.csv(
    "s3a://bronze/raw/sales_data.csv",
    header=True,
    inferSchema=True
)

print("📥 Données Bronze:")
df_bronze.show()

# Transformer
df_silver = df_bronze \
    .withColumn("date", to_date(col("date"))) \
    .withColumn("category", upper(col("category"))) \
    .withColumn("total", col("amount") * col("quantity")) \
    .dropDuplicates()

print("🔄 Données transformées:")
df_silver.show()

# Écrire vers Silver (Parquet partitionné)
df_silver.write \
    .mode("overwrite") \
    .partitionBy("category") \
    .parquet("s3a://silver/sales/")

print("✅ Données écrites vers silver/sales/")
"""
print(step3_code)

# Étape 4 : Agréger avec Spark SQL (Silver → Gold)

step4_code = """
# Lire depuis Silver
df_silver = spark.read.parquet("s3a://silver/sales/")

# Créer une vue temporaire
df_silver.createOrReplaceTempView("sales")

# Agrégation avec Spark SQL
df_gold = spark.sql(\"\"\"
    SELECT 
        category,
        COUNT(*) as num_transactions,
        SUM(quantity) as total_quantity,
        SUM(total) as total_revenue,
        AVG(total) as avg_transaction
    FROM sales
    GROUP BY category
    ORDER BY total_revenue DESC
\"\"\")

print("📊 Agrégations Gold:")
df_gold.show()

# Écrire vers Gold
df_gold.coalesce(1) \
    .write \
    .mode("overwrite") \
    .parquet("s3a://gold/category_summary/")

print("✅ Données écrites vers gold/category_summary/")
"""
print(step4_code)

# Étape 5 : Vérifier les résultats

step5_code = """
# Lister les fichiers créés
import subprocess

# Avec mc CLI
print("📁 Contenu de bronze/:")
!mc ls myminio/bronze/ --recursive

print("\n📁 Contenu de silver/:")
!mc ls myminio/silver/ --recursive

print("\n📁 Contenu de gold/:")
!mc ls myminio/gold/ --recursive

# Ou avec boto3
for bucket in ['bronze', 'silver', 'gold']:
    print(f"\n📁 {bucket}/")
    response = s3.list_objects_v2(Bucket=bucket)
    for obj in response.get('Contents', []):
        print(f"   {obj['Key']} ({obj['Size']} bytes)")
"""
print(step5_code)