O tempo de execução do Amazon EMR 7.5 para Apache Spark e Iceberg pode executar cargas de trabalho do Spark 3,6 vezes mais rápido que o Spark 3.5.3 e o Iceberg 1.6.1

O Tempo de execução do Amazon EMR para Apache Spark oferece um ambiente de tempo de execução de alto desempenho, mantendo 100% de compatibilidade de API com formato de tabela de código aberto Apache Spark e Apache Iceberg. Amazon EMR no EC2, Amazon EMR sem servidor, Amazon EMR no Amazon EKS, Amazon EMR no AWS Outposts e Cola AWS todos usam tempos de execução otimizados.

Nesta postagem, demonstramos os benefícios de desempenho do uso do Amazon EMR 7.5 para Spark e Iceberg em comparação com Spark 3.5.3 de código aberto com tabelas Iceberg 1.6.1 no benchmark TPC-DS 3TB v2.13.

Iceberg é um formato common de código aberto e de alto desempenho para grandes tabelas analíticas. Nossos benchmarks demonstram que o Amazon EMR pode executar cargas de trabalho TPC-DS de 3 TB 3,6 vezes mais rápido, reduzindo o tempo de execução de 1,54 horas para 0,42 horas. Além disso, a eficiência de custos melhora 2,9 vezes, com o custo complete diminuindo de US$ 16,00 para US$ 5,39 ao usar Amazon Elastic Compute Nuvem (Amazon EC2) Instâncias r5d.4xlarge sob demanda, proporcionando ganhos observáveis ​​para tarefas de processamento de dados.

Este é um aumento adicional de 32% em relação às otimizações fornecidas no Amazon EMR 7.1 abordadas em uma postagem anterior, O tempo de execução do Amazon EMR 7.1 para Apache Spark e Iceberg pode executar cargas de trabalho do Spark 2,7 vezes mais rápido que o Apache Spark 3.5.1 e o Iceberg 1.5.2. Desde então, continuamos adicionando mais suporte ao DataSource V2 para mais oito otimizações de consulta existentes no tempo de execução do EMR para Spark.

Além dessas melhorias específicas do DataSource V2, fizemos mais otimizações nos operadores Spark desde o Amazon EMR 7.1, o que também contribui para a aceleração adicional.

Resultados de benchmark do Amazon EMR 7.5 em comparação com o Spark 3.5.3 de código aberto e o Iceberg 1.6.1

Para avaliar o desempenho do motor Spark com o formato de tabela Iceberg, realizamos testes de benchmark usando o Conjunto de dados TPC-DS de 3 TB, versão 2.13 (nossos resultados derivados do conjunto de dados TPC-DS não são diretamente comparáveis ​​aos resultados oficiais do TPC-DS devido a diferenças de configuração). Os testes de benchmark para o tempo de execução do EMR para Spark e Iceberg foram realizados em clusters do Amazon EMR 7.5 EC2 versus Spark 3.5.3 de código aberto e Iceberg 1.6.1 em clusters do EC2.

As instruções de configuração e detalhes técnicos estão disponíveis em nosso Repositório GitHub. Para minimizar a influência de catálogos externos como Cola AWS e Hive, utilizamos o catálogo Hadoop para as tabelas Iceberg. Isso usa o sistema de arquivos subjacente, especificamente o Amazon S3, como catálogo. Podemos definir esta configuração configurando a propriedade spark.sql.catalog..sort. As tabelas de fatos usaram o particionamento padrão pela coluna de information, que possui um número de partições variando de 200 a 2.100. Nenhuma estatística pré-calculada foi usada para essas tabelas.

Executamos um complete de 104 consultas SparkSQL em três rodadas sequenciais, e o tempo médio de execução de cada consulta nessas rodadas foi obtido para comparação. O tempo médio de execução das três rodadas no Amazon EMR 7.5 com o Iceberg habilitado foi de 0,42 horas, demonstrando um aumento de velocidade de 3,6 vezes em comparação com o Spark 3.5.3 de código aberto e o Iceberg 1.6.1. A figura a seguir apresenta os tempos totais de execução em segundos.

A tabela a seguir resume as métricas.

*Estimativas de custos detalhadas são discutidas posteriormente nesta postagem.

