Lab 62 — Data lake em camadas: bronze, prata, ouro

# Quantos objetos existem, e quantos "tipos" de nome de arquivo aparecem?
aws s3api list-objects-v2 --bucket cadencia-lake \
  --query "length(Contents)" --output text
# 41312

aws s3api list-objects-v2 --bucket cadencia-lake \
  --query "Contents[].Key" --output text \
  | tr "\t" "\n" | sed -E "s/[0-9]+/#/g" | sort -u | wc -l
# 187 padroes de nome distintos, nenhum deles documentado

{
  "source": "cadencia.lake.validador",
  "detail-type": "Arquivo movido para quarentena",
  "detail": {
    "bucket": "cadencia-lake",
    "chaveOriginal": "bronze/pedidos/ano=2026/mes=08/arquivo-sem-particao-de-dia.parquet",
    "chaveQuarentena": "quarentena/pedidos/ano=2026/mes=08/arquivo-sem-particao-de-dia.parquet",
    "motivo": "caminho fora do contrato de bronze",
    "detectadoEm": "2026-08-07T09:14:02.118Z"
  }
}

# lake.tf — um bucket, tres prefixos, cada um com o ciclo de vida da sua camada

resource "aws_s3_bucket" "lake" {
  bucket = "${var.projeto}-lake"
}

# Bronze e imutavel por convencao de aplicacao (o job nunca sobrescreve uma
# partição existente), reforcada aqui por versionamento: se algo sobrescrever
# por engano, a versao anterior continua recuperavel. Nao e Object Lock —
# Object Lock impediria ate o dono apagar em caso de erro operacional legitimo,
# e a equipe de duas pessoas preferiu o botao de emergencia a rigidez total.
resource "aws_s3_bucket_versioning" "lake" {
  bucket = aws_s3_bucket.lake.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_server_side_encryption_configuration" "lake" {
  bucket = aws_s3_bucket.lake.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = aws_kms_key.lake.arn
    }
    bucket_key_enabled = true
  }
}

# Bloqueia acesso publico nas quatro dimensoes. Lake com dado de pedido de
# cliente nao tem cenario legitimo de leitura publica.
resource "aws_s3_bucket_public_access_block" "lake" {
  bucket                  = aws_s3_bucket.lake.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

# Cada camada tem um ciclo de vida DIFERENTE, porque cada uma serve a um uso
# diferente. E aqui que o contrato de camada vira configuracao, nao so convencao.
resource "aws_s3_bucket_lifecycle_configuration" "lake" {
  bucket = aws_s3_bucket.lake.id

  # BRONZE: existe para reprocessar, nao para consulta do dia a dia. Fica
  # barato rapido, porque quase nunca e lido depois que prata foi gerada.
  rule {
    id     = "bronze-esfria-rapido"
    status = "Enabled"
    filter { prefix = "bronze/" }
    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }
    transition {
      days          = 90
      storage_class = "GLACIER"
    }
  }

  # PRATA: consultada por investigacao, com menos frequencia que ouro mas mais
  # que bronze. Esfria, mas mais devagar.
  rule {
    id     = "prata-esfria-devagar"
    status = "Enabled"
    filter { prefix = "prata/" }
    transition {
      days          = 90
      storage_class = "STANDARD_IA"
    }
  }

  # OURO: e o que o BI consulta todo dia. Fica em Standard indefinidamente —
  # o volume e pequeno porque ja e agregado, entao o custo de nao esfriar e baixo.
  rule {
    id     = "ouro-permanece-quente"
    status = "Enabled"
    filter { prefix = "ouro/" }
    abort_incomplete_multipart_upload {
      days_after_initiation = 7
    }
  }

  # QUARENTENA: arquivo que quebrou o contrato. Ninguem investiga depois de
  # 60 dias — se ninguem olhou, nao vai olhar, e expirar evita acumulo silencioso.
  rule {
    id     = "quarentena-expira"
    status = "Enabled"
    filter { prefix = "quarentena/" }
    expiration {
      days = 60
    }
  }
}

resource "aws_kms_key" "lake" {
  description         = "Chave do data lake da Cadência"
  enable_key_rotation = true
}

