Lab 65 — Catálogo e ETL: Glue Catalog, crawler, job

# DDL escrita a mao, com o schema de seis meses atras (5 colunas).
aws athena start-query-execution --query-string "
  CREATE EXTERNAL TABLE pedidos (
    pedido_id string, loja_id string, valor_centavos double,
    status string, criado_em string
  )
  STORED AS PARQUET
  LOCATION 's3://cadencia-lake/prata/pedidos/'" \
  --result-configuration OutputLocation=s3://cadencia-lake/atenas-resultados/

aws athena start-query-execution \
  --query-string "SELECT canal_pedido FROM pedidos LIMIT 1" \
  --result-configuration OutputLocation=s3://cadencia-lake/atenas-resultados/
# Esperado: COLUMN_NOT_FOUND. A coluna existe no arquivo Parquet ha duas
# semanas — so nao existe na definicao que Theo escreveu.

{
  "source": "aws.glue",
  "detail-type": "Glue Crawler State Change",
  "resources": ["arn:aws:glue:us-east-1:111122223333:crawler/cadencia-raw-pedidos"],
  "detail": {
    "crawlerName": "cadencia-raw-pedidos",
    "state": "Succeeded",
    "startTime": "2026-08-08T02:00:04Z"
  }
}

# catalogo.tf — o banco, o crawler, e a decisao explicita sobre schema mudando

resource "aws_glue_catalog_database" "lake" {
  name = "${var.projeto}_lake"
}

data "aws_iam_policy_document" "crawler_assume" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["glue.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "crawler" {
  name               = "${var.projeto}-crawler"
  assume_role_policy = data.aws_iam_policy_document.crawler_assume.json
}

# Politica gerenciada da AWS para o papel do Glue: cobre CloudWatch Logs e as
# chamadas de API do proprio servico. Nao cobre S3 — isso e sempre por conta
# de quem escreve o Terraform, e e a policy abaixo.
resource "aws_iam_role_policy_attachment" "crawler_gerenciada" {
  role       = aws_iam_role.crawler.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
}

data "aws_iam_policy_document" "crawler_leitura" {
  statement {
    effect    = "Allow"
    actions   = ["s3:GetObject", "s3:ListBucket"]
    resources = [
      "arn:aws:s3:::${var.bucket_lake}",
      "arn:aws:s3:::${var.bucket_lake}/bronze/pedidos/*",
    ]
  }
}

resource "aws_iam_role_policy" "crawler_leitura" {
  role   = aws_iam_role.crawler.id
  policy = data.aws_iam_policy_document.crawler_leitura.json
}

resource "aws_glue_crawler" "raw_pedidos" {
  name          = "${var.projeto}-raw-pedidos"
  role          = aws_iam_role.crawler.arn
  database_name = aws_glue_catalog_database.lake.name
  table_prefix  = "raw_"

  s3_target {
    path = "s3://${var.bucket_lake}/bronze/pedidos/"
  }

  # Sem "schedule" aqui de proposito: quem dispara o crawler e o gatilho
  # agendado do Workflow (workflow.tf), nao o crawler sozinho. Um agendamento
  # dentro do crawler E outro no Workflow seriam dois relogios discordando.

  schema_change_policy {
    # UPDATE_IN_DATABASE: aplica a mudanca de tipo automaticamente. A escolha
    # e "velocidade sobre revisao" — time pequeno, sem quem leia log toda
    # manha. Times maiores, com coluna sensivel, preferem LOG + aprovacao.
    update_behavior = "UPDATE_IN_DATABASE"

    # NUNCA DELETE_FROM_DATABASE aqui: se o s3_target apontar, ainda que por
    # engano temporario, para um prefixo vazio, essa opcao apagaria do
    # catalogo particoes cujos arquivos continuam intactos no S3. Ver o
    # callout de risco na secao de Construcao.
    delete_behavior = "DEPRECATE_IN_DATABASE"
  }

  configuration = jsonencode({
    Version  = 1.0
    Grouping = { TableGroupingPolicy = "CombineCompatibleSchemas" }
  })
}

