DEV Community

Cover image for Como aprendi Apache Spark — Parte 12: CSV, JSON, Parquet e ORC — performance começa no armazenamento
renanpyd
renanpyd

Posted on

Como aprendi Apache Spark — Parte 12: CSV, JSON, Parquet e ORC — performance começa no armazenamento

#bigdata #spark #apache #python

Nota histórica: o título desta série faz referência à minha trajetória em Engenharia de Dados iniciada em 2012. As tecnologias e práticas apresentadas evoluíram em momentos diferentes. Este artigo revisita esses fundamentos sob a perspectiva das arquiteturas modernas de dados.

Nos capítulos anteriores, exploramos como o Apache Spark distribui o processamento entre Drivers, Executors e Tasks.

Entendemos partições, shuffle, memória, lineage, tolerância a falhas e dimensionamento de recursos.

Na Parte 11, discutimos uma questão importante:

Como utilizar CPU, memória e paralelismo de maneira eficiente?

Agora vamos examinar um fator que pode mudar significativamente o comportamento de uma aplicação sem alterar sua lógica de negócio:

O formato dos dados.

Imagine dois pipelines executando a mesma consulta sobre o mesmo conjunto de informações.

O primeiro lê arquivos CSV.

O segundo lê arquivos Parquet.

Ambos executam:

resultado = (
    vendas
    .filter("estado = 'SP'")
    .groupBy("categoria")
    .sum("valor")
)
Enter fullscreen mode Exit fullscreen mode

A lógica é idêntica.

Mas a quantidade de dados lidos, o trabalho de interpretação e o plano de execução podem ser diferentes.

Por quê?

Porque o Spark não processa dados independentemente de como eles estão armazenados.

O armazenamento influencia diretamente a leitura.

E a leitura influencia o custo do processamento.

1. O formato dos dados faz parte da arquitetura

Quando pensamos em performance, normalmente lembramos de:

  • Quantidade de Executors.
  • Memória disponível.
  • Número de partições.
  • Estratégias de join.
  • Shuffle.
  • CPU.

Mas existe uma camada anterior:

Como os dados foram gravados?

Considere:

                   DATA SOURCE
                       |
                       v
                 FILE FORMAT
                       |
                       v
                  DATA READER
                       |
                       v
                SPARK DATAFRAME
                       |
                       v
                 QUERY PLAN
                       |
                       v
                 EXECUTION
                       |
                       v
                    RESULT
Enter fullscreen mode Exit fullscreen mode

O formato pode determinar:

  • Como os registros são representados.
  • Como os tipos são armazenados.
  • Quais metadados estão disponíveis.
  • Quais colunas precisam ser lidas.
  • Quais filtros podem ser aplicados antecipadamente.
  • Quanto espaço os dados ocupam.
  • Quanto trabalho de parsing será necessário.

Portanto:

Otimização de Spark não começa necessariamente no Executor. Muitas vezes começa na organização dos dados.

2. Row-based vs. Columnar

Uma das distinções mais importantes é entre armazenamento orientado a linhas e armazenamento orientado a colunas.

Imagine uma tabela:

id cliente estado valor
1 Ana SP 100
2 Bruno RJ 200
3 Carla SP 300

Em uma representação orientada a linhas, os valores de cada registro são armazenados próximos uns dos outros.

Conceitualmente:

ROW-ORIENTED

[1, Ana, SP, 100]
[2, Bruno, RJ, 200]
[3, Carla, SP, 300]
Enter fullscreen mode Exit fullscreen mode

Em uma representação orientada a colunas, valores da mesma coluna são agrupados.

COLUMNAR

id:
[1, 2, 3]

cliente:
[Ana, Bruno, Carla]

estado:
[SP, RJ, SP]

valor:
[100, 200, 300]
Enter fullscreen mode Exit fullscreen mode

Essa diferença influencia o tipo de operação que cada formato favorece.

Se queremos recuperar registros completos, uma organização por linhas pode ser conveniente.

Se queremos calcular:

SELECT SUM(valor)
FROM vendas;
Enter fullscreen mode Exit fullscreen mode

uma organização colunar pode permitir ler principalmente a coluna valor.

É por isso que formatos colunares são tão importantes em workloads analíticos.

3. CSV: simples, interoperável e caro de interpretar

CSV significa Comma-Separated Values.

É um formato textual amplamente utilizado para intercâmbio de dados.

Exemplo:

id,cliente,estado,valor
1,Ana,SP,100
2,Bruno,RJ,200
3,Carla,SP,300
Enter fullscreen mode Exit fullscreen mode

Sua principal vantagem é a simplicidade.

