Apache Spark 4 em 2026: Novos Recursos, Structured Streaming e Perguntas de Entrevista

Apache Spark 4 representa a reformulação mais significativa do motor Spark desde a versão 3.0. Lançado em meados de 2025 com iteração rápida através das versões 4.1 e a próxima 4.2, a linha 4.x redefine como engenheiros de dados constroem pipelines, executam cargas de trabalho de streaming e manipulam SQL em escala. Este guia detalha as mudanças críticas, demonstra-as com código PySpark e Scala, e aborda as perguntas de entrevista mais frequentes que equipes de recrutamento fazem.

Modo ANSI SQL Agora Ativado por Padrão no Spark 4

O Spark 4.0 muda spark.sql.ansi.enabled para true. Todo overflow aritmético, cast inválido e divisão por zero que anteriormente retornava NULL ou truncava silenciosamente agora lança uma exceção. O objetivo: conformidade estrita com o padrão SQL, compatível com o comportamento do PostgreSQL e Oracle.

O impacto prático atinge principalmente pipelines ETL. Queries que dependiam de coerção implícita ou propagação silenciosa de NULL falharão em tempo de execução até serem explicitamente tratadas.

# ansi_mode_migration.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, try_cast

spark = SparkSession.builder.appName("ANSIMigration").getOrCreate()

# Isso agora lança ArithmeticException no modo ANSI do Spark 4
# SELECT CAST('abc' AS INT)  -- erro em tempo de execução

# Migração segura: use TRY_CAST para preservar comportamento NULL
df = spark.sql("""
    SELECT
        TRY_CAST(raw_value AS INT) AS parsed_value,
        TRY_CAST(amount AS DECIMAL(10,2)) AS safe_amount
    FROM raw_events
    WHERE TRY_CAST(raw_value AS INT) IS NOT NULL
""")

# Alternativamente, envolva aritmética em try_divide / try_add
result = df.withColumn(
    "ratio",
    col("safe_amount") / col("parsed_value")  # lança exceção em zero
)

Checklist de migração: audite cada chamada CAST, substitua por TRY_CAST onde comportamento NULL-safe é esperado, e adicione verificações explícitas de overflow em colunas numéricas. O guia de migração SQL do Spark documenta cada mudança comportamental.

Tipo de Dados VARIANT para Dados Semi-Estruturados

O Spark 4.0 adiciona um tipo de coluna nativo VARIANT projetado para dados semi-estruturados tipo JSON. Diferente de armazenar strings JSON brutas e analisá-las em tempo de query, VARIANT usa um formato binário otimizado com column shredding que entrega até 8x mais velocidade de leitura no acesso a campos aninhados.

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder.appName("VariantDemo").getOrCreate()
import spark.implicits._

// Cria uma tabela com coluna VARIANT
spark.sql("""
    CREATE TABLE events (
        event_id BIGINT,
        timestamp TIMESTAMP,
        payload VARIANT
    ) USING PARQUET
""")

// Insere dados JSON — automaticamente convertido para VARIANT
spark.sql("""
    INSERT INTO events VALUES
    (1, current_timestamp(),
     PARSE_JSON('{"user_id": 42, "action": "click", "metadata": {"page": "/home", "duration_ms": 320}}')),
    (2, current_timestamp(),
     PARSE_JSON('{"user_id": 87, "action": "purchase", "metadata": {"item_id": "SKU-9001", "amount": 49.99}}'))
""")

// Consulta campos aninhados com notação de ponto — usa shredding para acesso rápido
val clicks = spark.sql("""
    SELECT
        event_id,
        payload:user_id::INT AS user_id,
        payload:action::STRING AS action,
        payload:metadata.page::STRING AS page
    FROM events
    WHERE payload:action::STRING = 'click'
""")
clicks.show()

VARIANT elimina a necessidade de workarounds de schema-on-read. Em arquiteturas de data lake onde esquemas de eventos evoluem frequentemente, uma única coluna VARIANT absorve mudanças de esquema sem migrações DDL. O Spark 4.1 estendeu VARIANT com shredding no nível Parquet, reduzindo ainda mais o IO para acesso seletivo de campos.

Streaming em Modo Real-Time no Spark 4.1

Structured Streaming no Spark 4.1 introduz o modo Real-Time (RTM), o primeiro suporte oficial para latência sub-segundo no Spark. Para cargas de trabalho stateless, a latência cai para milissegundos de um dígito, competindo diretamente com Apache Flink em seu próprio território.

Três mudanças arquiteturais tornam isso possível: fluxo contínuo de dados (sem limites de micro-batch), agendamento de pipeline (sobreposição de estágios read/compute/write), e streaming shuffle (estado armazenado em memória em vez de checkpoints entre batches).

# real_time_streaming.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder \
    .appName("RTMDemo") \
    .config("spark.sql.streaming.mode", "realtime") \
    .getOrCreate()

schema = StructType([
    StructField("sensor_id", StringType()),
    StructField("temperature", DoubleType()),
    StructField("event_time", TimestampType())
])

# Lê do Kafka com modo Real-Time ativado
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "sensor-readings") \
    .option("startingOffsets", "latest") \
    .load()