output "bucket_lake" {
  value       = aws_s3_bucket.lake.id
  description = "Bucket unico com prefixos bronze/, prata/, ouro/ e quarentena/"
}

# catalogo-e-permissao.tf — tres bancos no Data Catalog, permissao por tabela/coluna

resource "aws_glue_catalog_database" "bronze" {
  name = "cadencia_bronze"
}

resource "aws_glue_catalog_database" "prata" {
  name = "cadencia_prata"
}

resource "aws_glue_catalog_database" "ouro" {
  name = "cadencia_ouro"
}

# Registra o bucket no Lake Formation. A partir daqui, acesso a qualquer
# recurso registrado passa pelo LF — mesmo que o papel IAM tenha permissao de
# S3, o LF e quem decide se a consulta do Athena enxerga a tabela.
resource "aws_lakeformation_resource" "lake" {
  arn = aws_s3_bucket.lake.arn
}

# Papel do engenheiro de dados: pode investigar, entao alcanca bronze e prata.
resource "aws_iam_role" "engenheiro_dados" {
  name               = "${var.projeto}-engenheiro-dados"
  assume_role_policy = data.aws_iam_policy_document.assume_por_identity_center.json
}

# Papel do analista de negocio: so precisa do agregado.
resource "aws_iam_role" "analista_negocio" {
  name               = "${var.projeto}-analista-negocio"
  assume_role_policy = data.aws_iam_policy_document.assume_por_identity_center.json
}

# NAO ha "s3:GetObject" nestes papeis para o bucket do lake. Essa e a decisao
# central: o acesso ao DADO vem do Lake Formation (credential vending), nao de
# uma policy de S3. Um papel com s3:GetObject direto no bucket contornaria o
# LF por completo — e essa e a falha da secao "quebrar de proposito".

# Engenheiro de dados: SELECT em bronze inteiro e em prata inteiro.
resource "aws_lakeformation_permissions" "eng_bronze" {
  principal   = aws_iam_role.engenheiro_dados.arn
  permissions = ["SELECT", "DESCRIBE"]

  table {
    database_name = aws_glue_catalog_database.bronze.name
    wildcard       = true
  }
}

resource "aws_lakeformation_permissions" "eng_prata" {
  principal   = aws_iam_role.engenheiro_dados.arn
  permissions = ["SELECT", "DESCRIBE"]

  table {
    database_name = aws_glue_catalog_database.prata.name
    wildcard       = true
  }
}

# Analista de negocio: SELECT so em ouro. Sem acesso a prata nem bronze — nem
# por wildcard, nem por tabela nomeada.
resource "aws_lakeformation_permissions" "analista_ouro" {
  principal   = aws_iam_role.analista_negocio.arn
  permissions = ["SELECT", "DESCRIBE"]

  table {
    database_name = aws_glue_catalog_database.ouro.name
    wildcard       = true
  }
}

# Restricao por COLUNA: mesmo dentro de prata, o papel de auditoria externa
# (existe so para este exemplo) nao ve a coluna de CPF. E a prova de que a
# granularidade e de coluna, nao so de tabela.
resource "aws_lakeformation_permissions" "auditoria_prata_sem_cpf" {
  principal   = aws_iam_role.auditoria_externa.arn
  permissions = ["SELECT"]

  table_with_columns {
    database_name    = aws_glue_catalog_database.prata.name
    name              = "pedidos"
    excluded_column_names = ["cpf_cliente"]
  }
}

# Papel de execucao dos jobs Glue: PRECISA de s3:GetObject/PutObject no bucket,
# porque quem grava a partir do dado bruto e o job, nao uma consulta do Athena.
# Este e o unico lugar em que acesso direto ao S3 e correto.
data "aws_iam_policy_document" "execucao_job" {
  statement {
    effect  = "Allow"
    actions = ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
    resources = [
      aws_s3_bucket.lake.arn,
      "${aws_s3_bucket.lake.arn}/*",
    ]
  }
  statement {
    effect  = "Allow"
    actions = ["glue:GetTable", "glue:UpdateTable", "glue:GetDatabase", "glue:CreateTable"]
    resources = ["*"]
    # As quatro acoes de catalogo do Glue nao aceitam ARN de tabela individual
    # neste conjunto — sao operacoes de metadado da conta/regiao. O acesso
    # granular ao DADO continua sendo o do Lake Formation, acima.
  }
}

