Databricks

# Instalação do Databricks CLI
pip install databricks-cli

# Configuração de autenticação
databricks configure --token

# Configuração do workspace
databricks workspace ls /

# Upload de notebook
databricks workspace import notebook.py /Users/user@company.com/notebook

# Databricks notebook cell
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp, when
from delta.tables import DeltaTable

# Spark session já configurado no Databricks
spark = spark

# Ler dados de origem
raw_data = spark.read.format("json") \
    .option("multiline", "true") \
    .load("/mnt/raw-data/events/")

# Transformações
cleaned_data = raw_data \
    .filter(col("event_type").isNotNull()) \
    .withColumn("processed_at", current_timestamp()) \
    .withColumn("is_valid", when(col("user_id").isNotNull(), True).otherwise(False))

# Escrever para Delta Lake
cleaned_data.write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .save("/mnt/delta/events_cleaned")

# Criar tabela Delta
spark.sql("""
    CREATE TABLE IF NOT EXISTS events_cleaned
    USING DELTA
    LOCATION '/mnt/delta/events_cleaned'
""")

print(f"Processados {cleaned_data.count()} registros")

-- Databricks SQL cell
-- Análise de eventos por dia
SELECT 
    DATE(processed_at) as date,
    event_type,
    COUNT(*) as event_count,
    COUNT(DISTINCT user_id) as unique_users
FROM events_cleaned
WHERE processed_at >= current_date() - INTERVAL 30 DAYS
GROUP BY DATE(processed_at), event_type
ORDER BY date DESC, event_count DESC;

-- Criar view materializada
CREATE OR REPLACE VIEW daily_events_summary AS
SELECT 
    DATE(processed_at) as date,
    event_type,
    COUNT(*) as total_events,
    COUNT(DISTINCT user_id) as unique_users,
    AVG(CASE WHEN is_valid THEN 1.0 ELSE 0.0 END) as validity_rate
FROM events_cleaned
GROUP BY DATE(processed_at), event_type;

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report

# Configurar MLflow
mlflow.set_experiment("/Users/user@company.com/fraud_detection")

# Carregar dados
df = spark.table("events_cleaned").toPandas()
X = df[['feature1', 'feature2', 'feature3']]
y = df['is_fraud']

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Iniciar run do MLflow
with mlflow.start_run():
    # Parâmetros do modelo
    n_estimators = 100
    max_depth = 10
    
    # Log de parâmetros
    mlflow.log_param("n_estimators", n_estimators)
    mlflow.log_param("max_depth", max_depth)
    
    # Treinar modelo
    model = RandomForestClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        random_state=42
    )
    model.fit(X_train, y_train)
    
    # Predições e métricas
    y_pred = model.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)
    
    # Log de métricas
    mlflow.log_metric("accuracy", accuracy)
    
    # Log do modelo
    mlflow.sklearn.log_model(model, "fraud_detection_model")
    
    print(f"Accuracy: {accuracy:.4f}")
    print(f"Run ID: {mlflow.active_run().info.run_id}")

import dlt
from pyspark.sql.functions import col, current_timestamp

# Bronze layer - dados brutos
@dlt.table(
    comment="Raw events data from source systems"
)
def bronze_events():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "json") \
        .load("/mnt/raw-data/events/")

# Silver layer - dados limpos
@dlt.table(
    comment="Cleaned and validated events"
)
@dlt.expect_or_drop("valid_event", "event_type IS NOT NULL")
@dlt.expect_or_drop("valid_user", "user_id IS NOT NULL")
def silver_events():
    return dlt.read_stream("bronze_events") \
        .select(
            col("event_id"),
            col("user_id"),
            col("event_type"),
            col("timestamp").cast("timestamp"),
            current_timestamp().alias("processed_at")
        )

# Gold layer - dados agregados
@dlt.table(
    comment="Daily event aggregations"
)
def gold_daily_events():
    return dlt.read("silver_events") \
        .groupBy(
            col("event_type"),
            date_format(col("timestamp"), "yyyy-MM-dd").alias("date")
        ) \
        .agg(
            count("*").alias("event_count"),
            countDistinct("user_id").alias("unique_users")
        )