O gráfico a seguir demonstra a melhoria de desempenho por consulta do Amazon EMR 7.5 em relação ao Spark 3.5.3 de código aberto e ao Iceberg 1.6.1. A extensão da aceleração varia de uma consulta para outra, com a mais rápida até 9,4 vezes mais rápida para q93, com o Amazon EMR superando o Spark de código aberto com tabelas Iceberg. O eixo horizontal organiza as consultas de benchmark do TPC-DS 3TB em ordem decrescente com base na melhoria de desempenho observada com o Amazon EMR, e o eixo vertical representa a magnitude dessa aceleração como uma proporção.

Comparação de custos

Nosso benchmark fornece dados de tempo de execução complete e média geométrica para avaliar o desempenho do Spark e do Iceberg em um cenário complexo de suporte à decisão do mundo actual. Para obter informações adicionais, também examinamos o aspecto do custo. Calculamos estimativas de custos usando fórmulas que levam em conta instâncias sob demanda do EC2, Loja de blocos elásticos da Amazon (Amazon EBS) e despesas do Amazon EMR.

• Custo do Amazon EC2 (inclui custo de SSD) = número de instâncias * taxa horária r5d.4xlarge * tempo de execução do trabalho em horas
  • taxa horária r5d.4xlarge = US$ 1,152 por hora em us-east-1
• Custo raiz do Amazon EBS = número de instâncias * Amazon EBS por taxa horária de GB * tamanho do quantity raiz EBS * tempo de execução do trabalho em horas
• Custo do Amazon EMR = número de instâncias * r5d.4xlarge Custo do Amazon EMR * tempo de execução do trabalho em horas
  • Custo 4xlarge do Amazon EMR = US$ 0,27 por hora
• Custo complete = custo do Amazon EC2 + custo raiz do Amazon EBS + custo do Amazon EMR

Os cálculos revelam que o benchmark Amazon EMR 7.5 produz uma melhoria de eficiência de custos de 2,9 vezes em relação ao Spark 3.5.3 de código aberto e ao Iceberg 1.6.1 na execução do trabalho de benchmark.

Além das métricas baseadas em tempo discutidas até agora, os dados dos logs de eventos do Spark mostram que o Amazon EMR digitalizou aproximadamente 3,4 vezes menos dados do Amazon S3 e 4,1 vezes menos registros do que a versão de código aberto no benchmark TPC-DS de 3 TB. Essa redução na verificação de dados do Amazon S3 contribui diretamente para a economia de custos das cargas de trabalho do Amazon EMR.

Execute benchmarks Spark de código aberto em tabelas Iceberg

Usamos clusters EC2 separados, cada um equipado com nove instâncias r5d.4xlarge, para testar o Spark 3.5.3 de código aberto e o Amazon EMR 7.5 para carga de trabalho Iceberg. O nó primário estava equipado com 16 vCPU e 128 GB de memória, e os oito nós de trabalho juntos tinham 128 vCPU e 1.024 GB de memória. Realizamos testes usando as configurações padrão do Amazon EMR para mostrar a experiência típica do usuário e ajustamos minimamente as configurações do Spark e do Iceberg para manter uma comparação equilibrada.

A tabela a seguir resume as configurações do Amazon EC2 para o nó primário e oito nós de trabalho do tipo r5d.4xlarge.

Pré-requisitos

Os seguintes pré-requisitos são necessários para executar o benchmarking:

1. Usando as instruções do repositório GitHub emr-spark-benchmarkconfigure os dados de origem TPC-DS no bucket S3 e no computador native.
2. Crie o aplicativo de benchmark seguindo as etapas fornecidas em Etapas para construir um aplicativo spark-benchmark-assembly e copie o aplicativo de benchmark para seu bucket S3. Alternativamente, copie spark-benchmark-assembly-3.5.3.jar para o seu balde S3.
3. Crie tabelas Iceberg a partir dos dados de origem do TPC-DS. Siga as instruções em GitHub para criar tabelas Iceberg usando o catálogo Hadoop. Por exemplo, o código a seguir usa um cluster do EMR 7.5 com Iceberg habilitado para criar as tabelas:

aws emr add-steps 
--cluster-id  --steps Sort=Spark,Title="Create Iceberg Tables",
Args=(--class,com.amazonaws.eks.tpcds.CreateIcebergTables,--conf,spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,
--conf,spark.sql.catalog.hadoop_catalog=org.apache.iceberg.spark.SparkCatalog,
--conf,spark.sql.catalog.hadoop_catalog.sort=hadoop,
--conf,spark.sql.catalog.hadoop_catalog.warehouse=s3:////,
--conf,spark.sql.catalog.hadoop_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO,
s3:////spark-benchmark-assembly-3.5.3.jar,s3://blogpost-sparkoneks-us-east-1/weblog/BLOG_TPCDS-TEST-3T-partitioned/,
/dwelling/hadoop/tpcds-kit/instruments,parquet,3000,true,,true,true),ActionOnFailure=CONTINUE --region 

