Eu sou Sarah Nascimento

from pyspark.sql import functions as F

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.databricks.io.cache.enabled", "true")

df_optimized = spark.table("silver.telemetry_raw") \
    .repartition(F.col("device_id")) \
    .sortWithinPartitions("event_time")

df_optimized.write.format("delta") \
    .option("dataChange", "false") \
    .mode("overwrite") \
    .saveAsTable("silver.telemetry_optimized")
...

-- Otimização de leitura e data skipping com Z-Order em tabelas Delta
OPTIMIZE gold.user_transactions
ZORDER BY (user_id, transaction_date);

-- Análise das estatísticas pós-otimização
DESCRIBE HISTORY gold.user_transactions;
ANALYZE TABLE gold.user_transactions COMPUTE STATISTICS;
...

from delta.tables import DeltaTable

# Instancia a tabela Delta destino
deltaTable = DeltaTable.forPath(spark, "s3://data-lake/silver/clients")

# Executa o Upsert (Merge) massivo de atualizações e inserções
deltaTable.alias("tgt").merge(
    df_updates.alias("src"),
    "tgt.client_id = src.client_id"
).whenMatchedUpdateAll(
).whenNotMatchedInsertAll().execute()
...

-- Criação de tabela particionada com auto-otimização (Databricks)
CREATE TABLE IF NOT EXISTS silver.events_log (
    event_id STRING,
    user_id STRING,
    event_type STRING,
    event_timestamp TIMESTAMP,
    processing_date DATE
)
USING DELTA
PARTITIONED BY (processing_date)
TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true',
    'delta.deletedFileRetentionDuration' = 'interval 15 days'
);
...

import requests

def fetch_paginated_data(api_url, headers):
    results = []
    page = 1
    
    while True:
        response = requests.get(f"{api_url}?page={page}", headers=headers)
        response.raise_for_status()
        data = response.json()
        
        if not data.get('items'): 
            break
            
        results.extend(data['items'])
        page += 1
        
    return results
...

# Acesso seguro via Azure Key Vault / Databricks Secrets
jdbc_user = dbutils.secrets.get(scope="kv-prod", key="db-user")
jdbc_pass = dbutils.secrets.get(scope="kv-prod", key="db-pass")

df_jdbc = spark.read.format("jdbc") \
    .option("url", "jdbc:postgresql://server:5432/prod_db") \
    .option("dbtable", "sales.orders") \
    .option("user", jdbc_user) \
    .option("password", jdbc_pass) \
    .option("fetchsize", 10000) \
    .load()
...

import httpx
import asyncio
from bs4 import BeautifulSoup

async def fetch_and_parse(url: str, proxy: str = None):
    limits = httpx.Limits(max_keepalive_connections=10, max_connections=20)
    
    async with httpx.AsyncClient(limits=limits, proxies=proxy) as client:
        response = await client.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10.0)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, 'lxml')
        payload = [
            {"id": node.get('data-id'), "val": node.find('span', class_='metric').text.strip()}
            for node in soup.select('div.data-row') if node.get('data-id')
        ]
        return payload
...

WITH RECURSIVE Hierarchy AS (
    SELECT id, parent_id, amount, 1 AS level
    FROM transactions.raw_ledger
    WHERE parent_id IS NULL
    
    UNION ALL
    
    SELECT child.id, child.parent_id, child.amount, h.level + 1
    FROM transactions.raw_ledger child
    INNER JOIN Hierarchy h ON child.parent_id = h.id
),
RankedLedger AS (
    SELECT *, 
           SUM(amount) OVER (PARTITION BY parent_id ORDER BY level DESC) as rolling_sum
    FROM Hierarchy
)
SELECT * FROM RankedLedger WHERE rolling_sum > 100000;
...