⚡ Apache Spark — Domina isto para AI

---

O que é o Apache Spark?

Apache Spark é um motor de processamento de dados em grande escala de código aberto projetado para rapidez e facilidade de uso. Ele fornece APIs em várias linguagens de programação, incluindo Python, R, Java e Scala, e é amplamente utilizado para tarefas de processamento de dados e computação distribuída, especialmente em contextos de big data e inteligência artificial.

Por que usar Spark para AI?

• Velocidade: Executa operações de leitura, processamento e escrita em memória, tornando-o muito mais rápido que soluções baseadas em disco.
• Flexibilidade: Suporta diversas bibliotecas para machine learning (MLlib), processamento de gráficos (GraphX) e SQL.
• Escalabilidade: Funciona em clusters de grande escala, suportando operações em conjuntos de dados massivos sem perda de desempenho.

DataFrames vs RDDs

Estruturas de Dados em Spark

DataFrames

• Definição: Estruturas de dados semelhantes a tabelas. Eles possuem esquema e são otimizados para consultas.
• Uso em AI: DataFrames são usados para modelagem de dados, onde é necessário manipular conjuntos de dados estruturados de forma eficiente.

RDDs (Resilient Distributed Datasets)

• Definição: Coleções de objetos que podem ser processadas em paralelo. Menos eficientes que DataFrames, são usados quando se precisa de mais controle.
• Uso em AI: Utilizado em operações complexas que não podem ser expressas em DataFrames.

Exemplos de Aplicação em AI

1. Machine Learning

• MLlib: Biblioteca do Spark para machine learning. Oferece funcionalidades como classificação, regressão, agrupamento, e recomendações.

2. Processamento de Dados

• ETL: Extração, Transformação e Carga de dados em pipelines de dados para alimentar modelos de IA.

3. Análise em Tempo Real

• Stream Processing: Análise de fluxos de dados em tempo real, útil para aplicações de IA que requerem decisões instantâneas.

Lazy Evaluation

• Transformations (map, filter, join) não executam imediatamente — constroem DAG.
• Actions (count, collect, write) disparam a execução.
• Benefício: Catalyst otimiza o plano antes de executar (predicate pushdown, pruning).

Partitions (🔥 MUITO perguntado)

• O quê: Divisão dos dados em blocos processados em paralelo.
• Repartition: Redistribui (shuffle). repartition(n) — n partições.
• Coalesce: Reduz partições sem shuffle (útil após filtros).
• Partition by: Em writes, partitionBy("date") cria pastas por coluna.
• Regra: Partições ≈ 2–4× cores do cluster. Muito pequenas = overhead; muito grandes = desequilíbrio.

Shuffle

• O quê: Redistribuição de dados entre nós (ex: em joins, groupBy).
• Porque é caro: I/O de rede, serialização, possível data skew.
• Mitigar: Broadcast join para tabelas pequenas, partition antes de join, evitar repartition desnecessário.

Joins

Broadcast Join

• Tabela pequena replicada para todos os nós. Sem shuffle na tabela pequena.
• broadcast(df_small) ou spark.conf.set("spark.sql.autoBroadcastJoinThreshold", ...)
• Quando: Tabela < ~100MB (depende da config).

Sort-Merge Join

• Ambas as tabelas partidas pela join key; merge local. Shuffle em ambas.
• Quando: Tabelas grandes.

Skew

• Algumas partições têm muito mais dados. Mitigar: salting (adicionar coluna aleatória para repartir), split da partição grande.

Performance Tuning

1. Partitioning: Partições adequadas, partitionBy em writes.
2. Caching: cache() ou persist() para reutilizar (evitar recompute).
3. Broadcast: Joins com tabelas pequenas.
4. Predicate pushdown: Filtros cedo (Catalyst faz automaticamente com Parquet).
5. Coalesce antes de write: Evitar muitos ficheiros pequenos.

Exemplos de Código para AI

Exemplo 1: Treinamento de um Modelo de Classificação Simples (PySpark)

from pyspark.sql import SparkSession
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler

# Cria uma sessão Spark
spark = SparkSession.builder.appName("AI Model").getOrCreate()

# Carrega os dados
data = spark.read.csv("data.csv", header=True, inferSchema=True)

# Preparação dos dados
assembler = VectorAssembler(inputCols=["feature1", "feature2"], outputCol="features")
data_transformed = assembler.transform(data)

# Treinamento do modelo
lr = LogisticRegression(featuresCol="features", labelCol="label")
model = lr.fit(data_transformed)

# Salva o modelo
model.save("logistic_regression_model")

Exemplo 2: Processamento de Dados em Fluxo (Stream Processing)

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

# Cria uma sessão Spark
spark = SparkSession.builder.appName("Stream Processing").getOrCreate()

# Configura a origem do stream
streaming_data = spark.readStream.schema(schema).json("/path/to/json/files")

# Processa os dados
result = streaming_data.groupBy("key").agg(count("value").alias("count"))

# Escreve o resultado em um sink
query = result.writeStream.outputMode("complete").format("console").start()
query.awaitTermination()

Frase que impressiona

"Apache Spark é a espinha dorsal para a computação em larga escala em projetos de AI, unindo velocidade e flexibilidade para resolver problemas complexos de dados."