Observe o native do armazém do catálogo Hadoop e o nome do banco de dados da etapa anterior. Usamos as mesmas tabelas iceberg para executar benchmarks com o Amazon EMR 7.5 e o Spark de código aberto.

Este aplicativo de benchmark é construído a partir do ramo tpcds-v2.13_iceberg. Se você estiver criando um novo aplicativo de benchmark, mude para o department correto depois de baixar o código-fonte do repositório GitHub.

Crie e configure um cluster YARN no Amazon EC2

Para comparar o desempenho do Iceberg entre o Amazon EMR no Amazon EC2 e o Spark de código aberto no Amazon EC2, siga as instruções no repositório GitHub emr-spark-benchmark para criar um cluster Spark de código aberto no Amazon EC2 usando Flintrock com oito nós de trabalho.

Com base na seleção do cluster para este teste, as seguintes configurações são usadas:

Certifique-se de substituir o espaço reservado no yarn-site.xml arquivo, com o endereço IP do nó primário do seu cluster Flintrock.

Execute o benchmark TPC-DS com Spark 3.5.3 e Iceberg 1.6.1

Conclua as etapas a seguir para executar o benchmark TPC-DS:

1. Faça login no nó primário do cluster de código aberto usando flintrock login $CLUSTER_NAME.
2. Envie seu trabalho do Spark:
   1. Escolha o native correto do armazém do catálogo Iceberg e o banco de dados que contém as tabelas Iceberg criadas.
   2. Os resultados são criados em s3:///benchmark_run.
   3. Você pode acompanhar o progresso em /media/ephemeral0/spark_run.log.

spark-submit 
--master yarn 
--deploy-mode consumer 
--class com.amazonaws.eks.tpcds.BenchmarkSQL 
--conf spark.driver.cores=4 
--conf spark.driver.reminiscence=10g 
--conf spark.executor.cores=16 
--conf spark.executor.reminiscence=100g 
--conf spark.executor.cases=8 
--conf spark.community.timeout=2000 
--conf spark.executor.heartbeatInterval=300s 
--conf spark.dynamicAllocation.enabled=false 
--conf spark.shuffle.service.enabled=false 
--conf spark.hadoop.fs.s3a.aws.credentials.supplier=com.amazonaws.auth.InstanceProfileCredentialsProvider 
--conf spark.hadoop.fs.s3.impl=org.apache.hadoop.fs.s3a.S3AFileSystem 
--conf spark.jars.packages=org.apache.hadoop:hadoop-aws:3.3.4,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,org.apache.iceberg:iceberg-aws-bundle:1.6.1 
--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions   
--conf spark.sql.catalog.native=org.apache.iceberg.spark.SparkCatalog    
--conf spark.sql.catalog.native.sort=hadoop  
--conf spark.sql.catalog.native.warehouse=s3a://// 
--conf spark.sql.defaultCatalog=native   
--conf spark.sql.catalog.native.io-impl=org.apache.iceberg.aws.s3.S3FileIO   
spark-benchmark-assembly-3.5.3.jar   
s3:///benchmark_run 3000 1 false  
q1-v2.13,q10-v2.13,q11-v2.13,q12-v2.13,q13-v2.13,q14a-v2.13,q14b-v2.13,q15-v2.13,q16-v2.13,
q17-v2.13,q18-v2.13,q19-v2.13,q2-v2.13,q20-v2.13,q21-v2.13,q22-v2.13,q23a-v2.13,q23b-v2.13,
q24a-v2.13,q24b-v2.13,q25-v2.13,q26-v2.13,q27-v2.13,q28-v2.13,q29-v2.13,q3-v2.13,q30-v2.13,
q31-v2.13,q32-v2.13,q33-v2.13,q34-v2.13,q35-v2.13,q36-v2.13,q37-v2.13,q38-v2.13,q39a-v2.13,
q39b-v2.13,q4-v2.13,q40-v2.13,q41-v2.13,q42-v2.13,q43-v2.13,q44-v2.13,q45-v2.13,q46-v2.13,
q47-v2.13,q48-v2.13,q49-v2.13,q5-v2.13,q50-v2.13,q51-v2.13,q52-v2.13,q53-v2.13,q54-v2.13,
q55-v2.13,q56-v2.13,q57-v2.13,q58-v2.13,q59-v2.13,q6-v2.13,q60-v2.13,q61-v2.13,q62-v2.13,
q63-v2.13,q64-v2.13,q65-v2.13,q66-v2.13,q67-v2.13,q68-v2.13,q69-v2.13,q7-v2.13,q70-v2.13,
q71-v2.13,q72-v2.13,q73-v2.13,q74-v2.13,q75-v2.13,q76-v2.13,q77-v2.13,q78-v2.13,q79-v2.13,
q8-v2.13,q80-v2.13,q81-v2.13,q82-v2.13,q83-v2.13,q84-v2.13,q85-v2.13,q86-v2.13,q87-v2.13,
q88-v2.13,q89-v2.13,q9-v2.13,q90-v2.13,q91-v2.13,q92-v2.13,q93-v2.13,q94-v2.13,q95-v2.13,
q96-v2.13,q97-v2.13,q98-v2.13,q99-v2.13,ss_max-v2.13    
true  > /media/ephemeral0/spark_run.log 2>&1 &!