parsed = raw_stream \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .select("data.*")

# Escreve para sink Kafka — end-to-end sub-segundo
query = parsed \
    .filter(col("temperature") > 80.0) \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("topic", "alerts") \
    .option("checkpointLocation", "/tmp/checkpoints/alerts") \
    .start()

query.awaitTermination()

Ativar RTM requer apenas uma mudança de configuração: spark.sql.streaming.mode = realtime. Nenhuma reescrita de código necessária.

Spark Connect e o Cliente PySpark Leve

O Spark 4.0 fornece um pacote standalone pyspark-client pesando apenas 1.5 MB, comparado à distribuição PySpark completa de mais de 300 MB.

# spark_connect_client.py
from pyspark.sql import SparkSession

# Conecta ao cluster Spark remoto — nenhuma JVM local necessária
spark = SparkSession.builder \
    .remote("sc://spark-cluster.internal:15002") \
    .getOrCreate()

# API DataFrame completa disponível através do cliente leve
df = spark.read.parquet("s3a://data-lake/events/2026/")

aggregated = df.groupBy("country", "event_type") \
    .count() \
    .orderBy("count", ascending=False)

aggregated.show(20)

API transformWithState para Lógica de Streaming Complexa

O Spark 4.0 substitui o limitado mapGroupsWithState por transformWithState.

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.streaming.StatefulProcessor
import java.time.Duration

case class UserEvent(userId: String, action: String, timestamp: Long)
case class SessionSummary(userId: String, actionCount: Int, durationMs: Long)

class SessionProcessor extends StatefulProcessor[String, UserEvent, SessionSummary] {

  @transient private var sessionStart: ValueState[Long] = _
  @transient private var actionCount: ValueState[Int] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    sessionStart = getHandle.getValueState[Long]("start", TTLConfig(Duration.ofMinutes(30)))
    actionCount = getHandle.getValueState[Int]("count", TTLConfig(Duration.ofMinutes(30)))
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[UserEvent],
      timerValues: TimerValues
  ): Iterator[SessionSummary] = {
    rows.flatMap { event =>
      if (!sessionStart.exists()) {
        sessionStart.update(event.timestamp)
        actionCount.update(1)
        Iterator.empty
      } else {
        val count = actionCount.get() + 1
        actionCount.update(count)
        if (count % 10 == 0) {
          val duration = event.timestamp - sessionStart.get()
          Iterator(SessionSummary(key, count, duration))
        } else Iterator.empty
      }
    }
  }
}

Spark Declarative Pipelines (SDP) no Spark 4.1

# declarative_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum, current_date

spark = SparkSession.builder.appName("SDP-Demo").getOrCreate()

@spark.declarative.table("clean_orders")
def clean_orders():
    return spark.read.table("raw_orders") \
        .filter(col("status") != "cancelled") \
        .filter(col("amount") > 0)

@spark.declarative.table("daily_revenue")
def daily_revenue():
    return spark.read.table("clean_orders") \
        .groupBy("order_date", "region") \
        .agg(_sum("amount").alias("total_revenue"))

@spark.declarative.table("revenue_summary")
def revenue_summary():
    return spark.read.table("daily_revenue") \
        .groupBy("region") \
        .agg(_sum("total_revenue").alias("grand_total")) \
        .orderBy("grand_total", ascending=False)

spark.declarative.run()

Perguntas de Entrevista sobre Spark 4 em Engenharia de Dados

Equipes de contratação em engenharia de dados priorizam conhecimento prático sobre mudanças arquiteturais que impactam pipelines de produção. As cinco perguntas abaixo aparecem em entrevistas técnicas para posições focadas em Spark 4.

1. O que muda quando o modo ANSI é ativado por padrão no Spark 4?

O modo ANSI transforma o comportamento de tratamento de erros do Spark de permissivo para rigoroso. Operações que anteriormente retornavam NULL ou truncavam valores silenciosamente agora lançam exceções em tempo de execução. Isso inclui overflow aritmético (excedendo limites INT/LONG), casts inválidos (string para número), e divisão por zero. A mudança visa compatibilidade com padrões SQL de bancos como PostgreSQL e Oracle. Pipelines de produção requerem migração: substituir CAST por TRY_CAST, adicionar validação explícita em operações aritméticas, e testar queries existentes antes de atualizar para Spark 4.

2. Explique o tipo de dados VARIANT e quando usá-lo.

VARIANT é um tipo de coluna otimizado para dados semi-estruturados como JSON ou Avro. Em vez de armazenar strings JSON brutas e analisá-las repetidamente em cada query, VARIANT usa um formato binário com column shredding que extrai campos frequentemente acessados para colunas separadas no Parquet. Isso entrega até 8x de melhoria de desempenho em queries com filtros em campos aninhados. Use VARIANT em data lakes onde esquemas de eventos evoluem frequentemente, como logs de aplicação, eventos de usuário, ou dados de IoT. O tipo absorve mudanças de esquema sem migrações DDL, eliminando o problema de schema drift em arquiteturas de lakehouse.

3. Como o modo Real-Time difere do streaming micro-batch?