Arquivos CSV podem ser produzidos e consumidos por uma enorme variedade de ferramentas.

Mas existem limitações.

CSV não preserva nativamente um schema tipado como o de um formato colunar.

O valor:

100
Enter fullscreen mode Exit fullscreen mode

pode representar um inteiro, decimal ou texto, dependendo da interpretação.

Além disso, delimitadores, aspas, escapes, valores nulos e codificação precisam ser tratados corretamente.

Em grandes volumes, esse parsing pode representar um custo relevante.

4. Lendo CSV com PySpark

Um exemplo básico:

from pyspark.sql import SparkSession

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

df_csv = (
    spark.read
    .option("header", True)
    .csv("/data/vendas.csv")
)

df_csv.printSchema()
Enter fullscreen mode Exit fullscreen mode

Sem schema explícito ou inferência habilitada, as colunas normalmente são interpretadas como strings.

Podemos utilizar:

df_csv = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("/data/vendas.csv")
)
Enter fullscreen mode Exit fullscreen mode

Mas a inferência exige examinar os dados para determinar tipos.

Em pipelines de produção, quando conhecemos o contrato dos dados, geralmente é melhor fornecer o schema.

5. Schema explícito: menos ambiguidade

Exemplo:

from pyspark.sql.types import (
    StructType,
    StructField,
    IntegerType,
    StringType,
    DecimalType
)

schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("cliente", StringType(), True),
    StructField("estado", StringType(), True),
    StructField("valor", DecimalType(12, 2), True)
])

df_csv = (
    spark.read
    .option("header", True)
    .schema(schema)
    .csv("/data/vendas.csv")
)
Enter fullscreen mode Exit fullscreen mode

Agora declaramos a estrutura esperada.

Isso traz benefícios:

  • Contrato de tipos mais claro.
  • Menor dependência de inferência.
  • Comportamento mais previsível.
  • Melhor rastreabilidade.
  • Validação mais objetiva.

Entretanto, declarar um campo como não anulável não substitui uma rotina completa de qualidade de dados. A validação de registros malformados e de restrições de negócio deve ser tratada explicitamente.

Essa preocupação é particularmente importante quando recebemos arquivos de sistemas externos.

6. JSON: flexibilidade com custo de parsing

JSON é outro formato muito utilizado.

Exemplo:

{
  "id": 1,
  "cliente": "Ana",
  "estado": "SP",
  "valor": 100
}
Enter fullscreen mode Exit fullscreen mode

Ele permite representar estruturas mais complexas, como arrays e objetos aninhados.

Exemplo:

{
  "pedido_id": 101,
  "cliente": {
    "id": 20,
    "nome": "Ana"
  },
  "itens": [
    {"produto": "A", "quantidade": 2},
    {"produto": "B", "quantidade": 1}
  ]
}
Enter fullscreen mode Exit fullscreen mode

Essa flexibilidade é útil em APIs, eventos e integrações.

Mas existe um custo.

O Spark precisa interpretar a estrutura textual e convertê-la para os tipos internos utilizados no processamento.

Além disso, schemas variáveis podem dificultar governança e evolução de pipelines.

7. Lendo JSON no Spark

Para arquivos JSON Lines, com um objeto JSON por linha:

df_json = spark.read.json(
    "/data/eventos.json"
)
Enter fullscreen mode Exit fullscreen mode

Podemos fornecer um schema explícito:

from pyspark.sql.types import (
    StructType,
    StructField,
    LongType,
    StringType
)

schema_eventos = StructType([
    StructField("event_id", LongType(), True),
    StructField("event_type", StringType(), True)
])

df_json = (
    spark.read
    .schema(schema_eventos)
    .json("/data/eventos.json")
)
Enter fullscreen mode Exit fullscreen mode

É importante distinguir JSON Lines de um documento JSON convencional contendo um array ou objeto multilinha.

Para determinados arquivos JSON multilinha, podemos precisar de:

df_json = (
    spark.read
    .option("multiLine", True)
    .json("/data/eventos_multilinha.json")
)
Enter fullscreen mode Exit fullscreen mode

O modo multilinha pode afetar a estratégia de leitura e o paralelismo. Por isso, não devemos habilitá-lo indiscriminadamente.

8. Parquet: armazenamento colunar para analytics

Agora chegamos a um dos formatos mais importantes no ecossistema moderno de Engenharia de Dados.

Apache Parquet é um formato colunar desenvolvido para armazenamento e processamento analítico.

Ele suporta:

  • Schema tipado.
  • Compressão.
  • Codificações eficientes.
  • Metadados.
  • Organização em grupos de linhas.
  • Leitura seletiva de colunas.
  • Estatísticas que podem auxiliar a eliminação de dados desnecessários.