Resuma os resultados

Após a conclusão do trabalho do Spark, recupere o arquivo de resultado do teste do bucket S3 de saída em s3:///benchmark_run/timestamp=xxxx/abstract.csv/xxx.csv. Isso pode ser feito por meio do console do Amazon S3 navegando até o native do bucket especificado ou usando o comando Interface de linha de comando da AWS (AWS CLI). O aplicativo de benchmark Spark organiza os dados criando uma pasta de carimbo de information/hora e colocando um arquivo de resumo em uma pasta chamada abstract.csv. Os arquivos CSV de saída contêm quatro colunas sem cabeçalhos:

• Nome da consulta
• Tempo médio
• Tempo mínimo
• Tempo máximo

Com os dados de três execuções de teste separadas com uma iteração de cada vez, podemos calcular a média e a média geométrica dos tempos de execução do benchmark.

Execute o benchmark TPC-DS com o tempo de execução EMR para Spark

A maioria das instruções é semelhante a Etapas para executar o Spark Benchmarking com alguns detalhes específicos do Iceberg.

Pré-requisitos

Conclua as seguintes etapas de pré-requisito:

1. Correr aws configure para configurar o shell da AWS CLI para apontar para a conta da AWS de benchmarking. Consulte Configurar a AWS CLI para obter instruções.
2. Carregue o arquivo JAR do aplicativo de benchmark no Amazon S3.

Implante o cluster do EMR e execute o job de benchmark

Conclua as etapas a seguir para executar o trabalho de referência:

1. Use o comando AWS CLI conforme mostrado em Implante o EMR no cluster EC2 e execute o trabalho de benchmark para ativar um EMR no cluster EC2. Certifique-se de ativar o Iceberg. Ver Crie um cluster Iceberg para mais detalhes. Escolha a versão correta do Amazon EMR, o tamanho do quantity raiz e a mesma configuração de recursos da configuração do Flintrock de código aberto. Consulte criar cluster para obter uma descrição detalhada das opções da AWS CLI.
2. Armazene o ID do cluster da resposta. Precisamos disso para a próxima etapa.
3. Envie o trabalho de referência no Amazon EMR usando add-steps da AWS CLI:
   1. Substituir com o ID do cluster da Etapa 2.
   2. A aplicação de benchmark está em s3:///spark-benchmark-assembly-3.5.3.jar.
   3. Escolha o native correto do armazém do catálogo Iceberg e o banco de dados que contém as tabelas Iceberg criadas. Deve ser o mesmo usado para a execução do benchmark TPC-DS de código aberto.
   4. Os resultados estarão em s3:///benchmark_run.