output "banco_do_lake" {
  value       = aws_glue_catalog_database.lake.name
  description = "nome do banco no Glue Data Catalog"
}

# workflow.tf — a dependencia crawler -> job vira garantia do proprio Glue

resource "aws_glue_workflow" "pedidos" {
  name = "${var.projeto}-pedidos"
}

# Gatilho 1: agenda. So inicia o CRAWLER — nunca o job diretamente. O job so
# comeca depois que o catalogo, de fato, terminou de atualizar (gatilho 2).
resource "aws_glue_trigger" "agenda_diaria" {
  name          = "${var.projeto}-agenda-diaria"
  type          = "SCHEDULED"
  schedule      = "cron(0 2 * * ? *)" # 02:00 UTC
  workflow_name = aws_glue_workflow.pedidos.name

  actions {
    crawler_name = aws_glue_crawler.raw_pedidos.name
  }
}

# Gatilho 2: condicional. Observa o ESTADO do crawler dentro do mesmo
# Workflow — nao um horario fixo "3h, torcendo para o crawler ja ter
# acabado". Se o crawler demorar ou falhar, o job simplesmente nao comeca.
resource "aws_glue_trigger" "apos_catalogo_atualizado" {
  name          = "${var.projeto}-apos-catalogo"
  type          = "CONDITIONAL"
  workflow_name = aws_glue_workflow.pedidos.name

  predicate {
    conditions {
      crawler_name = aws_glue_crawler.raw_pedidos.name
      crawl_state  = "SUCCEEDED"
    }
  }

  actions {
    job_name = aws_glue_job.transformar_pedidos.name
    # Sem "--dt" aqui: o job calcula "ontem" sozinho quando o argumento nao
    # vem. Reprocessamento manual (reprocessar.tf/lambda) passa --dt explicito.
  }
}

# job.tf — o job idempotente, e o alerta que substitui silencio por sinal

data "aws_iam_policy_document" "job_assume" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["glue.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "job" {
  name               = "${var.projeto}-job-transformar-pedidos"
  assume_role_policy = data.aws_iam_policy_document.job_assume.json
}

resource "aws_iam_role_policy_attachment" "job_gerenciada" {
  role       = aws_iam_role.job.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
}

# O papel MAIS SENSIVEL deste laboratorio: ele tem s3:DeleteObject, porque a
# idempotencia exige apagar a particao antes de escrever. O escopo por
# PREFIXO e a unica coisa que impede um bug no calculo de "dt" (fuso horario,
# por exemplo) de apagar dado fora do que era pretendido. Ver o callout de
# risco na secao de Seguranca.
data "aws_iam_policy_document" "job_leitura_e_escrita" {
  statement {
    effect    = "Allow"
    actions   = ["s3:GetObject", "s3:ListBucket"]
    resources = [
      "arn:aws:s3:::${var.bucket_lake}",
      "arn:aws:s3:::${var.bucket_lake}/bronze/pedidos/*",
    ]
  }
  statement {
    effect    = "Allow"
    actions   = ["s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListBucket"]
    resources = [
      "arn:aws:s3:::${var.bucket_lake}",
      "arn:aws:s3:::${var.bucket_lake}/prata/pedidos/*",
    ]
  }
}

resource "aws_iam_role_policy" "job_leitura_e_escrita" {
  role   = aws_iam_role.job.id
  policy = data.aws_iam_policy_document.job_leitura_e_escrita.json
}

resource "aws_glue_job" "transformar_pedidos" {
  name     = "${var.projeto}-transformar-pedidos"
  role_arn = aws_iam_role.job.arn

  glue_version      = "4.0"
  worker_type       = "G.1X"
  number_of_workers = 4

  command {
    script_location = "s3://${var.bucket_scripts}/jobs/transformar_pedidos.py"
    python_version   = "3"
  }

  default_arguments = {
    # Evita RELER arquivo ja processado entre execucoes agendadas. Isto NAO
    # e a defesa contra duplicar dado num rerun manual do mesmo dia — essa
    # defesa e a sobrescrita de particao, dentro do script.
    "--job-bookmark-option"             = "job-bookmark-enable"
    "--enable-continuous-cloudwatch-log" = "true"
    "--enable-metrics"                  = "true"
    "--TempDir"                         = "s3://${var.bucket_scripts}/tmp/"
  }

  # Sem retry automatico: reexecutar sozinho sobre uma falha nao investigada
  # so multiplicaria a mesma janela de risco (particao vazia). Rerun manual,
  # depois de olhar o log, e mais seguro — e e seguro porque o job e idempotente.
  max_retries = 0
  timeout     = 30
}

