Francisco Escobar

Posted on Jul 15

🏗️ Proyecto Data Lake & Analytics en AWS: Guía Completa para Principiantes

#aws #tutorial #bigdata #datalake

📋 Introducción al Proyecto

Un Data Lake es como un gran almacén donde puedes guardar todos tus datos sin importar su formato (archivos, bases de datos, logs, etc.). En este proyecto aprenderás a construir un sistema completo de análisis de datos usando servicios de AWS.

🎯 ¿Qué vamos a construir?

Un Data Lake en Amazon S3 con datos organizados
Procesos automáticos de transformación de datos
Consultas SQL sobre grandes volúmenes de datos
Dashboards interactivos para visualizar insights

🛠️ Servicios AWS que usaremos:

Amazon S3: Almacenamiento del Data Lake
AWS Glue: Procesamiento y transformación de datos (ETL)
Amazon Athena: Consultas SQL serverless
Amazon QuickSight: Visualización y dashboards

📊 Etapa 1: Preparación y Configuración Inicial

Paso 1.1: Configurar tu cuenta AWS

# Instalar AWS CLI
pip install awscli

# Configurar credenciales
aws configure

Paso 1.2: Crear estructura de carpetas del proyecto

data-lake-project/
├── raw-data/           # Datos originales
├── processed-data/     # Datos transformados
├── glue-scripts/       # Scripts de ETL
├── athena-queries/     # Consultas SQL
└── quicksight-config/  # Configuración de dashboards

Paso 1.3: Preparar datos de ejemplo

Para este tutorial, usaremos datos de ventas simulados:

Archivo: sales_data.csv
Campos: date, product_id, category, quantity, price, customer_id, region

🪣 Etapa 2: Construir el Data Lake en S3

Paso 2.1: Crear buckets S3

# Crear bucket principal del data lake
aws s3 mb s3://mi-data-lake-proyecto

# Crear estructura de carpetas
aws s3api put-object --bucket mi-data-lake-proyecto --key raw-data/
aws s3api put-object --bucket mi-data-lake-proyecto --key processed-data/
aws s3api put-object --bucket mi-data-lake-proyecto --key glue-scripts/

Paso 2.2: Implementar particionado

El particionado mejora el rendimiento organizando datos por fecha:

s3://mi-data-lake-proyecto/raw-data/
├── year=2024/
│   ├── month=01/
│   │   ├── day=01/
│   │   │   └── sales_data.parquet
│   │   └── day=02/
│   └── month=02/
└── year=2023/

Paso 2.3: Subir datos iniciales

# Subir archivo de datos
aws s3 cp sales_data.csv s3://mi-data-lake-proyecto/raw-data/year=2024/month=01/day=01/

💡 Tip para principiantes:

El particionado es como organizar archivos en carpetas por fecha. Esto hace que las consultas sean más rápidas porque solo buscan en las carpetas relevantes.

🔧 Etapa 3: Configurar AWS Glue para ETL

Paso 3.1: Crear un Glue Crawler

Un Crawler es como un robot que examina tus datos y crea un catálogo automáticamente.

# Configuración del Crawler via boto3
import boto3

glue_client = boto3.client('glue')

crawler_config = {
    'Name': 'sales-data-crawler',
    'Role': 'arn:aws:iam::tu-account:role/GlueServiceRole',
    'DatabaseName': 'sales_database',
    'Targets': {
        'S3Targets': [
            {
                'Path': 's3://mi-data-lake-proyecto/raw-data/'
            }
        ]
    }
}

glue_client.create_crawler(**crawler_config)

Paso 3.2: Crear un job ETL de Glue

# Script ETL básico (glue-scripts/transform_sales_data.py)
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

# Inicializar contexto
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

# Leer datos del Data Lake
datasource = glueContext.create_dynamic_frame.from_catalog(
    database="sales_database",
    table_name="raw_data"
)

# Transformaciones
# 1. Limpiar datos nulos
cleaned_data = DropNullFields.apply(frame=datasource)

# 2. Agregar columna de revenue
def calculate_revenue(rec):
    rec["revenue"] = rec["quantity"] * rec["price"]
    return rec

transformed_data = Map.apply(frame=cleaned_data, f=calculate_revenue)