Considere uma consulta:

SELECT estado, SUM(valor)
FROM vendas
GROUP BY estado;
Enter fullscreen mode Exit fullscreen mode

Se o dataset possui 80 colunas, mas a consulta utiliza apenas duas, o Parquet pode permitir uma leitura muito mais seletiva do que um arquivo textual tradicional.

Essa é uma de suas principais vantagens.

9. Lendo e escrevendo Parquet

Leitura:

df_parquet = spark.read.parquet(
    "/data/vendas_parquet"
)
Enter fullscreen mode Exit fullscreen mode

Escrita:

(
    df_csv.write
    .mode("overwrite")
    .parquet("/data/vendas_parquet")
)
Enter fullscreen mode Exit fullscreen mode

Agora temos um dataset armazenado em formato colunar.

O schema é preservado nos metadados do formato.

Podemos inspecioná-lo:

df_parquet.printSchema()
Enter fullscreen mode Exit fullscreen mode

E executar:

from pyspark.sql.functions import col, sum as spark_sum

resultado = (
    df_parquet
    .filter(col("estado") == "SP")
    .groupBy("cliente")
    .agg(
        spark_sum("valor").alias("total")
    )
)

resultado.show()
Enter fullscreen mode Exit fullscreen mode

Essa consulta pode se beneficiar de diferentes otimizações de leitura e execução.

10. Column Pruning

Column Pruning significa evitar a leitura de colunas que não são necessárias para a consulta.

Imagine um dataset com:

customer_id
customer_name
email
phone
address
city
state
country
product_id
product_name
quantity
price
discount
tax
shipping
...
Enter fullscreen mode Exit fullscreen mode

Mas a consulta utiliza apenas:

resultado = df_parquet.select(
    "customer_id",
    "price"
)
Enter fullscreen mode Exit fullscreen mode

Com um formato colunar e um leitor compatível, Spark pode solicitar apenas as colunas necessárias.

Conceitualmente:

PARQUET DATASET
      |
      +-- customer_id  [READ]
      +-- customer_name
      +-- email
      +-- phone
      +-- address
      +-- state
      +-- price        [READ]
      +-- tax
      +-- shipping
Enter fullscreen mode Exit fullscreen mode

Isso pode reduzir leitura, transferência e trabalho de decodificação.

A economia real depende do layout físico, da compressão, do sistema de armazenamento e do plano de execução.

11. Predicate Pushdown

Outro conceito importante é Predicate Pushdown.

Imagine:

filtrado = df_parquet.filter(
    col("estado") == "SP"
)
Enter fullscreen mode Exit fullscreen mode

O Spark pode encaminhar filtros compatíveis ao leitor da fonte.

Com Parquet, estatísticas e metadados podem ajudar a evitar a leitura de determinados grupos de dados que não satisfazem o filtro.

Conceitualmente:

PARQUET FILE
    |
    +-- Row Group A: estado = RJ
    |
    +-- Row Group B: estado = SP
    |
    +-- Row Group C: estado = MG
Enter fullscreen mode Exit fullscreen mode

Para uma consulta restrita a SP, metadados adequados podem permitir ignorar alguns grupos.

Mas há uma ressalva importante:

Predicate Pushdown não significa que todo filtro será resolvido integralmente pelo armazenamento.

O Spark ainda pode precisar avaliar linhas após a leitura.

O resultado depende do predicado, do formato, das estatísticas disponíveis e do leitor utilizado.

12. Como verificar o plano de leitura

Podemos utilizar:

filtrado.explain("formatted")
Enter fullscreen mode Exit fullscreen mode

No plano físico, vale procurar informações como:

ReadSchema
PushedFilters
PartitionFilters
DataFilters
Enter fullscreen mode Exit fullscreen mode

Esses campos ajudam a entender quais colunas estão sendo solicitadas e como os filtros estão sendo tratados.

Por exemplo, um ReadSchema contendo apenas duas colunas indica que a leitura foi planejada para solicitar somente essas colunas.

Já PushedFilters mostra filtros encaminhados à fonte.

Entretanto, a presença de um filtro no plano não comprova, sozinha, quantos bytes foram efetivamente economizados.

Para isso, precisamos combinar análise do plano com métricas de execução e medições do armazenamento.

13. ORC: outro formato colunar importante

ORC — Optimized Row Columnar — também é um formato colunar voltado a workloads analíticos.

Assim como Parquet, oferece recursos como:

  • Armazenamento tipado.
  • Compressão.
  • Estatísticas.
  • Leitura seletiva.
  • Organização eficiente para consultas analíticas.