aws emr add-steps   --cluster-id 
--steps Sort=Spark,Title="SPARK Iceberg EMR TPCDS Benchmark Job",
Args=(--class,com.amazonaws.eks.tpcds.BenchmarkSQL,
--conf,spark.driver.cores=4,
--conf,spark.driver.reminiscence=10g,
--conf,spark.executor.cores=16,
--conf,spark.executor.reminiscence=100g,
--conf,spark.executor.cases=8,
--conf,spark.community.timeout=2000,
--conf,spark.executor.heartbeatInterval=300s,
--conf,spark.dynamicAllocation.enabled=false,
--conf,spark.shuffle.service.enabled=false,
--conf,spark.sql.iceberg.data-prefetch.enabled=true,
--conf,spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,
--conf,spark.sql.catalog.native=org.apache.iceberg.spark.SparkCatalog,
--conf,spark.sql.catalog.native.sort=hadoop,
--conf,spark.sql.catalog.native.warehouse=s3:///,
--conf,spark.sql.defaultCatalog=native,
--conf,spark.sql.catalog.native.io-impl=org.apache.iceberg.aws.s3.S3FileIO,
s3:///spark-benchmark-assembly-3.5.3.jar,
s3:///benchmark_run,3000,1,false,
'q1-v2.13,q10-v2.13,q11-v2.13,q12-v2.13,q13-v2.13,q14a-v2.13,q14b-v2.13,q15-v2.13,q16-v2.13,q17-v2.13,q18-v2.13,q19-v2.13,q2-v2.13,q20-v2.13,q21-v2.13,q22-v2.13,q23a-v2.13,q23b-v2.13,q24a-v2.13,q24b-v2.13,q25-v2.13,q26-v2.13,q27-v2.13,q28-v2.13,q29-v2.13,q3-v2.13,q30-v2.13,q31-v2.13,q32-v2.13,q33-v2.13,q34-v2.13,q35-v2.13,q36-v2.13,q37-v2.13,q38-v2.13,q39a-v2.13,q39b-v2.13,q4-v2.13,q40-v2.13,q41-v2.13,q42-v2.13,q43-v2.13,q44-v2.13,q45-v2.13,q46-v2.13,q47-v2.13,q48-v2.13,q49-v2.13,q5-v2.13,q50-v2.13,q51-v2.13,q52-v2.13,q53-v2.13,q54-v2.13,q55-v2.13,q56-v2.13,q57-v2.13,q58-v2.13,q59-v2.13,q6-v2.13,q60-v2.13,q61-v2.13,q62-v2.13,q63-v2.13,q64-v2.13,q65-v2.13,q66-v2.13,q67-v2.13,q68-v2.13,q69-v2.13,q7-v2.13,q70-v2.13,q71-v2.13,q72-v2.13,q73-v2.13,q74-v2.13,q75-v2.13,q76-v2.13,q77-v2.13,q78-v2.13,q79-v2.13,q8-v2.13,q80-v2.13,q81-v2.13,q82-v2.13,q83-v2.13,q84-v2.13,q85-v2.13,q86-v2.13,q87-v2.13,q88-v2.13,q89-v2.13,q9-v2.13,q90-v2.13,q91-v2.13,q92-v2.13,q93-v2.13,q94-v2.13,q95-v2.13,q96-v2.13,q97-v2.13,q98-v2.13,q99-v2.13,ss_max-v2.13',
true,),ActionOnFailure=CONTINUE --region 

Resuma os resultados

Após a conclusão da etapa, você poderá ver o resultado resumido do benchmark em s3:///benchmark_run/timestamp=xxxx/abstract.csv/xxx.csv da mesma forma que na execução anterior e calcule a média e a média geométrica dos tempos de execução da consulta.

Limpar

Para evitar cobranças futuras, exclua os recursos criados seguindo as instruções fornecidas no Seção de limpeza do repositório GitHub.

Resumo

O Amazon EMR aprimora consistentemente o tempo de execução do EMR para Spark quando usado com tabelas Iceberg, alcançando um desempenho 3,6 vezes mais rápido que o Spark 3.5.3 de código aberto e o Iceberg 1.6.1 com EMR 7.5 no TPC-DS 3 TB, v2.13. Este é um aumento adicional de 32% em relação ao EMR 7.1. Incentivamos você a se manter atualizado com as versões mais recentes do Amazon EMR para aproveitar ao máximo as melhorias contínuas de desempenho.

Para se manter informado, inscreva-se no weblog AWS Massive Knowledge Feed RSSonde você pode encontrar atualizações sobre o tempo de execução do EMR para Spark e Iceberg, bem como dicas sobre práticas recomendadas de configuração e recomendações de ajuste.

Sobre os Autores

Atul Félix Payapilly é engenheiro de desenvolvimento de software program do Amazon EMR na Amazon Net Providers.

Udit Mehrotra é gerente de engenharia de EMR na Amazon Net Providers.