# bronze_para_prata.py — job do Glue (PySpark). Le uma particao, escreve outra.
#
# Roda com job bookmark ligado: cada execucao processa so os arquivos novos
# daquela particao desde a ultima vez. Reexecutar o MESMO dia nao duplica,
# porque a escrita usa partitionOverwriteMode=dynamic: substitui so a
# particao gravada, nunca o restante da tabela.

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext
from pyspark.sql import functions as F
from pyspark.sql.window import Window

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

# Necessario para o INSERT OVERWRITE atingir so a particao escrita.
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

particao = args['data_particao']  # ex.: '2026-08-07'

bruto = glueContext.create_dynamic_frame.from_catalog(
    database="cadencia_bronze",
    table_name="pedidos",
    push_down_predicate=f"data_particao = '{particao}'",
    transformation_ctx="bruto_bookmark",  # ancora do job bookmark
).toDF()

# Deduplicacao: a captura continua pode entregar o MESMO pedido mais de uma
# vez (reenvio apos falha de rede). Mantem a versao mais recente por chave.
janela = Window.partitionBy("pedido_id").orderBy(F.col("capturado_em").desc())
limpo = (
    bruto
    .withColumn("linha", F.row_number().over(janela))
    .filter(F.col("linha") == 1)
    .drop("linha")
    # Tipagem explicita: bronze guarda como chegou (string), prata corrige.
    .withColumn("valor_total", F.col("valor_total").cast("decimal(12,2)"))
    .withColumn("criado_em", F.to_timestamp("criado_em"))
    # Registro sem chave primaria nao entra em prata — vai para quarentena
    # por um segundo job, fora do escopo deste laboratorio.
    .filter(F.col("pedido_id").isNotNull())
    .withColumn("data_particao", F.lit(particao))
)

saida = glueContext.create_dynamic_frame.from_dataframe(limpo, glueContext, "saida")

# updateBehavior=UPDATE_IN_DATABASE: o proprio job mantem o catalogo de prata
# atualizado. Nao existe crawler agendado sobre prata neste desenho.
glueContext.write_dynamic_frame.from_catalog(
    frame=saida,
    database="cadencia_prata",
    table_name="pedidos",
    additional_options={
        "partitionKeys": ["data_particao"],
        "updateBehavior": "UPDATE_IN_DATABASE",
    },
)

job.commit()

# prata_para_ouro.py — job do Glue (PySpark). So agrega; nunca le bronze.

import sys
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext
from pyspark.sql import functions as F