Modo Real-Time elimina limites de micro-batch e processa dados em fluxo contínuo. Três mudanças arquiteturais reduzem latência para milissegundos de um dígito: (1) continuous data flow sem esperar acúmulo de registros, (2) pipeline scheduling que sobrepõe estágios de leitura, computação e escrita, e (3) streaming shuffle que mantém estado em memória em vez de checkpoints em disco. Micro-batch tradicional espera um intervalo fixo (padrão 500ms) antes de processar cada batch, introduzindo latência mínima mesmo com baixo volume de dados. RTM compete diretamente com Apache Flink em casos de uso de latência ultrabaixa como detecção de fraude em tempo real e sistemas de alertas.

4. O que é transformWithState e por que substituir mapGroupsWithState?

transformWithState é uma API de estado arbitrário que substitui mapGroupsWithState e flatMapGroupsWithState. A API anterior tinha limitações: não suportava múltiplas variáveis de estado por chave, não permitia TTL por estado individual, e exigia gerenciamento manual de timeouts. transformWithState introduz StatefulProcessor com suporte nativo para múltiplos ValueState e ListState, TTL configurável por variável de estado, e gerenciamento automático de timers. Casos de uso incluem rastreamento de sessões complexas, agregações com janelas customizadas, e máquinas de estado finitas em streaming. A migração requer refatoração de código mas oferece flexibilidade significativamente maior para lógica stateful.

5. Qual problema o Spark Connect resolve?

Spark Connect desacopla cliente e servidor através de um protocolo gRPC, permitindo que aplicações usem a API DataFrame completa sem instalar uma JVM local ou distribuição Spark de 300+ MB. O cliente pyspark-client pesa apenas 1.5 MB e serializa operações DataFrame para o servidor remoto. Isso resolve três problemas: (1) ambientes de notebook leves sem overhead de JVM, (2) isolamento de versão entre cliente e cluster Spark, e (3) execução de código PySpark de qualquer linguagem que implemente o protocolo Connect. O trade-off é latência adicional de rede em operações de coleta de dados, tornando-o inadequado para iteração local intensiva sobre pequenos datasets.

Construindo um Pipeline ETL Completo com Spark 4

Um pipeline ETL de produção em Spark 4 combina streaming em modo Real-Time, tipo VARIANT para ingestão de eventos semi-estruturados, e modo ANSI para validação rigorosa de dados. O exemplo abaixo implementa arquitetura medallion (bronze-silver-gold) lendo eventos de Kafka, transformando com VARIANT, e escrevendo em Delta Lake.

# spark4_etl_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, current_timestamp, parse_json

spark = SparkSession.builder \
    .appName("Spark4ETL") \
    .config("spark.sql.ansi.enabled", "true") \
    .getOrCreate()

bronze = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "user-events") \
    .load() \
    .select(
        col("key").cast("string").alias("event_key"),
        parse_json(col("value").cast("string")).alias("payload"),
        col("timestamp").alias("kafka_ts")
    )

silver = bronze.select(
    col("event_key"),
    col("payload:user_id").cast("int").alias("user_id"),
    col("payload:action").cast("string").alias("action"),
    col("payload:metadata.session_id").cast("string").alias("session_id"),
    col("kafka_ts"),
    current_timestamp().alias("processed_at")
).filter(
    col("user_id").isNotNull()
)

query = silver.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3a://checkpoints/user-events/") \
    .option("mergeSchema", "true") \
    .toTable("gold.user_events")

query.awaitTermination()

Este pipeline demonstra padrões essenciais de Spark 4: camada bronze com VARIANT para flexibilidade de esquema, camada silver com validação ANSI, e sink Delta com merge automático de schema. A arquitetura suporta evolução de eventos sem downtime e garante qualidade de dados através de validação rigorosa.

Conclusão

Apache Spark 4 estabelece nova baseline para pipelines de dados modernos em 2026. As mudanças críticas que impactam engenheiros de dados incluem:

• Modo ANSI por padrão: Validação rigorosa substitui comportamento permissivo, exigindo migração de pipelines existentes com TRY_CAST e tratamento explícito de erros
• Tipo VARIANT: Desempenho até 8x melhor em dados semi-estruturados elimina workarounds de schema-on-read em arquiteturas de lakehouse
• Real-Time Mode: Latência sub-segundo em streaming stateless compete diretamente com Apache Flink para casos de uso de baixa latência
• transformWithState: API de estado arbitrário com suporte a múltiplos estados, TTL configurável e gerenciamento automático de timers
• Spark Connect: Cliente leve de 1.5 MB habilita execução remota sem JVM local, ideal para notebooks e ambientes serverless
• Declarative Pipelines: Definição de DAGs de transformação como funções Python reduz boilerplate e melhora testabilidade

Pipelines de produção em Spark 4 combinam esses recursos para arquiteturas medallion resilientes com validação rigorosa, ingestão flexível de eventos semi-estruturados, e latência otimizada para casos de uso real-time. A transição de Spark 3.x exige planejamento cuidadoso, especialmente para migração de modo ANSI, mas entrega ganhos significativos de desempenho e confiabilidade.