Exemplo:

df_orc = spark.read.orc(
    "/data/vendas_orc"
)
Enter fullscreen mode Exit fullscreen mode

Escrita:

(
    df_csv.write
    .mode("overwrite")
    .orc("/data/vendas_orc")
)
Enter fullscreen mode Exit fullscreen mode

ORC possui forte presença histórica no ecossistema Hive e em determinadas arquiteturas analíticas.

Parquet e ORC são ambos formatos colunares capazes de oferecer boa performance.

A escolha deve considerar o ecossistema, a compatibilidade e os workloads reais.

Não existe um vencedor universal.

14. Comparando os quatro formatos

Característica CSV JSON Parquet ORC
Representação Texto tabular Texto estruturado Binário colunar Binário colunar
Schema tipado embutido Não Não como contrato rígido Sim Sim
Estruturas aninhadas Limitadas Sim Sim Sim
Leitura seletiva de colunas Limitada Limitada Sim Sim
Compressão e codificação colunar Não Não Sim Sim
Uso frequente Intercâmbio APIs e eventos Data Lakes analíticos Analytics e ecossistema Hive
Facilidade de leitura humana Alta Alta Baixa Baixa

A escolha depende da finalidade.

Para integração simples entre sistemas, CSV pode ser suficiente.

Para eventos e APIs, JSON pode ser conveniente.

Para processamento analítico recorrente, Parquet e ORC frequentemente oferecem vantagens importantes.

15. Compressão: menos bytes não significa sempre menos tempo

Formatos colunares podem utilizar diferentes codecs de compressão.

No Parquet, por exemplo, podemos configurar:

(
    df.write
    .option("compression", "snappy")
    .mode("overwrite")
    .parquet("/data/vendas_snappy")
)
Enter fullscreen mode Exit fullscreen mode

Também existem outros codecs, conforme o ambiente e a versão utilizados.

Compressão envolve um trade-off:

Menos dados armazenados
          |
          v
Menos bytes para transferir
          |
          v
Mais trabalho de compressão/descompressão
Enter fullscreen mode Exit fullscreen mode

Em determinadas situações, reduzir I/O compensa o custo de CPU.

Em outras, codecs mais pesados podem aumentar a latência.

Por isso, a escolha deve considerar:

  • Frequência de leitura.
  • Frequência de escrita.
  • Custo de armazenamento.
  • CPU disponível.
  • Latência.
  • Taxa de compressão.

16. Partition Pruning

Agora precisamos distinguir dois conceitos que possuem nomes semelhantes.

Na Parte 3, estudamos as partições utilizadas pelo Spark para distribuir o processamento.

Aqui vamos falar também de particionamento físico de datasets por diretórios.

Imagine um Data Lake organizado assim:

vendas/
  ano=2025/
    mes=01/
    mes=02/
  ano=2026/
    mes=01/
    mes=02/
Enter fullscreen mode Exit fullscreen mode

Podemos ler:

df = spark.read.parquet(
    "/data/vendas"
)
Enter fullscreen mode Exit fullscreen mode

E filtrar:

resultado = df.filter(
    (col("ano") == 2026)
    & (col("mes") == 2)
)
Enter fullscreen mode Exit fullscreen mode

Quando o particionamento é reconhecido corretamente, o Spark pode evitar a leitura de diretórios que não correspondem aos filtros.

Esse mecanismo é conhecido como Partition Pruning.

17. Predicate Pushdown vs. Partition Pruning

Esses mecanismos são diferentes.

Partition Pruning utiliza informações de particionamento do dataset para eliminar arquivos ou diretórios que não precisam ser considerados.

Predicate Pushdown encaminha predicados ao leitor da fonte para permitir otimizações adicionais.

Conceitualmente:

QUERY
  |
  v
Partition Pruning
  |
  v
Relevant Files
  |
  v
Predicate Pushdown
  |
  v
Relevant Data Groups / Records
  |
  v
Spark Processing
Enter fullscreen mode Exit fullscreen mode

Os dois podem trabalhar juntos.

E podem reduzir significativamente o volume de dados efetivamente processado.

18. Escrevendo dados particionados

Exemplo:

(
    df.write
    .mode("overwrite")
    .partitionBy("ano", "mes")
    .parquet("/data/vendas_particionadas")
)
Enter fullscreen mode Exit fullscreen mode

Isso pode gerar uma estrutura semelhante a:

vendas_particionadas/
  ano=2025/
    mes=1/
      part-00000-...
  ano=2026/
    mes=1/
      part-00000-...
    mes=2/
      part-00000-...