resource "aws_sns_topic" "alertas" {
  name = "${var.projeto}-alertas-pipeline"
}

resource "aws_cloudwatch_event_rule" "job_falhou" {
  name        = "${var.projeto}-job-pedidos-falhou"
  description = "Glue Job State Change = FAILED ou TIMEOUT para o job de pedidos"

  event_pattern = jsonencode({
    source      = ["aws.glue"]
    detail-type = ["Glue Job State Change"]
    detail = {
      jobName = [aws_glue_job.transformar_pedidos.name]
      state   = ["FAILED", "TIMEOUT", "ERROR"]
    }
  })
}

resource "aws_cloudwatch_event_target" "avisa_time" {
  rule = aws_cloudwatch_event_rule.job_falhou.name
  arn  = aws_sns_topic.alertas.arn
}

# transformar_pedidos.py — le pelo catalogo, escreve com overwrite de particao

import sys
from datetime import datetime, timedelta, timezone

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

BUCKET = "cadencia-lake"
DATABASE = "cadencia_lake"
RAW_TABLE = "raw_pedidos"
PARTICAO = "dt"

args = getResolvedOptions(sys.argv, ["JOB_NAME"])

# --dt e OPCIONAL: o gatilho agendado do Workflow nao passa nada, e o job
# calcula "ontem" sozinho. Reprocessamento manual (ver secao de construcao
# do utilitario em C#) passa --dt explicito para escolher outro dia.
data_ref = None
for i, arg in enumerate(sys.argv):
    if arg == "--dt" and i + 1 < len(sys.argv):
        data_ref = sys.argv[i + 1]
if data_ref is None:
    data_ref = (datetime.now(timezone.utc).date() - timedelta(days=1)).isoformat()

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# 1. Le a particao do dia PELO CATALOGO — nao por um caminho decorado no
#    script. push_down_predicate filtra por PARTICAO JA CATALOGADA: se o
#    crawler nao rodou hoje, a particao nao existe no catalogo e a leitura
#    volta vazia, mesmo que o arquivo ja esteja fisicamente no S3.
bruto = glueContext.create_dynamic_frame.from_catalog(
    database=DATABASE,
    table_name=RAW_TABLE,
    push_down_predicate=f"{PARTICAO} = '{data_ref}'",
    transformation_ctx="fonte_raw_pedidos",  # chave do job bookmark
)

if bruto.count() == 0:
    print(f"nenhum arquivo novo para {PARTICAO}={data_ref}; nada a fazer")
    job.commit()
    sys.exit(0)

# 2. Transformacao com tipo EXPLICITO. A inferencia ja aconteceu no crawler;
#    aqui so convertemos para o tipo que a tabela prata deve ter.
df = bruto.toDF()
prata = (
    df.withColumn("valor_centavos", F.col("valor_centavos").cast("bigint"))
      .withColumn("criado_em", F.to_timestamp("criado_em"))
      .withColumn(PARTICAO, F.lit(data_ref))
      # dropDuplicates aqui protege contra o MESMO arquivo de origem reenviado
      # com o mesmo pedido_id — NAO e a defesa contra rodar o job duas vezes.
      # Essa defesa e o passo 3, abaixo.
      .dropDuplicates(["pedido_id"])
)

