Cloud Platforms para Data Engineering

Stack de Datos AWS 2026

AWS S3: Mejores Prácticas

"""
Trabajar con S3 eficientemente desde Python
boto3 + s3fs + pandas
"""
import boto3
import s3fs
import pandas as pd
from pathlib import Path

# ── BOTO3: AWS SDK ──────────────────────────────────────────────────────────
s3 = boto3.client('s3',
    region_name='us-east-1',
    # Usar IAM roles en producción, nunca credenciales hardcoded
)

# Subir archivo
s3.upload_file('local_file.parquet', 'my-data-lake', 'raw/orders/2026/01/01/data.parquet')

# Listar objetos con paginación
paginator = s3.get_paginator('list_objects_v2')
pages = paginator.paginate(Bucket='my-data-lake', Prefix='raw/orders/2026/')

all_keys = []
for page in pages:
    if 'Contents' in page:
        all_keys.extend([obj['Key'] for obj in page['Contents']])

# Leer Parquet directamente desde S3 (sin descargar)
fs = s3fs.S3FileSystem()
df = pd.read_parquet('s3://my-data-lake/silver/orders/year=2026/month=1/', 
                      filesystem=fs,
                      columns=['order_id', 'customer_id', 'amount'])

# ── BEST PRACTICES para S3 Data Lake ─────────────────────────────────────
# 1. Particionamiento Hive-style: year=YYYY/month=MM/day=DD/
# 2. Usar prefixes para distribución de carga (key prefix sharding)
# 3. Tamaño ideal de archivos: 128MB - 1GB
# 4. Habilitar S3 Versioning para tablas críticas
# 5. S3 Lifecycle policies para archivar datos antiguos a Glacier
# 6. Encryption at rest: S3-SSE o KMS para datos sensibles
# 7. VPC Endpoints: acceso a S3 sin internet (seguridad)

# ── AWS GLUE: ETL Serverless ───────────────────────────────────────────────
# Glue Job en PySpark - se ejecuta en un cluster managed por AWS
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

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

# Leer desde Glue Data Catalog
orders_dyf = glueContext.create_dynamic_frame.from_catalog(
    database='raw_data',
    table_name='orders',
    push_down_predicate=f"run_date = '{args['run_date']}'",
)

# Convertir a DataFrame Spark para transformaciones complejas
orders_df = orders_dyf.toDF()
# ... transformaciones ...

# Escribir de vuelta al Data Lake con Glue
output_dyf = DynamicFrame.fromDF(orders_df, glueContext, "output")
glueContext.write_dynamic_frame.from_options(
    frame=output_dyf,
    connection_type='s3',
    connection_options={
        'path': 's3://my-data-lake/silver/orders/',
        'partitionKeys': ['year', 'month'],
    },
    format='glueparquet',
)

job.commit()

# ── AMAZON ATHENA: SQL sobre S3 ────────────────────────────────────────────
import boto3
import time
import pandas as pd

athena = boto3.client('athena', region_name='us-east-1')

def run_athena_query(sql: str, database: str, output_bucket: str) -> pd.DataFrame:
    """Ejecuta una query en Athena y retorna un DataFrame."""
    response = athena.start_query_execution(
        QueryString=sql,
        QueryExecutionContext={'Database': database},
        ResultConfiguration={
            'OutputLocation': f's3://{output_bucket}/athena-results/',
            'EncryptionConfiguration': {'EncryptionOption': 'SSE_S3'}
        },
        WorkGroup='primary'
    )
    
    execution_id = response['QueryExecutionId']
    
    # Esperar a que termine
    while True:
        result = athena.get_query_execution(QueryExecutionId=execution_id)
        state = result['QueryExecution']['Status']['State']
        
        if state == 'SUCCEEDED':
            break
        elif state in ['FAILED', 'CANCELLED']:
            raise Exception(f"Query failed: {result['QueryExecution']['Status']}")
        
        time.sleep(2)
    
    # Leer resultados desde S3
    output_path = result['QueryExecution']['ResultConfiguration']['OutputLocation']
    return pd.read_csv(output_path)

# Uso
df = run_athena_query(
    "SELECT customer_id, SUM(amount) FROM silver.orders WHERE year='2026' GROUP BY 1",
    database='silver',
    output_bucket='my-query-results'
)

Stack de Datos GCP 2026

# ── GOOGLE CLOUD STORAGE ─────────────────────────────────────────────────
from google.cloud import storage

storage_client = storage.Client()
bucket = storage_client.bucket('my-data-lake-gcs')

# Upload
blob = bucket.blob('raw/orders/2026/01/01/data.parquet')
blob.upload_from_filename('local_data.parquet')

# Con metadata y content type
blob.content_type = 'application/octet-stream'
blob.metadata = {'source': 'orders-api', 'version': '2'}
blob.patch()

# Leer Parquet directamente con GCSFileSystem
import gcsfs
fs = gcsfs.GCSFileSystem()
df = pd.read_parquet('gcs://my-data-lake-gcs/silver/orders/', filesystem=fs)

# ── BIGQUERY: Analytics Serverless ────────────────────────────────────────
from google.cloud import bigquery

bq = bigquery.Client()

# Insertar desde DataFrame
job_config = bigquery.LoadJobConfig(
    schema=[
        bigquery.SchemaField("order_id", "INTEGER"),
        bigquery.SchemaField("amount", "FLOAT"),
        bigquery.SchemaField("created_at", "TIMESTAMP"),
    ],
    write_disposition="WRITE_APPEND",
    time_partitioning=bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,
        field="created_at"
    ),
    clustering_fields=["customer_id", "status"]
)

job = bq.load_table_from_dataframe(df, "project.dataset.orders", job_config=job_config)
job.result()  # Esperar que termine

# Ejecutar query y obtener resultado
query = """
    SELECT customer_id, SUM(amount) AS total
    FROM `project.dataset.orders`
    WHERE DATE(created_at) = @run_date
    GROUP BY customer_id
"""
job_config_q = bigquery.QueryJobConfig(
    query_parameters=[bigquery.ScalarQueryParameter("run_date", "DATE", "2026-01-15")]
)
df_result = bq.query(query, job_config=job_config_q).to_dataframe()

# ── APACHE BEAM con Dataflow ───────────────────────────────────────────────
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions

options = PipelineOptions()
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = 'my-project'
google_cloud_options.region = 'us-central1'
google_cloud_options.staging_location = 'gs://my-bucket/staging'
google_cloud_options.temp_location = 'gs://my-bucket/temp'
options.view_as(beam.options.pipeline_options.StandardOptions).runner = 'DataflowRunner'

with beam.Pipeline(options=options) as p:
    orders = (
        p
        | 'ReadFromPubSub' >> beam.io.ReadFromPubSub(topic='projects/myproject/topics/orders')
        | 'ParseJSON' >> beam.Map(json.loads)
        | 'FilterValid' >> beam.Filter(lambda x: x.get('amount', 0) > 0)
        | 'TransformOrders' >> beam.Map(lambda x: {**x, 'revenue_cat': 'large' if x['amount'] > 500 else 'small'})
        | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
            'my-project:analytics.orders_stream',
            schema='order_id:INTEGER,amount:FLOAT,revenue_cat:STRING',
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
        )
    )

Stack de Datos Azure 2026