Enter fullscreen mode Exit fullscreen mode

A quantidade de arquivos depende do particionamento de execução, da distribuição dos dados e da escrita.

O partitionBy() não garante um único arquivo por diretório.

Essa distinção é importante.

19. O perigo do particionamento excessivo

Imagine particionar dados por:

customer_id
Enter fullscreen mode Exit fullscreen mode

quando existem milhões de clientes.

Podemos acabar criando um número enorme de diretórios e arquivos pequenos.

Conceitualmente:

customer_id=1/
customer_id=2/
customer_id=3/
...
customer_id=1000000/
Enter fullscreen mode Exit fullscreen mode

Isso pode gerar:

  • Sobrecarga de metadados.
  • Listagem de muitos objetos.
  • Arquivos pequenos.
  • Maior custo operacional.
  • Planejamento de leitura mais caro.

Portanto:

Particionar fisicamente um Data Lake não significa simplesmente escolher todas as colunas utilizadas em filtros.

Precisamos avaliar cardinalidade, distribuição, volume por partição e padrões reais de consulta.

20. Small Files Problem

Um dos problemas mais comuns em Data Lakes é o excesso de arquivos pequenos.

Imagine dois datasets com o mesmo volume total.

Dataset A 

100 arquivos de 100 MB
Enter fullscreen mode Exit fullscreen mode

Total aproximado:

10 GB
Enter fullscreen mode Exit fullscreen mode

Dataset B 

100.000 arquivos de 100 KB
Enter fullscreen mode Exit fullscreen mode

Total aproximado:

10 GB
Enter fullscreen mode Exit fullscreen mode

Apesar do volume semelhante, o comportamento pode ser muito diferente.

No Dataset B, Spark e o sistema de armazenamento precisam lidar com um número muito maior de objetos.

Isso pode aumentar custos de:

  • Listagem.
  • Metadados.
  • Abertura de arquivos.
  • Planejamento.
  • Scheduling.
  • Operações remotas.

Em armazenamento de objetos, esse overhead pode ser especialmente relevante.

21. Por que arquivos pequenos aparecem?

Existem várias causas possíveis:

  • Microbatches frequentes.
  • Escritas muito particionadas.
  • Muitos Executors escrevendo simultaneamente.
  • Particionamento por colunas de alta cardinalidade.
  • Pipelines incrementais sem compactação.
  • Estratégias inadequadas de saída.

Imagine um pipeline que escreve pequenos arquivos a cada minuto.

Mesmo que cada execução seja rápida, o acúmulo ao longo de meses pode produzir uma estrutura difícil de consultar.

Por isso, otimização de armazenamento não é apenas uma atividade pontual.

Também é manutenção operacional.

22. repartition() e coalesce()

Spark oferece mecanismos para modificar o número de partições de um DataFrame.

Exemplo:

df_reparticionado = df.repartition(16)
Enter fullscreen mode Exit fullscreen mode

Essa operação normalmente envolve redistribuição de dados, podendo introduzir shuffle.

Já:

df_reduzido = df.coalesce(4)
Enter fullscreen mode Exit fullscreen mode

é frequentemente utilizada para reduzir o número de partições sem um shuffle completo.

Mas existe uma ressalva:

Reduzir partições não significa necessariamente equilibrar os dados.

coalesce() pode preservar desequilíbrios e reduzir o paralelismo.

Portanto, a escolha depende do objetivo.

23. Controlando arquivos de saída

Imagine que queremos escrever um dataset com um número mais controlado de partições de execução.

Podemos utilizar:

(
    df.repartition(16)
    .write
    .mode("overwrite")
    .parquet("/data/saida")
)
Enter fullscreen mode Exit fullscreen mode

Em uma escrita simples, isso pode produzir aproximadamente um arquivo de dados por partição de escrita, mas o resultado não deve ser tratado como garantia universal.

A quantidade real de arquivos pode variar por causa de particionamento físico, limites de registros por arquivo e comportamento da plataforma.

Também existe:

(
    df.write
    .option("maxRecordsPerFile", 500_000)
    .mode("overwrite")
    .parquet("/data/saida")
)
Enter fullscreen mode Exit fullscreen mode

Esse parâmetro limita a quantidade de registros por arquivo, mas não garante um tamanho específico em MB.

O tamanho depende da estrutura e da compressibilidade dos dados.

24. Por que coalesce(1) pode ser perigoso?

É comum encontrar:

df.coalesce(1).write.csv(
    "/data/saida"
)
Enter fullscreen mode Exit fullscreen mode