args = getResolvedOptions(sys.argv, ['JOB_NAME', 'data_particao'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

particao = args['data_particao']

prata = glueContext.create_dynamic_frame.from_catalog(
    database="cadencia_prata",
    table_name="pedidos",
    push_down_predicate=f"data_particao = '{particao}'",
    transformation_ctx="prata_bookmark",
).toDF()

# O UNICO ponto do pipeline em que a granularidade muda de evento para
# resumo. E por isso que ouro existe separado de prata: sao duas perguntas
# diferentes ("o que aconteceu" vs "quanto vendeu cada loja").
agregado = (
    prata
    .groupBy("loja_id", "data_particao")
    .agg(
        F.count("pedido_id").alias("qtd_pedidos"),
        F.sum("valor_total").alias("valor_total_dia"),
        F.avg("valor_total").alias("ticket_medio"),
    )
)

saida = glueContext.create_dynamic_frame.from_dataframe(agregado, glueContext, "saida")

glueContext.write_dynamic_frame.from_catalog(
    frame=saida,
    database="cadencia_ouro",
    table_name="vendas_diarias_por_loja",
    additional_options={
        "partitionKeys": ["data_particao"],
        "updateBehavior": "UPDATE_IN_DATABASE",
    },
)

job.commit()

// ContratoBronzeValidator.cs — Lambda em .NET 8, acionada por evento de criacao
// de objeto no prefixo bronze/. E o unico ponto do pipeline que decide se um
// arquivo "conta" como dado de bronze ou vai para quarentena.

using Amazon.Lambda.Core;
using Amazon.Lambda.S3Events;
using Amazon.S3;
using Amazon.S3.Model;
using System.Text.RegularExpressions;

[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]

namespace Cadência.Lake.Validador;

public class Function
{
    // O contrato de bronze, em uma expressao: dominio conhecido, particionado
    // por ano/mes/dia, e extensao que o job sabe ler. Qualquer coisa fora
    // disto nao e "bronze invalido" — e simplesmente NAO bronze.
    private static readonly Regex CaminhoValido = new(
        @"^bronze/(pedidos|estoque|clientes)/ano=\d{4}/mes=\d{2}/dia=\d{2}/[\w-]+\.parquet$",
        RegexOptions.Compiled);

    private static readonly IAmazonS3 _s3 = new AmazonS3Client();

    public async Task FunctionHandler(S3Event evento, ILambdaContext contexto)
    {
        foreach (var registro in evento.Records)
        {
            var bucket = registro.S3.Bucket.Name;
            var chave = Uri.UnescapeDataString(registro.S3.Object.Key);

            var motivo = await Validar(bucket, chave);

            if (motivo is not null)
            {
                await MoverParaQuarentena(bucket, chave, motivo);
                contexto.Logger.LogWarning(
                    $"quarentena: {chave} — {motivo}");
                continue;
            }

            contexto.Logger.LogInformation($"contrato ok: {chave}");
        }
    }

    private async Task<string?> Validar(string bucket, string chave)
    {
        // 1. Caminho e nome. Nao abre o arquivo se o caminho ja reprova —
        //    e a verificacao mais barata, entao vem primeiro.
        if (!CaminhoValido.IsMatch(chave))
            return $"caminho fora do contrato de bronze: '{chave}'";

        // 2. Tamanho minimo. Arquivo de 0 byte e sintoma comum de escrita
        //    interrompida no meio (upload multipart que falhou parcialmente).
        var cabecalho = await _s3.GetObjectMetadataAsync(bucket, chave);
        if (cabecalho.ContentLength < 12) // menor Parquet valido tem magic bytes
            return $"arquivo suspeito de escrita incompleta: {cabecalho.ContentLength} bytes";

        // NAO verificado aqui: schema interno do Parquet. Abrir e ler o
        // rodape do arquivo custa uma leitura completa do objeto na Lambda,
        // e o job Glue ja reprova schema incompativel na proxima execucao.
        // Declarar esse limite é mais honesto que fingir uma validação de
        // schema que este laboratório não implementa.
        return null;
    }

    private async Task MoverParaQuarentena(string bucket, string chave, string motivo)
    {
        var destino = chave.Replace("bronze/", "quarentena/", StringComparison.Ordinal);

        await _s3.CopyObjectAsync(new CopyObjectRequest
        {
            SourceBucket = bucket,
            SourceKey = chave,
            DestinationBucket = bucket,
            DestinationKey = destino,
            MetadataDirective = S3MetadataDirective.REPLACE,
        });

        // O motivo fica no proprio objeto — quem for investigar a quarentena
        // nao precisa cruzar com log do CloudWatch para saber por que caiu ali.
        await _s3.PutObjectTaggingAsync(new PutObjectTaggingRequest
        {
            BucketName = bucket,
            Key = destino,
            Tagging = new Tagging
            {
                TagSet = [new Tag { Key = "motivo-quarentena", Value = motivo[..Math.Min(motivo.Length, 200)] }],
            },
        });

        await _s3.DeleteObjectAsync(bucket, chave);
    }
}

#!/usr/bin/env bash
set -euo pipefail
terraform apply -auto-approve

# Publica os dois jobs do Glue a partir do script local.
aws s3 cp bronze_para_prata.py "s3://$(terraform output -raw bucket_lake)/scripts/"
aws s3 cp prata_para_ouro.py "s3://$(terraform output -raw bucket_lake)/scripts/"

# Publica a Lambda validadora (build previo com dotnet lambda package).
aws lambda update-function-code \
  --function-name cadencia-contrato-bronze-validator \
  --zip-file fileb://bin/Release/net8.0/ContratoBronzeValidator.zip

echo "lake no ar; proxima etapa: provas.sh"

BUCKET=$(terraform output -raw bucket_lake)

# ── Prova 1: a estrutura existe, e cada camada tem dono de schema ────────────
aws s3 ls "s3://$BUCKET/" 
# Esperado: bronze/  prata/  ouro/  quarentena/  scripts/
aws glue get-tables --database-name cadencia_prata --query "TableList[].Name" --output text
# Esperado: pedidos — registrada PELO JOB, sem crawler agendado sobre prata

# ── Prova 2: bytes varridos, ouro contra bronze, mesma pergunta de negocio ───
aws athena start-query-execution \
  --query-string "SELECT sum(valor_total_dia) FROM cadencia_ouro.vendas_diarias_por_loja WHERE data_particao LIKE '2026-07%'" \
  --result-configuration OutputLocation=s3://$BUCKET/consultas/ \
  --query-execution-context Database=cadencia_ouro
# aws athena get-query-execution --query-execution-id <id> --query "QueryExecution.Statistics.DataScannedInBytes"
# Esperado: poucas centenas de MB. A mesma pergunta contra bronze sem filtro de
# particao varre gigabytes — e a diferenca e a prova da secao de custo.

# ── Prova 3: a negativa do Lake Formation, testada de verdade ────────────────
aws sts assume-role --role-arn "$(terraform output -raw papel_analista_negocio)" \
  --role-session-name teste-negativa --query "Credentials" --output json > /tmp/cred.json
# com essas credenciais, tente consultar cadencia_prata.pedidos no Athena
# Esperado: erro de permissao do Lake Formation, nao um resultado vazio

# ── Prova 4: reprocessar um dia nao duplica os outros ────────────────────────
aws athena start-query-execution --query-execution-context Database=cadencia_prata \
  --query-string "SELECT count(*) FROM pedidos" --result-configuration OutputLocation=s3://$BUCKET/consultas/
# anote o numero, rode o job bronze_para_prata de novo para o MESMO dia, repita a
# contagem. Esperado: numero identico. Se cresceu, o overwrite nao esta dynamic.

# ── Prova 5: contrato invalido cai em quarentena, nao no catalogo ────────────
echo "lixo" | aws s3 cp - "s3://$BUCKET/bronze/sem-particao.txt"
sleep 5
aws s3 ls "s3://$BUCKET/quarentena/" | grep sem-particao
# Esperado: o arquivo aparece em quarentena, com a tag motivo-quarentena, e NUNCA
# aparece em bronze/ nem em nenhuma tabela do catalogo.

# 1. Revogue as permissoes do Lake Formation ANTES de tentar remover o catalogo —
#    o Terraform normalmente resolve isso na ordem certa, mas se voce interrompeu
#    o apply no meio, pode ficar permissao orfa apontando para tabela que sumiu.
aws lakeformation list-permissions --query "PrincipalResourcePermissions" --output table

# 2. Esvazie o bucket ANTES do destroy: bucket com objeto (e com versionamento,
#    com VERSOES de objeto) faz o destroy falhar silenciosamente na aparencia — o
#    erro aparece, mas e facil de rodar de novo sem investigar por que falhou.
aws s3api list-object-versions --bucket cadencia-lake \
  --query "length(Versions)" --output text
aws s3 rm "s3://cadencia-lake/" --recursive
aws s3api list-object-versions --bucket cadencia-lake \
  --query "Versions[].{Key:Key,VersionId:VersionId}" --output json \
  | jq -c ".[]" | while read -r v; do
    key=$(echo "$v" | jq -r .Key); id=$(echo "$v" | jq -r .VersionId)
    aws s3api delete-object --bucket cadencia-lake --key "$key" --version-id "$id"
  done

# 3. Agora o Terraform consegue remover bucket, catalogo, permissoes e a chave KMS.
terraform destroy -auto-approve

# 4. CHAVE KMS: entra em exclusao PENDENTE, nao some na hora — continua existindo
#    (sem cobrar por uso, mas ocupando o limite de chaves da conta) pelo periodo
#    de espera configurado.
aws kms list-keys --query "Keys[?KeyId=='$(terraform output -raw kms_key_id 2>/dev/null)']"

# 5. Prova final: nada com o nome do projeto de pe.
aws resourcegroupstaggingapi get-resources \
  --tag-filters Key=Projeto,Values=cadencia-lake \
  --query "ResourceTagMappingList[].ResourceARN" --output table