# 3. Escribir datos procesados
glueContext.write_dynamic_frame.from_options(
    frame=transformed_data,
    connection_type="s3",
    connection_options={
        "path": "s3://mi-data-lake-proyecto/processed-data/"
    },
    format="parquet",
    format_options={
        "compression": "snappy"
    }
)

job.commit()

Paso 3.3: Ejecutar el job ETL

# Ejecutar job via CLI
aws glue start-job-run --job-name transform-sales-data

🔍 ¿Qué hace el ETL?

Extract: Extrae datos del S3 raw-data
Transform: Limpia y calcula nuevas columnas (revenue)
Load: Guarda los datos transformados en processed-data

🔍 Etapa 4: Consultas con Amazon Athena

Paso 4.1: Configurar Athena

-- Crear base de datos
CREATE DATABASE IF NOT EXISTS sales_analytics;

-- Crear tabla externa apuntando a S3
CREATE EXTERNAL TABLE sales_analytics.processed_sales (
    date string,
    product_id string,
    category string,
    quantity int,
    price decimal(10,2),
    customer_id string,
    region string,
    revenue decimal(10,2)
)
PARTITIONED BY (
    year int,
    month int,
    day int
)
STORED AS PARQUET
LOCATION 's3://mi-data-lake-proyecto/processed-data/';

Paso 4.2: Consultas de análisis

-- Consulta 1: Top 10 productos por revenue
SELECT 
    product_id,
    SUM(revenue) as total_revenue,
    COUNT(*) as total_sales
FROM sales_analytics.processed_sales
WHERE year = 2024
GROUP BY product_id
ORDER BY total_revenue DESC
LIMIT 10;

-- Consulta 2: Ventas por región y mes
SELECT 
    region,
    month,
    SUM(revenue) as monthly_revenue,
    AVG(price) as avg_price
FROM sales_analytics.processed_sales
WHERE year = 2024
GROUP BY region, month
ORDER BY region, month;

-- Consulta 3: Análisis de categorías
SELECT 
    category,
    COUNT(DISTINCT customer_id) as unique_customers,
    SUM(quantity) as total_quantity,
    SUM(revenue) as total_revenue
FROM sales_analytics.processed_sales
GROUP BY category;

📊 Ventajas de Athena:

No necesitas servidores (serverless)
Pagas solo por las consultas que ejecutas
Conecta directamente con S3

📈 Etapa 5: Dashboards con QuickSight

Paso 5.1: Configurar QuickSight

Acceder a Amazon QuickSight
Crear cuenta (si es primera vez)
Otorgar permisos para acceder a S3 y Athena

Paso 5.2: Crear dataset

# Configuración del dataset via API
import boto3

quicksight = boto3.client('quicksight')

dataset_config = {
    'AwsAccountId': 'tu-account-id',
    'DataSetId': 'sales-dataset',
    'Name': 'Sales Analytics Dataset',
    'PhysicalTableMap': {
        'sales-table': {
            'RelationalTable': {
                'DataSourceArn': 'arn:aws:quicksight:region:account:datasource/athena-datasource',
                'Schema': 'sales_analytics',
                'Name': 'processed_sales'
            }
        }
    }
}

quicksight.create_data_set(**dataset_config)

Paso 5.3: Crear visualizaciones

Dashboard 1: Overview de Ventas

Gráfico de líneas: Revenue por mes
Gráfico de barras: Top 10 productos
Mapa: Ventas por región
KPI cards: Total revenue, Total orders, Avg order value

Dashboard 2: Análisis de Productos

Treemap: Revenue por categoría
Scatter plot: Precio vs Cantidad vendida
Tabla: Detalle de productos con filtros

Dashboard 3: Análisis de Clientes

Histograma: Distribución de clientes por revenue
Gráfico de área: Clientes activos por mes
Funnel: Customer journey

Paso 5.4: Configurar filtros y parámetros

{
  "filters": [
    {
      "name": "date_filter",
      "type": "date_range",
      "default": "last_30_days"
    },
    {
      "name": "region_filter",
      "type": "multi_select",
      "values": ["North", "South", "East", "West"]
    },
    {
      "name": "category_filter",
      "type": "dropdown",
      "values": ["Electronics", "Clothing", "Books"]
    }
  ]
}