# 3. IDEMPOTENCIA REAL: apaga a particao de destino inteira antes de escrever.
#    O escritor do Glue ACRESCENTA arquivo por padrao — sem este passo, rodar
#    o job duas vezes sobre o mesmo dia duplicaria cada linha da particao.
#
#    ATENCAO: a janela entre apagar e escrever NAO e atomica. Se o job morrer
#    aqui, a particao fica vazia ate a proxima execucao bem-sucedida. E um
#    risco aceito e documentado — ver o callout na secao de construcao.
s3 = boto3.client("s3")
prefixo_destino = f"prata/pedidos/{PARTICAO}={data_ref}/"
paginador = s3.get_paginator("list_objects_v2")
chaves = [
    obj["Key"]
    for pagina in paginador.paginate(Bucket=BUCKET, Prefix=prefixo_destino)
    for obj in pagina.get("Contents", [])
]
if chaves:
    s3.delete_objects(Bucket=BUCKET, Delete={"Keys": [{"Key": k} for k in chaves]})

saida = DynamicFrame.fromDF(prata, glueContext, "saida")

# 4. Grava E atualiza o catalogo no MESMO passo, sem precisar de um segundo
#    crawler. useGlueParquetWriter=True e OBRIGATORIO para o catalogo aceitar
#    atualizacao de schema em Parquet — sem ele, a chamada grava o arquivo e
#    ignora silenciosamente a parte de atualizar o catalogo.
sink = glueContext.getSink(
    connection_type="s3",
    path=f"s3://{BUCKET}/prata/pedidos/",
    enableUpdateCatalog=True,
    updateBehavior="UPDATE_IN_DATABASE",
    partitionKeys=[PARTICAO],
)
sink.setFormat("parquet", useGlueParquetWriter=True)
sink.setCatalogInfo(catalogDatabase=DATABASE, catalogTableName="prata_pedidos")
sink.writeFrame(saida)

job.commit()

// Reprocessar.cs — reprocessar um dia e seguro PORQUE o job e idempotente

using Amazon.Glue;
using Amazon.Glue.Model;
using Amazon.Lambda.Core;

namespace Cadencia.Operacao;

public record ReprocessarPedido(string Dia); // "2026-08-07", validado antes de chegar aqui

public class ReprocessarHandler
{
    // Um so cliente por invocacao fria; reutilizado entre chamadas quentes.
    private static readonly AmazonGlueClient _glue = new();

    // Exposta via API Gateway com autorizacao restrita ao time de dados —
    // este endpoint chama StartJobRun, e por isso nao pode ser publico.
    public async Task<string> ReprocessarAsync(ReprocessarPedido pedido, ILambdaContext contexto)
    {
        // Nao ha protecao extra de "clique duplo" aqui, e e proposital: a
        // protecao E o job. Chamar isto duas vezes para o MESMO dia
        // sobrescreve a MESMA particao duas vezes — o resultado final e
        // identico, e essa e a garantia que a secao de provas mediu.
        var resposta = await _glue.StartJobRunAsync(new StartJobRunRequest
        {
            JobName = "cadencia-transformar-pedidos",
            Arguments = new Dictionary<string, string> { ["--dt"] = pedido.Dia },
        });

        contexto.Logger.LogInformation(
            $"reprocessamento iniciado para dt={pedido.Dia}: JobRunId={resposta.JobRunId}");

        return resposta.JobRunId;
    }
}

# provas.sh — cinco medicoes; nenhuma conclusao vem de "parece que atualizou"

PROJETO=cadencia; BANCO=cadencia_lake; DIA=2026-08-07

# ── Prova 1: o Workflow roda ponta a ponta e a particao fica com N linhas ────
aws glue start-workflow-run --name "${PROJETO}-pedidos"
aws glue get-workflow --name "${PROJETO}-pedidos" --include-graph \
  --query 'Workflow.LastRun.Statistics'
# Esperado: TotalActions == SucceededActions, e nenhuma FailedAction.

N1=$(aws athena start-query-execution \
  --query-string "SELECT COUNT(*) FROM ${BANCO}.prata_pedidos WHERE dt = '${DIA}'" \
  --result-configuration OutputLocation=s3://cadencia-lake/atenas-resultados/ \
  --query 'QueryExecutionId' --output text)
