🏗️ DALX Data Lake

🔐 Login

• Acesse qualquer serviço via link acima
• GitHub OAuth org: binariocloud
• Airflow pede login extra peça ao admin
• MinIO pede access key peça ao admin

🗃️ Trino (SQL)

• Externo (browser) 200.218.226.44:8090
• Interno (DAGs) trino.query.svc.cluster.local:8080
• User seu.nome
• Password (vazio)
• Catalogs bronze, silver, gold

💾 MinIO (S3)

• Endpoint http://200.218.226.44:9000
• Region us-east-1
• Buckets bronze, silver, gold

mc alias set dalx http://200.218.226.44:9000 KEY SECRET
mc ls dalx/bronze/

🐍 Python (Trino) — dentro das DAGs

from trino.dbapi import connect

# Usar endereço INTERNO do cluster nas DAGs
conn = connect(
    host="trino.query.svc.cluster.local",
    port=8080,
    user="airflow",
    catalog="silver",
    schema="default",
)
cur = conn.cursor()
cur.execute("SHOW TABLES")
print(cur.fetchall())

🐍 Python (MinIO/boto3)

import boto3

s3 = boto3.client('s3',
    endpoint_url='http://200.218.226.44:9000',
    aws_access_key_id='KEY',
    aws_secret_access_key='SECRET',
    region_name='us-east-1'
)
objs = s3.list_objects_v2(Bucket='bronze')
for o in objs.get('Contents', []):
    print(o['Key'])

🔎 Elasticsearch

# Listar índices
curl http://200.218.226.44:9200/_cat/indices?v

# Buscar
curl http://200.218.226.44:9200/INDEX/_search \
  -H 'Content-Type: application/json' \
  -d '{"query":{"match_all":{}}}'

⚙️ Connections disponíveis

• trino_default Trino (silver catalog)
• minio_s3 MinIO S3 (todos os buckets)
• elasticsearch_default Elasticsearch
• airbyte_default Airbyte API

📝 DAG — Trino (SQL no Silver/Gold)

from airflow import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from datetime import datetime

with DAG("trino_silver_to_gold",
         start_date=datetime(2026, 1, 1),
         schedule="@daily",
         catchup=False) as dag:

    create_gold = SQLExecuteQueryOperator(
        task_id="create_gold_table",
        conn_id="trino_default",
        sql="""
            CREATE TABLE gold.default.vendas_diarias AS
            SELECT date(created_at) as dia,
                   count(*) as total,
                   sum(valor) as receita
            FROM silver.default.vendas
            GROUP BY 1
        """,
    )

📝 DAG — MinIO/S3 (Upload/Download)

from airflow import DAG
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.operators.python import PythonOperator
from datetime import datetime

def upload_to_bronze():
    hook = S3Hook(aws_conn_id="minio_s3")
    hook.load_string(
        string_data='{"id":1,"nome":"teste"}',
        key="dados/exemplo.json",
        bucket_name="bronze",
        replace=True
    )

def list_bronze():
    hook = S3Hook(aws_conn_id="minio_s3")
    keys = hook.list_keys(bucket_name="bronze")
    print(f"Arquivos no bronze: {keys}")

with DAG("minio_exemplo",
         start_date=datetime(2026, 1, 1),
         schedule=None,
         catchup=False) as dag:

    upload = PythonOperator(
        task_id="upload",
        python_callable=upload_to_bronze)
    listar = PythonOperator(
        task_id="listar",
        python_callable=list_bronze)
    upload >> listar

📝 DAG — Elasticsearch (Indexar)

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import json, urllib.request

ES_URL = "http://elasticsearch.busca.svc.cluster.local:9200"

def indexar_documento():
    doc = {"titulo": "Teste", "conteudo": "Hello DALX"}
    req = urllib.request.Request(
        f"{ES_URL}/meu-indice/_doc/1",
        data=json.dumps(doc).encode(),
        headers={"Content-Type": "application/json"},
        method="PUT"
    )
    resp = urllib.request.urlopen(req)
    print(resp.read().decode())

with DAG("elasticsearch_indexar",
         start_date=datetime(2026, 1, 1),
         schedule=None,
         catchup=False) as dag:

    PythonOperator(
        task_id="indexar",
        python_callable=indexar_documento)

📝 DAG — Airbyte (Trigger Sync)

from airflow import DAG
from airflow.providers.http.operators.http import SimpleHttpOperator
from datetime import datetime

with DAG("airbyte_trigger_sync",
         start_date=datetime(2026, 1, 1),
         schedule="@daily",
         catchup=False) as dag:

    # Substitua CONNECTION_ID pelo ID da conexão no Airbyte
    trigger = SimpleHttpOperator(
        task_id="trigger_sync",
        http_conn_id="airbyte_default",
        endpoint="/api/v1/connections/sync",
        method="POST",
        headers={"Content-Type": "application/json"},
        data='{"connectionId": "SEU_CONNECTION_ID"}',
        response_check=lambda r: r.status_code == 200,
    )

⚡ Redis (Serving Layer)

• Host 200.218.226.44:6379
• Interno redis.cache.svc.cluster.local:6379
• Conn Airflow redis_default
• Memória 64MB (LRU eviction)
• Sem senha (protegido por NetworkPolicy)

# CLI
redis-cli -h 200.218.226.44 -p 6379
> PING
> SET chave "valor"
> GET chave

📝 DAG — Gold → Redis (Serving)

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import json

def gold_to_redis():
    """Materializa dados da Gold no Redis."""
    from trino.dbapi import connect
    import redis

    # Lê do Gold (Trino)
    trino_conn = connect(
        host="trino.query.svc.cluster.local",
        port=8080, user="airflow",
        catalog="gold", schema="default"
    )
    cur = trino_conn.cursor()
    cur.execute("""
        SELECT produto, sum(receita) as total
        FROM vendas_diarias
        GROUP BY produto ORDER BY total DESC
        LIMIT 100
    """)
    rows = cur.fetchall()

    # Escreve no Redis
    r = redis.Redis(
        host="redis.cache.svc.cluster.local",
        port=6379, decode_responses=True
    )
    # Hash com ranking de produtos
    pipeline = r.pipeline()
    for produto, total in rows:
        pipeline.hset("gold:top_produtos",
                      produto, str(total))
    pipeline.set("gold:top_produtos:updated_at",
                 datetime.now().isoformat())
    pipeline.execute()
    print(f"Cached {len(rows)} produtos no Redis")

with DAG("gold_to_redis",
         start_date=datetime(2026, 1, 1),
         schedule="*/15 * * * *",
         catchup=False) as dag:

    PythonOperator(
        task_id="materializar_redis",
        python_callable=gold_to_redis)