🔄 Etapa 6: Automatización y Monitoreo

Paso 6.1: Automatizar el pipeline con EventBridge

# Configurar trigger automático
import boto3

events_client = boto3.client('events')

rule_config = {
    'Name': 'daily-etl-trigger',
    'ScheduleExpression': 'cron(0 2 * * ? *)',  # Diario a las 2 AM
    'Description': 'Trigger ETL job daily',
    'State': 'ENABLED'
}

events_client.put_rule(**rule_config)

# Asociar con Glue job
events_client.put_targets(
    Rule='daily-etl-trigger',
    Targets=[
        {
            'Id': '1',
            'Arn': 'arn:aws:glue:region:account:job/transform-sales-data',
            'RoleArn': 'arn:aws:iam::account:role/EventBridgeRole'
        }
    ]
)

Paso 6.2: Configurar alertas con CloudWatch

# Crear alarma para job failures
cloudwatch = boto3.client('cloudwatch')

alarm_config = {
    'AlarmName': 'glue-job-failure-alarm',
    'ComparisonOperator': 'GreaterThanThreshold',
    'EvaluationPeriods': 1,
    'MetricName': 'glue.driver.aggregate.numFailedTasks',
    'Namespace': 'AWS/Glue',
    'Period': 300,
    'Statistic': 'Sum',
    'Threshold': 0,
    'ActionsEnabled': True,
    'AlarmActions': [
        'arn:aws:sns:region:account:glue-alerts'
    ]
}

cloudwatch.put_metric_alarm(**alarm_config)

💰 Etapa 7: Optimización de Costos

Paso 7.1: Estrategias de storage en S3

# Configurar lifecycle policies
s3_client = boto3.client('s3')

lifecycle_config = {
    'Rules': [
        {
            'ID': 'archive-old-data',
            'Status': 'Enabled',
            'Filter': {'Prefix': 'raw-data/'},
            'Transitions': [
                {
                    'Days': 30,
                    'StorageClass': 'STANDARD_IA'
                },
                {
                    'Days': 90,
                    'StorageClass': 'GLACIER'
                }
            ]
        }
    ]
}

s3_client.put_bucket_lifecycle_configuration(
    Bucket='mi-data-lake-proyecto',
    LifecycleConfiguration=lifecycle_config
)

Paso 7.2: Optimizar consultas Athena

-- Usar particiones en WHERE
SELECT * FROM sales_analytics.processed_sales
WHERE year = 2024 AND month = 1;  -- ✅ Eficiente

-- Evitar SELECT *
SELECT product_id, SUM(revenue)    -- ✅ Específico
FROM sales_analytics.processed_sales
GROUP BY product_id;

-- Usar formato columnar (Parquet)
-- Ya configurado en nuestro ETL ✅

🚀 Etapa 8: Mejoras Avanzadas

Paso 8.1: Implementar Data Quality

# Glue Data Quality script
import boto3
from awsglue.context import GlueContext

def validate_data_quality(df):
    """Validar calidad de los datos"""

    # 1. Verificar duplicados
    duplicate_count = df.groupBy("product_id", "date").count().filter("count > 1").count()

    # 2. Verificar valores nulos críticos
    null_checks = {
        'product_id': df.filter(df.product_id.isNull()).count(),
        'price': df.filter(df.price.isNull()).count(),
        'quantity': df.filter(df.quantity.isNull()).count()
    }

    # 3. Verificar rangos de valores
    price_outliers = df.filter((df.price < 0) | (df.price > 10000)).count()

    return {
        'duplicates': duplicate_count,
        'null_values': null_checks,
        'price_outliers': price_outliers
    }

Paso 8.2: Implementar Data Lineage

# Tracking de linaje de datos
data_lineage = {
    'source': 's3://mi-data-lake-proyecto/raw-data/',
    'transformations': [
        {
            'step': 'clean_nulls',
            'description': 'Remove null values',
            'timestamp': '2024-01-01T10:00:00Z'
        },
        {
            'step': 'calculate_revenue',
            'description': 'Add revenue column',
            'timestamp': '2024-01-01T10:05:00Z'
        }
    ],
    'destination': 's3://mi-data-lake-proyecto/processed-data/'
}