A intenção geralmente é produzir um único arquivo.

Para datasets pequenos, isso pode ser aceitável.

Mas para grandes volumes, concentrar a escrita em uma única partição pode criar um gargalo.

Conceitualmente:

100 Partitions
      |
      v
coalesce(1)
      |
      v
1 Output Partition
      |
      v
Limited Parallelism
Enter fullscreen mode Exit fullscreen mode

Isso pode aumentar o tempo de escrita e a pressão de recursos.

Além disso, a API de escrita Spark normalmente produz um diretório de saída contendo arquivos de dados e metadados, não necessariamente um arquivo com o nome exato informado.

Em produção, geralmente prefiro manter a escrita distribuída e utilizar estratégias apropriadas de compactação quando necessário.

25. Um experimento reproduzível

Vamos construir um pequeno laboratório para comparar CSV, JSON, Parquet e ORC.

O objetivo não é provar que um formato será sempre mais rápido.

É aprender a medir.

Primeiro, criamos um dataset sintético:

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col,
    concat,
    lit,
    pmod
)

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("BenchmarkFormatos")
    .getOrCreate()
)

base = spark.range(0, 1_000_000)

dados = base.select(
    col("id"),
    concat(
        lit("cliente_"),
        col("id").cast("string")
    ).alias("cliente"),
    pmod(col("id"), lit(27))
        .cast("int")
        .alias("estado_codigo"),
    (pmod(col("id"), lit(10_000)) / 100.0)
        .alias("valor")
)
Enter fullscreen mode Exit fullscreen mode

Agora escrevemos nos quatro formatos:

diretorio = "/tmp/spark-format-benchmark"

(
    dados.write
    .mode("overwrite")
    .option("header", True)
    .csv(f"{diretorio}/csv")
)

(
    dados.write
    .mode("overwrite")
    .json(f"{diretorio}/json")
)

(
    dados.write
    .mode("overwrite")
    .parquet(f"{diretorio}/parquet")
)

(
    dados.write
    .mode("overwrite")
    .orc(f"{diretorio}/orc")
)
Enter fullscreen mode Exit fullscreen mode

Esse diretório é apropriado para um experimento local. Em um cluster, os caminhos precisam ser acessíveis conforme o ambiente de execução.

26. Medindo o tempo de leitura e consulta

Vamos utilizar uma função simples de benchmark.

from time import perf_counter

def medir_consulta(nome, carregar):
    inicio = perf_counter()

    df = carregar()

    resultado = (
        df
        .filter(col("estado_codigo") == 10)
        .select("id", "valor")
        .agg({"valor": "sum"})
        .collect()
    )

    duracao = perf_counter() - inicio

    print(
        f"{nome}: {duracao:.3f}s | "
        f"resultado={resultado}"
    )

    return duracao
Enter fullscreen mode Exit fullscreen mode

Para evitar que o CSV dependa de inferência de schema, utilizamos o schema conhecido:

schema = dados.schema
Enter fullscreen mode Exit fullscreen mode

Agora executamos:

tempos = {}

tempos["CSV"] = medir_consulta(
    "CSV",
    lambda: (
        spark.read
        .option("header", True)
        .schema(schema)
        .csv(f"{diretorio}/csv")
    )
)

tempos["JSON"] = medir_consulta(
    "JSON",
    lambda: (
        spark.read
        .schema(schema)
        .json(f"{diretorio}/json")
    )
)

tempos["Parquet"] = medir_consulta(
    "Parquet",
    lambda: spark.read.parquet(
        f"{diretorio}/parquet"
    )
)

tempos["ORC"] = medir_consulta(
    "ORC",
    lambda: spark.read.orc(
        f"{diretorio}/orc"
    )
)
Enter fullscreen mode Exit fullscreen mode

Os resultados dependem da máquina, do sistema de arquivos, da versão do Spark e das condições de execução.

Não devemos apresentar tempos hipotéticos como resultados medidos.

27. Como tornar o benchmark mais confiável

Uma única execução não é suficiente para conclusões definitivas.

Existem fatores como:

  • Inicialização da JVM.
  • Aquecimento de código.
  • Cache do sistema operacional.
  • Cache de metadados.
  • Concorrência.
  • Variações de disco.
  • Quantidade de arquivos.
  • Compressão.
  • Planejamento da consulta.

Para um benchmark mais rigoroso, devemos:

  1. Executar uma rodada de aquecimento.
  2. Repetir cada cenário várias vezes.
  3. Alternar a ordem dos formatos.
  4. Registrar mediana e dispersão.
  5. Comparar volumes físicos.
  6. Inspecionar os planos de execução.
  7. Monitorar CPU, I/O e memória.
  8. Utilizar datasets representativos.