echo "guarde N1 (contagem apos a primeira execucao)"

# ── Prova 2: rodar o JOB de novo para o MESMO dia nao duplica nada ───────────
aws glue start-job-run --job-name "${PROJETO}-transformar-pedidos" \
  --arguments "{\"--dt\":\"${DIA}\"}"
aws glue wait job-run-complete --job-name "${PROJETO}-transformar-pedidos" \
  --run-id "$(aws glue get-job-runs --job-name ${PROJETO}-transformar-pedidos \
              --query 'JobRuns[0].Id' --output text)"
N2=$(aws athena start-query-execution \
  --query-string "SELECT COUNT(*) FROM ${BANCO}.prata_pedidos WHERE dt = '${DIA}'" \
  --result-configuration OutputLocation=s3://cadencia-lake/atenas-resultados/ \
  --query 'QueryExecutionId' --output text)
# Esperado: N2 == N1. Se N2 == 2 * N1, o passo de apagar a particao nao rodou.

# ── Prova 3: objetos no S3 foram SUBSTITUIDOS, nao acumulados ────────────────
aws s3api list-objects-v2 --bucket cadencia-lake \
  --prefix "prata/pedidos/dt=${DIA}/" --query 'length(Contents)'
# Esperado: contagem de arquivo estavel entre a primeira e a segunda execucao,
# nao crescente. Contagem crescente indica escrita por acrescimo, nao overwrite.

# ── Prova 4: o catalogo aprendeu a coluna nova SEM um segundo crawler ────────
aws glue get-table --database-name "${BANCO}" --name prata_pedidos \
  --query 'Table.StorageDescriptor.Columns[].Name'
# Esperado: 7 colunas, incluindo "canal_pedido". Se faltar, o
# enableUpdateCatalog do job nao aplicou — confira useGlueParquetWriter.

# ── Prova 5: falha real dispara alerta, nao fica em silencio ─────────────────
# Revogue temporariamente s3:DeleteObject do papel do job, rode de novo, e
# confirme a notificacao:
aws sns list-subscriptions-by-topic --topic-arn "$(terraform output -raw topico_alertas)" \
  --query 'Subscriptions[].[Endpoint,SubscriptionArn]'
aws glue get-job-runs --job-name "${PROJETO}-transformar-pedidos" \
  --query 'JobRuns[0].[JobRunState,ErrorMessage]'
# Esperado: JobRunState = FAILED, e a mensagem cita AccessDenied em DeleteObject.

#!/usr/bin/env bash
# limpar.sh — o que o destroy nao leva, e o que continua cobrando
set -euo pipefail

PROJETO="${PROJETO:?defina PROJETO}"

# 1. Derrube o que o Terraform administra (crawler, job, workflow, gatilhos,
#    regra do EventBridge, topico SNS, papeis).
terraform destroy -auto-approve

# 2. O QUE O DESTROY NAO TOCA: o BANCO do catalogo pode ficar com tabelas
#    orfas se voce criou alguma manualmente por fora do Terraform durante o
#    laboratorio (por exemplo, testando o antipadrao da arquitetura minima).
aws glue get-tables --database-name "${PROJETO}_lake" \
  --query 'TableList[].Name' --output table

# 3. Os DADOS em si: bronze e prata continuam no S3 do L62. Este laboratorio
#    nao os apaga porque nao foi ele quem os criou — apague-os apenas se voce
#    tambem estiver encerrando o L62.
aws s3 ls "s3://cadencia-lake/prata/pedidos/" --recursive --summarize \
  | tail -2

# 4. RESULTADOS DE CONSULTA DO ATHENA: cada execucao grava um arquivo no
#    bucket de resultados, e ele NAO tem ciclo de vida por padrao.
aws s3 rm "s3://cadencia-lake/atenas-resultados/" --recursive

# 5. Prova final: nada com o nome do projeto de pe fora do que era esperado.
aws resourcegroupstaggingapi get-resources \
  --tag-filters Key=Projeto,Values="${PROJETO}" \
  --query "ResourceTagMappingList[].ResourceARN" --output table