A necessidade de otimizar a escrita no Apache Spark

Artigo
01/10/2024

As cargas de trabalho analíticas em mecanismos de processamento de Big Data, como o Apache Spark, têm um desempenho mais eficiente ao usar tamanhos de arquivo maiores padronizados. A relação entre o tamanho do arquivo, o número de arquivos, o número de trabalhadores do Spark e suas configurações, desempenham um papel crítico no desempenho. A ingestão de cargas de trabalho em tabelas de data lake pode ter a característica herdada de escrever constantemente muitos arquivos pequenos; Este cenário é comumente conhecido como o "problema de arquivo pequeno".

Otimize Write é um recurso Delta Lake on Synapse que reduz o número de arquivos gravados e visa aumentar o tamanho do arquivo individual dos dados gravados. Ele otimiza dinamicamente partições enquanto gera arquivos com um tamanho padrão de 128 MB. O tamanho do arquivo de destino pode ser alterado de acordo com os requisitos de carga de trabalho usando configurações.

Esse recurso atinge o tamanho do arquivo usando uma fase extra de embaralhamento de dados sobre partições, causando um custo de processamento extra durante a gravação dos dados. A pequena penalidade de escrita deve ser superada pela eficiência de leitura nas tabelas.

Nota

Está disponível em Synapse Pools para versões do Apache Spark acima de 3.1.

Benefícios de otimizar gravações

Está disponível em tabelas Delta Lake para padrões de gravação em lote e streaming.
Não há necessidade de alterar o spark.write padrão de comando. O recurso é habilitado por uma definição de configuração ou uma propriedade de tabela.
Ele reduz o número de transações de gravação em comparação com o comando OTIMIZE.
As operações OTIMIZE serão mais rápidas, pois operarão em menos arquivos.
O comando VACUUM para exclusão de arquivos antigos não referenciados também funcionará mais rapidamente.
As consultas verificarão menos arquivos com tamanhos de arquivo mais ideais, melhorando o desempenho de leitura ou o uso de recursos.

Otimizar cenários de uso de gravação

Quando Utilizar

Tabelas particionadas Delta lake sujeitas a padrões de gravação que geram tamanhos de arquivos subótimos (menos de 128 MB) ou não padronizados (arquivos com tamanhos diferentes entre si).
Quadros de dados reparticionados que serão gravados no disco com tamanho de arquivo abaixo do ideal.
Tabelas particionadas Delta lake direcionadas por comandos SQL em pequenos lotes como UPDATE, DELETE, MERGE, CREATE TABLE AS SELECT, INSERT INTO, etc.
Cenários de ingestão de streaming com padrões de dados anexados a tabelas particionadas do lago Delta onde a latência de gravação extra é tolerável.

Quando evitá-lo

Tabelas não particionadas.
Casos de uso em que a latência de gravação extra não é aceitável.
Grandes tabelas com cronogramas de otimização bem definidos e padrões de leitura.

Como ativar e desativar o recurso otimizar gravação

O recurso otimizar gravação está desativado por padrão. No Pool do Spark 3.3, ele é habilitado por padrão para tabelas particionadas.

Depois que a configuração for definida para o pool ou sessão, todos os padrões de gravação do Spark usarão a funcionalidade.

Para usar o recurso otimizar gravação, habilite-o usando a seguinte configuração:

Scala e PySpark

spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", "true")

SQL do Apache Spark

SET `spark.microsoft.delta.optimizeWrite.enabled` = true

Para verificar o valor de configuração atual, use o comando conforme mostrado:

Scala e PySpark

spark.conf.get("spark.microsoft.delta.optimizeWrite.enabled")

SQL do Apache Spark

SET `spark.microsoft.delta.optimizeWrite.enabled`

Para desativar o recurso otimizar gravação, altere a seguinte configuração conforme mostrado:

Scala e PySpark

spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", "false")

SQL do Apache Spark

SET `spark.microsoft.delta.optimizeWrite.enabled` = false

Controlando otimizar a gravação usando propriedades de tabela

Em novas tabelas

CREATE TABLE <table_name> TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true)

Scala

Usando a API do DeltaTableBuilder:

val table = DeltaTable.create()
  .tableName("<table_name>")
  .addColumnn("<colName>", <dataType>)
  .location("<table_location>")
  .property("delta.autoOptimize.optimizeWrite", "true") 
  .execute()

Nos quadros existentes

ALTER TABLE <table_name> SET TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true)

Scala

Usando a API do DeltaTableBuilder

val table = DeltaTable.replace()
  .tableName("<table_name>")
  .location("<table_location>")
  .property("delta.autoOptimize.optimizeWrite", "true") 
  .execute()

Como obter & alterar a configuração atual de tamanho máximo de arquivo para Otimizar gravação

Para obter o valor de configuração atual, use os comandos abaixo. O padrão é 128 MB.

Scala e PySpark

spark.conf.get("spark.microsoft.delta.optimizeWrite.binSize")

SET `spark.microsoft.delta.optimizeWrite.binSize`

Para alterar o valor de configuração

Scala e PySpark

spark.conf.set("spark.microsoft.delta.optimizeWrite.binSize", "134217728")

SET `spark.microsoft.delta.optimizeWrite.binSize` = 134217728

Partilhar via