Também é importante separar dois experimentos diferentes:

Benchmark de escrita: mede o custo de produzir cada formato.

Benchmark de leitura: mede o custo de consultar dados já armazenados.

Misturar os dois pode esconder trade-offs importantes.

28. Como inspecionar as otimizações

Vamos comparar os planos físicos.

df_csv = (
    spark.read
    .option("header", True)
    .schema(schema)
    .csv(f"{diretorio}/csv")
)

df_parquet = spark.read.parquet(
    f"{diretorio}/parquet"
)
Enter fullscreen mode Exit fullscreen mode

Definimos consultas equivalentes:

consulta_csv = (
    df_csv
    .filter(col("estado_codigo") == 10)
    .select("id", "valor")
)

consulta_parquet = (
    df_parquet
    .filter(col("estado_codigo") == 10)
    .select("id", "valor")
)
Enter fullscreen mode Exit fullscreen mode

Agora:

consulta_csv.explain("formatted")
Enter fullscreen mode Exit fullscreen mode

E:

consulta_parquet.explain("formatted")
Enter fullscreen mode Exit fullscreen mode

Compare:

ReadSchema
PushedFilters
DataFilters
Enter fullscreen mode Exit fullscreen mode

A análise mostra como o Spark planeja a leitura de cada formato.

O objetivo não é apenas medir segundos.

É compreender por que os resultados são diferentes.

29. O formato influencia o custo do Data Lake

Imagine um Data Lake que armazena centenas de terabytes.

O formato influencia:

Storage Size
     |
     v
Data Scanned
     |
     v
I/O Operations
     |
     v
Compute Time
     |
     v
Infrastructure Cost
Enter fullscreen mode Exit fullscreen mode

Um formato colunar bem organizado pode reduzir a quantidade de dados lidos em determinadas consultas.

Isso pode diminuir:

  • Tempo de processamento.
  • Tráfego de dados.
  • Utilização de CPU.
  • Custo de armazenamento.
  • Custo computacional.

Mas a economia depende do workload.

Uma transformação que precisa ler todas as colunas e todos os registros pode apresentar ganhos diferentes de uma consulta altamente seletiva.

30. Parquet não resolve tudo

É importante evitar a conclusão:

Basta converter tudo para Parquet e a performance estará resolvida.

Não é assim.

Podemos ter um dataset Parquet com:

  • Milhões de arquivos pequenos.
  • Particionamento físico inadequado.
  • Data skew.
  • Compressão mal escolhida.
  • Schemas inconsistentes.
  • Estatísticas pouco úteis.
  • Leituras repetidas desnecessárias.
  • Joins caros.
  • Layout incompatível com os filtros.

O formato é apenas uma dimensão da arquitetura.

Uma organização ruim pode comprometer parte das vantagens do armazenamento colunar.

31. Formato de arquivo vs. formato de tabela

Outra distinção importante nas arquiteturas modernas é separar o formato físico dos arquivos do formato de tabela.

Por exemplo:

Apache Parquet
Enter fullscreen mode Exit fullscreen mode

é um formato de arquivo.

Já tecnologias como:

Apache Iceberg
Delta Lake
Apache Hudi
Enter fullscreen mode Exit fullscreen mode

adicionam mecanismos de gerenciamento de tabelas e metadados, com capacidades que variam entre elas.

Esses mecanismos podem envolver:

  • Controle de snapshots.
  • Evolução de schema.
  • Gerenciamento de arquivos.
  • Operações transacionais.
  • Evolução de partições.
  • Manutenção e compactação.

Muitas dessas tecnologias utilizam Parquet como formato físico de dados.

Portanto:

TABLE FORMAT
     |
     v
TABLE METADATA
     |
     v
DATA FILES
     |
     v
PARQUET
Enter fullscreen mode Exit fullscreen mode

Essa separação é fundamental para compreender arquiteturas Lakehouse.

Mas formatos de tabela merecem um capítulo próprio.

32. Como escolher um formato?

Minha decisão começa pelo objetivo.

Intercâmbio simples entre sistemas

CSV pode ser suficiente quando o contrato é claro e o volume é administrável.

Eventos, APIs e estruturas semiestruturadas

JSON pode ser conveniente, especialmente em camadas de ingestão.

Consultas analíticas recorrentes

Parquet costuma ser uma excelente opção inicial.

Ambientes com integração forte ao ecossistema ORC/Hive

ORC pode ser uma escolha natural.

Tabelas analíticas com requisitos transacionais e evolução operacional

Vale avaliar um formato de tabela apropriado, além do formato físico.

Não se trata de escolher a tecnologia mais sofisticada.

Trata-se de escolher a representação adequada ao uso.

33. Uma arquitetura por camadas

Em um Data Lake moderno, podemos encontrar:

                   DATA SOURCES
                        |
          +-------------+-------------+
          |             |             |
          v             v             v
         APIs         Files         Events
          |             |             |
          +-------------+-------------+
                        |
                        v
                    RAW LAYER
                 JSON / CSV / etc.
                        |
                        v
                 VALIDATION & ETL
                        |
                        v
                 CURATED LAYER
                    PARQUET
                        |
                        v
                 ANALYTICAL TABLES
                        |
                        v
                 SQL / BI / ML / AI
Enter fullscreen mode Exit fullscreen mode

Essa é uma arquitetura conceitual, não uma regra universal.

Dependendo do sistema, podemos preservar dados brutos em formatos originais e utilizar formatos colunares em camadas de consumo.

Também podemos utilizar formatos de tabela para gerenciar datasets analíticos.

O importante é que cada camada tenha uma responsabilidade clara.

34. Checklist de produção

Antes de definir o armazenamento de um pipeline Spark, gosto de responder:

  • Qual é a origem dos dados?
  • O schema é estável?
  • Existem estruturas aninhadas?
  • Quais colunas são consultadas?
  • Quais filtros são mais frequentes?
  • Qual é a cardinalidade das colunas candidatas a particionamento?
  • Quantos arquivos serão produzidos?
  • Qual é o tamanho médio dos arquivos?
  • Existe necessidade de compactação?
  • Quais codecs são suportados?
  • O formato permite column pruning?
  • Existe predicate pushdown?
  • O particionamento permite pruning?
  • Como será tratada a evolução de schema?
  • Existe necessidade de transações ou snapshots?
  • Qual é o custo de armazenamento e leitura?
  • Como a performance será medida?

Essas perguntas ajudam a evitar decisões baseadas apenas em preferência tecnológica.

35. A principal lição deste capítulo

O Spark pode executar exatamente a mesma lógica sobre formatos diferentes.

Mas o trabalho necessário para obter os dados pode variar bastante.

Uma consulta analítica não depende apenas de:

CPU
Memory
Executors
Partitions
Enter fullscreen mode Exit fullscreen mode

Ela também depende de:

File Format
Compression
Metadata
Column Selection
Filter Pushdown
Physical Partitioning
File Size
Enter fullscreen mode Exit fullscreen mode

Por isso:

Performance em Spark começa antes da execução: começa na forma como os dados são armazenados, organizados e disponibilizados para leitura.

36. Como penso hoje

Quando recebo um pipeline Spark lento, não começo necessariamente aumentando recursos.

Primeiro investigo:

Qual formato está sendo lido?
            |
            v
Quantos arquivos existem?
            |
            v
Qual é o tamanho médio?
            |
            v
O schema está correto?
            |
            v
Todas as colunas são necessárias?
            |
            v
Os filtros estão sendo aproveitados?
            |
            v
Existe partition pruning?
            |
            v
O plano físico está adequado?
            |
            v
Existe shuffle desnecessário?
            |
            v
Qual é o gargalo real?
Enter fullscreen mode Exit fullscreen mode

Às vezes, melhorar o layout dos dados traz mais benefício do que adicionar Executors.

E essa é uma das lições mais importantes da Engenharia de Dados moderna.

Próximo artigo

Agora entendemos como CSV, JSON, Parquet e ORC influenciam o processamento.

Mas existe uma evolução importante que precisamos explorar:

Como transformar arquivos em tabelas analíticas confiáveis?

No próximo capítulo, vamos entrar no universo das arquiteturas Lakehouse e dos formatos de tabela.

Vamos estudar:

  • Data Lake vs. Data Warehouse vs. Lakehouse.
  • Apache Iceberg.
  • Delta Lake.
  • Apache Hudi.
  • Formatos de arquivo vs. formatos de tabela.
  • Snapshots.
  • Evolução de schema.
  • ACID.
  • Time Travel.
  • Compactação.
  • Gerenciamento de metadados.
  • Estratégias de atualização incremental.
  • Integração com Spark SQL.

A pergunta central será:

Por que armazenar arquivos Parquet não é suficiente para construir uma plataforma analítica transacional e governável?

Série: Como aprendi Apache Spark

← Parte 11: Dimensionamento de recursos, paralelismo e custos

Próximo: Parte 13 — Apache Iceberg, Delta Lake e Hudi: a evolução do Data Lake para Lakehouse

Top comments (0)