As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá. # Melhorar a performance do Spark com Amazon S3 O Amazon EMR oferece atributos para ajudar na otimização da performance ao usar o Spark para consultar, ler e gravar dados salvos no Amazon S3. O [S3 Select](https://aws.amazon.com/blogs/aws/s3-glacier-select/) pode melhorar a performance de consultas para arquivos CSV e JSON em algumas aplicações ao enviar o processamento para o Amazon S3. O committer otimizado para EMRFS S3 é uma alternativa à [OutputCommitter](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/OutputCommitter.html)classe, que usa o recurso de uploads de várias partes do EMRFS para melhorar o desempenho ao gravar arquivos Parquet no Amazon S3 usando Spark e conjuntos de dados. DataFrames **Topics** + [Usar o S3 Select com Spark para melhorar a performance das consultas](emr-spark-s3select.md) + [EMR Spark MagicCommitProtocol](emr-spark-magic-commit-protocol.md) + [Usar o confirmador otimizado para EMRFS S3](emr-spark-s3-optimized-committer.md) + [Use o protocolo de confirmação otimizada para EMRFS S3](emr-spark-s3-optimized-commit-protocol.md) + [Tentar novamente solicitações do Amazon S3 com EMRFS](emr-spark-emrfs-retry.md) # Usar o S3 Select com Spark para melhorar a performance das consultas **Importante** O Amazon S3 Select não está mais disponível para novos clientes. Os clientes existentes do Amazon S3 Select podem continuar usando o recurso normalmente. [Saiba mais](https://aws.amazon.com/blogs/storage/how-to-optimize-querying-your-data-in-amazon-s3/) Com as versões 5.17.0 e posteriores do Amazon EMR, você pode usar o [S3 Select](https://aws.amazon.com/blogs/aws/s3-glacier-select/) com Spark no Amazon EMR. O *S3 Select* possibilita que as aplicações recuperem apenas um subconjunto dos dados de um objeto. Para o Amazon EMR, o trabalho computacional de filtrar grandes conjuntos de dados para processamento é enviado do cluster para o Amazon S3, o que pode melhorar a performance em algumas aplicações e reduz o volume de dados transferidos entre o Amazon EMR e o Amazon S3. O S3 Select é compatível com arquivos CSV e JSON usando os valores `s3selectCSV` e `s3selectJSON` para especificar o formato de dados. Para ter mais informações e exemplos, consulte [Especificar o S3 Select no seu código](#emr-spark-s3select-specify). ## O S3 Select é adequado para minha aplicação? Recomendamos que você avalie seus aplicativos com e sem o S3 Selecione para ver se o uso pode ser adequado para o aplicativo. Use as seguintes diretrizes para determinar se o seu aplicativo é adequado para o uso do S3 Select: + Sua consulta filtra mais de metade do conjunto de dados original. + Sua conexão de rede entre o Amazon S3 e o cluster do Amazon EMR tem boa velocidade de transferência e largura de banda disponível. O Amazon S3 não compacta respostas HTTP. Portanto, é provável que o tamanho da resposta aumente para arquivos de entrada compactados. ## Considerações e limitações + A criptografia do lado do servidor do Amazon S3 com chaves de criptografia fornecidas pelo cliente (SSE-C) e a criptografia do lado do cliente não são compatíveis. + A propriedade `AllowQuotedRecordDelimiters` não é compatível. Se essa propriedade for especificada, a consulta falhará. + Somente arquivos CSV e JSON no formato UTF-8 são compatíveis. Não CSVs há suporte para várias linhas. + Somente arquivos descompactados ou gzip são compatíveis. + As opções CSV e JSON do Spark, como `nanValue`, `positiveInf`, `negativeInf` e opções relacionadas a registros corrompidos (por exemplo, modo dropmalformed e failfast) não são compatíveis. + O uso de vírgulas (,) em casas decimais não é compatível. Por exemplo, `10,000` não é compatível, mas `10000` é. + Caracteres de comentário na última linha não são compatíveis. + Linhas vazias no final de um arquivo não são processadas. + Os seguintes filtros não são enviados para o Amazon S3: + Funções agregadas, como `COUNT()` e `SUM()`. + Filtros que `CAST()` um atributo. Por exemplo, .`CAST(stringColumn as INT) = 1` + Filtros com um atributo que é um objeto ou complexo. Por exemplo, .`intArray[1] = 1, objectColumn.objectNumber = 1` + Filtros para os quais o valor não é um valor literal. Por exemplo, `intColumn1 = intColumn2`. + Somente [tipos de dados compatíveis com o S3 Select](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-glacier-select-sql-reference-data-types.html) são compatíveis com as limitações documentadas. ## Especificar o S3 Select no seu código Os exemplos a seguir demonstram como especificar o S3 Select para CSV usando Scala, SQL, R e. PySpark Você pode usar o S3 Select para JSON da mesma forma. Para obter uma lista de opções, os valores padrão e limitações, consulte [Opções](#emr-spark-s3select-specify-options). ------ #### [ PySpark ] ``` spark .read .format("s3selectCSV") // "s3selectJson" for Json .schema(...) // optional, but recommended .options(...) // optional .load("s3://path/to/my/datafiles") ``` ------ #### [ R ] ``` read.df("s3://path/to/my/datafiles", "s3selectCSV", schema, header = "true", delimiter = "\t") ``` ------ #### [ Scala ] ``` spark .read .format("s3selectCSV") // "s3selectJson" for Json .schema(...) // optional, but recommended .options(...) // optional. Examples: // .options(Map("quote" -> "\'", "header" -> "true")) or // .option("quote", "\'").option("header", "true") .load("s3://path/to/my/datafiles") ``` ------ #### [ SQL ] ``` CREATE TEMPORARY VIEW MyView (number INT, name STRING) USING s3selectCSV OPTIONS (path "s3://path/to/my/datafiles", header "true", delimiter "\t") ``` ------ ### Opções As seguintes opções estão disponíveis ao usar `s3selectCSV` e `s3selectJSON`. Se não for especificado, os valores padrão serão usados. #### Opções com o S3selectCSV | Opção | Padrão | Usage | | --- | --- | --- | | `compression` | `"none"` | Indica se a compactação é usada. `"gzip"` é a única configuração compatível além de `"none"`. | | `delimiter` | "," | Especifica o delimitador de campo. | | `quote` | `'\"'` | Especifica o caractere de aspas. Não há suporte para a especificação de uma string vazia e isso resulta em um erro de XML malformado. | | `escape` | `'\\'` | Especifica o caractere de escape. | | `header` | `"false"` | `"false"` especifica que não há cabeçalho. `"true"` especifica que o cabeçalho está na primeira linha. Somente cabeçalhos na primeira linha são suportados e linhas vazias antes de um cabeçalho não são compatíveis. | | comment | `"#"` | Especifica o caractere de comentário. O indicador de comentários não pode ser desativado. Em outras palavras, um valor de `\u0000` não é suportado. | | `nullValue` | "" | | #### Opções com S3selectJSON | Opção | Padrão | Usage | | --- | --- | --- | | `compression` | `"none"` | Indica se a compactação é usada. `"gzip"` é a única configuração compatível além de `"none"`. | | `multiline` | "falso" | `"false"` especifica que o JSON está no formato `LINES` do S3 Select, o que significa que cada linha nos dados de entrada contém um único objeto JSON. `"true"` especifica que o JSON está no formato `DOCUMENT` do S3 Select, o que significa que um objeto JSON podem abranger várias linhas nos dados de entrada. | # EMR Spark MagicCommitProtocol Do EMR 6.15.0 em diante, MagicCommitProtocol torna-se o padrão FileCommitProtocol para o Spark ao utilizar o sistema de arquivos S3A. ## MagicCommitProtocol MagicCommitProtocol É uma implementação alternativa do [FileCommitProtocol](https://dlcdn.apache.org/spark/docs/2.4.2/api/java/org/apache/spark/internal/io/FileCommitProtocol.html)que é otimizada para gravar arquivos com o EMR Spark no Amazon S3 ao usar o sistema de arquivos S3A. Este protocolo visa melhorar a performance da aplicação, evitando o uso de operações de renomeação no Amazon S3 durante as fases de commit de trabalhos e tarefas. Essa MagicCommitProtocol é a FileCommitProtocol implementação padrão usada pelo Spark em execução no Amazon Elastic Map Reduce (EMR) quando o sistema de arquivos S3A é utilizado. O usa MagicCommitProtocol internamente o [MagicV2Committer](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/s3a-magicv2-committer.html) para realizar as gravações de arquivos no Amazon S3. Para operações de inserção estática, MagicCommitProtocol ele grava os arquivos no local de saída do trabalho durante a fase de confirmação da tarefa. Por outro lado, para operações de substituição de inserção dinâmica, os arquivos gravados por tentativas de tarefas aparecem apenas no local de saída do trabalho após a confirmação do trabalho. Isso é feito exportando os metadados de confirmação de volta para o driver do Spark na chamada de confirmação da tarefa. ## Habilitando MagicCommitProtocol O MagicCommitProtocol é habilitado por padrão para o Spark executado no Amazon Elastic Map Reduce (EMR) ao usar o sistema de arquivos S3A. Para usar o sistema de arquivos S3A, você pode: 1. Usar o esquema de arquivos como `s3a://` ao definir a tabela, a partição ou o diretório. 1. Defina a configuração `fs.s3.impl=org.apache.hadoop.fs.s3a.S3AFileSystem` em core-site.xml. ## Desativando o MagicCommitProtocol 1. Você pode definir `spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol` como “false” com codificação rígida em um `SparkConf`, transmitindo-o como um parâmetro `--conf` no shell Spark ou nas ferramentas `spark-submit` e `spark-sql` ou em `conf/spark-defaults.conf`. Para obter mais informações, consulte [Configuração do Spark](https://spark.apache.org/docs/latest/configuration.html) na documentação do Apache Spark. O exemplo a seguir mostra como desabilitar MagicCommitProtocol durante a execução de um `spark-sql` comando. ``` spark-sql \ --conf spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol=false \ -e "INSERT OVERWRITE TABLE target_table SELECT * FROM source_table;" ``` 1. Use a classificação de configuração `spark-defaults` para definir a propriedade `spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol` como false. Para obter mais informações, consulte [Configure applications](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html). ## MagicCommitProtocol considerações + Para inserção de partição estática, nos executores do Spark, o MagicCommitProtocol consome uma pequena quantidade de memória para cada arquivo gravado por uma tentativa de tarefa até que a tarefa seja confirmada ou abortada. Na maioria dos trabalhos, a quantidade de memória consumida é insignificante. Não há necessidade de memória extra no driver Spark + Para inserção dinâmica de partições, nos drivers do Spark, é MagicCommitProtocol necessária memória para armazenar informações de metadados de cada arquivo confirmado até que o trabalho seja confirmado ou abortado. Na maioria dos trabalhos, a configuração padrão de memória do driver do Spark é insignificante. Para trabalhos que têm tarefas de execução prolongada que gravam um grande número de arquivos, a memória que o confirmador consome pode ser perceptível e precisar de ajustes na memória alocada para executores do Spark. Você pode ajustar a memória usando a `spark.driver.memory` propriedade dos drivers do Spark e a `spark.executor.memory` propriedade dos executores do Spark. Como diretriz, uma única tarefa gravando 100.000 arquivos normalmente exigiria 200 MB adicionais de memória. Para obter mais informações, consulte [Propriedades da aplicação](https://spark.apache.org/docs/latest/configuration.html#application-properties) na documentação de configuração do Apache Spark. # Usar o confirmador otimizado para EMRFS S3 O committer otimizado para EMRFS S3 é uma [OutputCommitter](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/OutputCommitter.html)implementação alternativa que é otimizada para gravar arquivos no Amazon S3 ao usar o EMRFS. O confirmador otimizado para EMRFS S3 melhora a performance da aplicação ao evitar operações de listar e renomear no Amazon S3 durante o trabalho e fases de confirmação da tarefa. O confirmador está disponível com as versões 5.19.0 e posteriores do Amazon EMR e é habilitado por padrão nas versões 5.20.0 e posteriores do Amazon EMR. O committer é usado para trabalhos do Spark que usam Spark ou conjuntos de dados DataFrames. Desde o Amazon EMR 6.4.0, esse confirmador pode ser usado para todos os formatos comuns, incluindo parquet, ORC e formatos baseados em texto (incluindo CSV e JSON). Para versões anteriores ao Amazon EMR 6.4.0, somente o formato Parquet é compatível. Há circunstâncias em que o committer não é usado. Para obter mais informações, consulte [Requisitos do confirmador otimizado para EMRFS S3](emr-spark-committer-reqs.md). **Topics** + [Requisitos do confirmador otimizado para EMRFS S3](emr-spark-committer-reqs.md) + [O confirmador otimizado para EMRFS S3 e carregamentos multipart](emr-spark-committer-multipart.md) + [Considerações sobre ajuste de tarefas](emr-spark-committer-tuning.md) + [Ative o confirmador otimizado para EMRFS S3 para o Amazon EMR 5.19.0](emr-spark-committer-enable.md) # Requisitos do confirmador otimizado para EMRFS S3 O committer otimizado para EMRFS S3 é usado quando as seguintes condições são atendidas: + Você executa trabalhos do Spark que usam o Spark ou conjuntos de dados para gravar arquivos no Amazon S3. DataFrames Desde o Amazon EMR 6.4.0, esse confirmador pode ser usado para todos os formatos comuns, incluindo parquet, ORC e formatos baseados em texto (incluindo CSV e JSON). Para versões anteriores ao Amazon EMR 6.4.0, somente o formato Parquet é compatível. + Carregamentos multipart estão habilitados no Amazon EMR. Esse é o padrão. Para obter mais informações, consulte [O confirmador otimizado para EMRFS S3 e carregamentos multipart](emr-spark-committer-multipart.md). + O suporte a formato de arquivo integrado do Spark é usado. O suporte a formato de arquivo integrado é usado nas seguintes circunstâncias: + Para tabelas do Hive Metastore, quando `spark.sql.hive.convertMetastoreParquet` é definido como `true` para tabelas do Parquet ou `spark.sql.hive.convertMetastoreOrc` é definido como `true` para tabelas do Orc com as versões 6.4.0 ou superiores do Amazon EMR. Essas são as configurações padrão. + Quando os trabalhos são gravados em fontes de dados ou tabelas do Parquet, por exemplo, a tabela de destino é criada com a cláusula `USING parquet`. + Quando os trabalhos gravam em tabelas Parquet de metastore do Hive não particionadas. O suporte ao Parquet incorporado do Spark oferece suporte a tabelas particionadas do Hive, o que é uma limitação conhecida. Para obter mais informações, consulte [Conversão da tabela Parquet do Hive metastore](https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#hive-metastore-parquet-table-conversion) no Apache Spark e no Guia de conjuntos de dados. DataFrames + As operações de trabalhos do Spark que gravam em um local de partição padrão, como `${table_location}/k1=v1/k2=v2/`, usam o confirmador. O confirmador não será usado se uma operação de trabalho gravar em um local de partição personalizado, por exemplo, se o local de uma partição personalizado for definido usando o comando `ALTER TABLE SQL`. + Os valores a seguir para o Spark devem ser usados: + A propriedade `spark.sql.parquet.fs.optimized.committer.optimization-enabled` deve ser definida como `true`. Essa é a configuração padrão com as versões 5.20.0 e posteriores do Amazon EMR. Com o Amazon EMR 5.19.0, o valor padrão é `false`. Para obter informações sobre como configurar esse valor, consulte [Ative o confirmador otimizado para EMRFS S3 para o Amazon EMR 5.19.0](emr-spark-committer-enable.md). + Se você está gravando em tabelas da metastore do Hive não particionadas, somente os formatos de arquivo Parquet e Orc são compatíveis. `spark.sql.hive.convertMetastoreParquet` deve ser definido como `true` se você está gravando em tabelas da metastore do Hive não particionadas em Paquet. `spark.sql.hive.convertMetastoreOrc` deve ser definido como `true` se estiver gravando em tabelas da metastore do Hive não particionadas em Orc. Essas são as configurações padrão. + `spark.sql.parquet.output.committer.class` deve ser definido como `com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter`. Essa é a configuração padrão. + `spark.sql.sources.commitProtocolClass` deve ser definido como `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` ou `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol`. `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` é a configuração padrão para as versões 5.30.0 e superiores do Amazon EMR série 5.x e para as versões 6.2.0 e superiores do Amazon EMR série 6.x. `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol` é a configuração padrão para as versões anteriores do Amazon EMR. + Se os trabalhos do Spark substituírem os conjuntos de dados Parquet por colunas de partição dinâmica, as opções de gravação `partitionOverwriteMode` e `spark.sql.sources.partitionOverwriteMode` deverão ser definidas como `static`. Essa é a configuração padrão. **nota** A opção de gravação `partitionOverwriteMode` foi introduzida no Spark 2.4.0. Para a versão 2.3.2 do Spark, incluída com a versão 5.19.0 do Amazon EMR, defina a propriedade `spark.sql.sources.partitionOverwriteMode`. ## Ocasiões em que o confirmador otimizado para EMRFS S3 não é usado Geralmente, o confirmador otimizado para EMRFS S3 não é usado nas situações a seguir. **** | Situação | Por que o confirmador não é usado | | --- | --- | | Quando você grava no HDFS | O confirmador só permite gravação no Amazon S3 com o uso do EMRFS. | | Quando você usa o sistema de arquivos S3A | O confirmador só é compatível com o EMRFS. | | Quando você usa MapReduce nossa API RDD do Spark | O committer só oferece suporte ao uso de SparkSQL ou Dataset DataFrame. APIs | Os exemplos do Scala a seguir demonstram algumas situações adicionais que impedem que o confirmador otimizado para EMRFS S3 seja usado por completo (o primeiro exemplo) e em parte (o segundo exemplo). **Example – Modo de substituição de partição dinâmica** O exemplo do Scala a seguir instrui o Spark a usar um algoritmo de confirmação diferente, o que impede totalmente o uso do confirmador otimizado para EMRFS S3. O código define a propriedade `partitionOverwriteMode` como `dynamic` para subtituir somente as partições nas quais você está gravando dados. Em seguida, as colunas de partição dinâmica são especificadas por `partitionBy` e o modo de gravação é definido como `overwrite`. ``` val dataset = spark.range(0, 10) .withColumn("dt", expr("date_sub(current_date(), id)")) dataset.write.mode("overwrite") .option("partitionOverwriteMode", "dynamic") .partitionBy("dt") .parquet("s3://amzn-s3-demo-bucket1/output") ``` Você deve definir todas as três configurações para evitar o uso do confirmador otimizado para EMRFS S3. Quando você faz isso, o Spark executa um algoritmo de confirmação diferente, especificado no protocolo de confirmação do Spark. Para versões do Amazon EMR 5.x anteriores à 5.30.0 e para versões do Amazon EMR 6.x anteriores à 6.2.0, o protocolo de confirmação usa o diretório de teste do Spark, que é um diretório temporário criado no local de saída que começa com `.spark-staging`. O algoritmo renomeia sequencialmente diretórios de partição, o que pode afetar negativamente a performance. Para obter mais informações sobre as versões 5.30.0 e posteriores e 6.2.0 e posteriores do Amazon EMR, consulte. [Use o protocolo de confirmação otimizada para EMRFS S3](emr-spark-s3-optimized-commit-protocol.md) O algoritmo no Spark 2.4.0 segue estas etapas: 1. As tentativas de tarefa gravam a saída nos diretórios de partição do diretório de teste do Spark, por exemplo, `${outputLocation}/spark-staging-${jobID}/k1=v1/k2=v2/`. 1. Para cada partição gravada, a tentativa de tarefa acompanha os caminhos de partição relativos, por exemplo, `k1=v1/k2=v2`. 1. Quando uma tarefa é concluída com êxito, ela fornece o driver com todos os caminhos de partição relativos que ela controlou. 1. Depois que todas as tarefas forem concluídas, a fase de confirmação do trabalho coletará todos os diretórios da partição que as tentativas de tarefas bem-sucedidas gravaram no diretório de preparação do Spark. O Spark renomeia sequencialmente cada um desses diretórios para o local de saída final usando as operações para renomear a árvore de diretórios. 1. O diretório de preparação é excluído antes de a fase de confirmação de trabalho ser concluída. **Example – Local de partição personalizado** Neste exemplo, o código Scala insere em duas partições. Uma partição tem um local de partição personalizado. A outra partição usa o local de partição padrão. O committer otimizado para EMRFS S3 é usado somente para gravar a saída da tarefa na partição que usa o local de partição padrão. ``` val table = "dataset" val location = "s3://bucket/table" spark.sql(s""" CREATE TABLE $table (id bigint, dt date) USING PARQUET PARTITIONED BY (dt) LOCATION '$location' """) // Add a partition using a custom location val customPartitionLocation = "s3://bucket/custom" spark.sql(s""" ALTER TABLE $table ADD PARTITION (dt='2019-01-28') LOCATION '$customPartitionLocation' """) // Add another partition using default location spark.sql(s"ALTER TABLE $table ADD PARTITION (dt='2019-01-29')") def asDate(text: String) = lit(text).cast("date") spark.range(0, 10) .withColumn("dt", when($"id" > 4, asDate("2019-01-28")).otherwise(asDate("2019-01-29"))) .write.insertInto(table) ``` O código Scala cria os seguintes objetos do Amazon S3: ``` custom/part-00001-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet custom_$folder$ table/_SUCCESS table/dt=2019-01-29/part-00000-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet table/dt=2019-01-29_$folder$ table_$folder$ ``` Ao gravar em partições em locais personalizados, o Spark usa um algoritmo de confirmação semelhante ao exemplo anterior, que é descrito abaixo. Como no exemplo anterior, o algoritmo resulta em renomeações sequenciais, o que pode afetar negativamente a performance. 1. Ao gravar a saída em uma partição em um local personalizado, as tarefas gravam em um arquivo no diretório de preparação do Spark, que é criado no local de saída final. O nome do arquivo inclui um UUID aleatório para evitar colisões de nomes de arquivos. A tentativa de tarefa controla cada arquivo junto com o caminho de saída final desejado. 1. Quando uma tarefa é concluída com êxito, ela fornece o driver com os arquivos e os caminhos desejados de saída final. 1. Depois que todas as tarefas forem concluídas, a fase de confirmação do trabalho renomeará sequencialmente todos os arquivos que foram gravados para partições em locais personalizados em seus caminhos de saída final. 1. O diretório de preparação é excluído antes de a fase de confirmação de trabalho ser concluída. # O confirmador otimizado para EMRFS S3 e carregamentos multipart Para usar o confirmador otimizado para EMRFS S3, uploads de várias partes devem estar habilitados no Amazon EMR. Multipart uploads são habilitados por padrão. Você pode habilitá-los novamente, se necessário. Para obter mais informações, consulte [Configure multipart upload for Amazon S3](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#Config_Multipart) (Configurar o carregamento fracionado no Amazon S3) no *Guia de gerenciamento do Amazon EMR*. O EMRFS S3 Optimized Committer usa as características semelhantes a transações de multipart uploads para garantir que os arquivos gravados por tentativas de tarefas aparecem apenas no local de saída do trabalho após a confirmação da tarefa. Ao usar uploads de várias partes dessa forma, o committer melhora o desempenho da confirmação da tarefa em relação à versão 2 do FileOutputCommitter algoritmo padrão. Ao usar o EMRFS S3 Optimized Committer, há algumas diferenças fundamentais de comportamento em relação ao comportamento tradicional de multipart uploads a considerar: + Os multipart uploads são sempre executados, independentemente do tamanho do arquivo. Isso é diferente do comportamento padrão do EMRFS, em que a propriedade `fs.s3n.multipart.uploads.split.size` controla o tamanho do arquivo no qual multipart uploads são acionados. + Os multipart uploads são deixados incompletos por um período mais longo até que a tarefa seja confirmada ou cancelada. Isso é diferente do comportamento padrão do EMRFS no qual um multipart upload é concluído quando uma tarefa é concluída ao gravar um determinado arquivo. Devido a essas diferenças, se uma JVM do executor do Spark apresenta falha ou é eliminada enquanto as tarefas estão executando e gravando dados no Amazon S3, é mais provável que os carregamentos multipart partes sejam abandonados. Por esse motivo, quando você usa o EMRFS S3 Optimized Committer, certifique-se de seguir as melhores práticas para gerenciar multipart uploads com falha. Para obter mais informações, consulte [Práticas recomendadas](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#emr-bucket-bestpractices) para trabalhar com buckets do Amazon S3 no *Guia de gerenciamento do Amazon EMR*. # Considerações sobre ajuste de tarefas O EMRFS S3 Optimized Committer consome uma pequena quantidade de memória para cada arquivo gravado por uma tentativa de tarefa até que a tarefa seja confirmada ou cancelada. Na maioria dos trabalhos, a quantidade de memória consumida é insignificante. Para trabalhos que têm tarefas de longa execução que gravam um grande número de arquivos, a memória que o committer consome pode ser perceptível e exigem ajustes na memória alocada para executores do Spark. É possível ajustar a memória do executor usando a propriedade `spark.executor.memory`. Como diretriz, uma única tarefa gravando 100.000 arquivos normalmente exigiria 100 MB adicionais de memória. Para obter mais informações, consulte [Propriedades da aplicação](https://spark.apache.org/docs/latest/configuration.html#application-properties) na documentação de configuração do Apache Spark. # Ative o confirmador otimizado para EMRFS S3 para o Amazon EMR 5.19.0 Se você estiver usando o Amazon EMR 5.19.0, poderá definir manualmente a propriedade `spark.sql.parquet.fs.optimized.committer.optimization-enabled` como `true` quando criar um cluster ou no Spark, se estiver usando o Amazon EMR. ## Habilitar o confirmador otimizado para EMRFS S3 quando criar um cluster Use a classificação de configuração `spark-defaults` para definir a propriedade `spark.sql.parquet.fs.optimized.committer.optimization-enabled` como `true`. Para obter mais informações, consulte [Configurar aplicações](emr-configure-apps.md). ## Habilitar o confirmador otimizado para EMRFS S3 no Spark Você pode definir `spark.sql.parquet.fs.optimized.committer.optimization-enabled` como `true` codificando-o em um `SparkConf`, transmitindo-o como um parâmetro `--conf` no shell Spark ou nas ferramentas `spark-submit` e `spark-sql` ou em `conf/spark-defaults.conf`. Para obter mais informações, consulte [Configuração do Spark](https://spark.apache.org/docs/latest/configuration.html) na documentação do Apache Spark. O exemplo a seguir mostra como habilitar o committer ao executar um comando spark-sql. ``` spark-sql \ --conf spark.sql.parquet.fs.optimized.committer.optimization-enabled=true \ -e "INSERT OVERWRITE TABLE target_table SELECT * FROM source_table;" ``` # Use o protocolo de confirmação otimizada para EMRFS S3 O protocolo de confirmação otimizado para S3 do EMRFS é uma [FileCommitProtocol](https://spark.apache.org/docs/2.2.0//api/java/org/apache/spark/internal/io/FileCommitProtocol.html)implementação alternativa otimizada para gravar arquivos com a substituição de partição dinâmica do Spark no Amazon S3 ao usar o EMRFS. O protocolo melhora a performance da aplicação ao evitar operações de renomeação no Amazon S3 durante a fase de confirmação do trabalho de substituição de partição dinâmica do Spark. Observe que [EMRFS S3-optimized committer](emr-spark-s3-optimized-committer.html) também melhora o a performance ao evitar operações de renomeação. No entanto, ele não funciona para casos de substituição de partição dinâmica, embora as melhorias no protocolo de confirmação sejam direcionados apenas a casos de substituição de partição dinâmica. O confirmador está disponível com as versões 5.30.0 e posteriores e 6.2.0 e posteriores do Amazon EMR e é habilitado por padrão. O Amazon EMR adicionou uma melhoria no paralelismo a partir da versão 5.31.0. O protocolo é usado para trabalhos do Spark que usam Spark ou conjuntos de DataFrames dados. Há circunstâncias em que o protocolo de confirmação não é usado. Para obter mais informações, consulte [Requisitos do protocolo de confirmação otimizado para EMRFS S3](emr-spark-committer-reqs.md). **Topics** + [Requisitos do protocolo de confirmação otimizado para EMRFS S3](emr-spark-commit-protocol-reqs.md) + [O protocolo de confirmação otimizado para EMRFS S3 e carregamentos multipart](emr-spark-commit-protocol-multipart.md) + [Considerações sobre ajuste de tarefas](emr-spark-commit-protocol-tuning.md) # Requisitos do protocolo de confirmação otimizado para EMRFS S3 O protocolo de confirmação otimizado para EMRFS S3 é usado quando as seguintes condições são atendidas: + Você executa trabalhos do Spark que usam o Spark ou conjuntos de dados para substituir tabelas particionadas. DataFrames + Você executa trabalhos do Spark cujo modo de substituição de partição é `dynamic`. + Carregamentos multipart estão habilitados no Amazon EMR. Esse é o padrão. Para obter mais informações, consulte [O protocolo de confirmação otimizado para EMRFS S3 e carregamentos multipart](emr-spark-commit-protocol-multipart.md). + O cache do sistema de arquivos para o EMRFS está habilitado. Esse é o padrão. Verifique se a configuração `fs.s3.impl.disable.cache` está definida como `false`. + O suporte integrado de fonte de dados do Spark é usado. O suporte integrado ao Parquet é usado nas seguintes circunstâncias: + Quando os trabalhos gravam em fontes de dados ou tabelas integradas. + Quando os trabalhos gravam em tabelas do Parquet do metastore do Hive. Isso acontece quando `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastoreParquet` são definidos como verdadeiros. Essas são as configurações padrão. + Quando os trabalhos gravam na tabela do ORC do metastore do Hive. Isso acontece quando `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastoreOrc` são definidos como `true`. Essas são as configurações padrão. + As operações de trabalhos do Spark que gravam em um local de partição padrão, por exemplo, `${table_location}/k1=v1/k2=v2/`, usam o confirmador. O protocolo não será usado se uma operação de trabalho gravar em um local de partição personalizado, por exemplo, se o local de uma partição personalizado for definido usando o comando `ALTER TABLE SQL`. + Os valores a seguir para o Spark devem ser usados: + `spark.sql.sources.commitProtocolClass` deve ser definido como `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol`. Essa é a configuração padrão para as versões 5.30.0 e superiores e 6.2.0 e superiores do Amazon EMR. + A opção de gravação `partitionOverwriteMode` ou `spark.sql.sources.partitionOverwriteMode` deve ser definida como `dynamic`. A configuração padrão é `static`. **nota** A opção de gravação `partitionOverwriteMode` foi introduzida no Spark 2.4.0. Para a versão 2.3.2 do Spark, incluída com a versão 5.19.0 do Amazon EMR, defina a propriedade `spark.sql.sources.partitionOverwriteMode`. + Se os trabalhos do Spark substituírem a tabela do Parquet do metastore do Hive, `spark.sql.hive.convertMetastoreParquet`, `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastore.partitionOverwriteMode` deverão ser configurados como `true`. Existem as configurações padrão. + Se os trabalhos do Spark substituírem a tabela do ORC do metastore do Hive, `spark.sql.hive.convertMetastoreOrc`, `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastore.partitionOverwriteMode` deverão ser configurados como `true`. Existem as configurações padrão. **Example – Modo de substituição de partição dinâmica** Neste exemplo do Scala, a otimização é acionada. Primeiro, você define a propriedade `partitionOverwriteMode` como `dynamic`. Isso só substitui as partições nas quais você está gravando dados. Em seguida, você especifica as colunas de partição dinâmica com `partitionBy` e define o modo de gravação como `overwrite`. ``` val dataset = spark.range(0, 10) .withColumn("dt", expr("date_sub(current_date(), id)")) dataset.write.mode("overwrite") // "overwrite" instead of "insert" .option("partitionOverwriteMode", "dynamic") // "dynamic" instead of "static" .partitionBy("dt") // partitioned data instead of unpartitioned data .parquet("s3://amzn-s3-demo-bucket1/output") // "s3://" to use Amazon EMR file system, instead of "s3a://" or "hdfs://" ``` ## Quando o protocolo de confirmação otimizado para EMRFS S3 não é usado Geralmente, o protocolo de confirmação otimizado para EMRFS S3 funciona da mesma forma que o protocolo de confirmação padrão do Spark de código aberto, `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol`. A otimização não ocorrerá nas situações a seguir. **** | Situação | Por que o protocolo de confirmação não é usado | | --- | --- | | Quando você grava no HDFS | O protocolo de confirmação só permite gravação no Amazon S3 com o uso do EMRFS. | | Quando você usa o sistema de arquivos S3A | O protocolo de confirmação só é compatível com EMRFS. | | Quando você usa MapReduce nossa API RDD do Spark | O protocolo de confirmação só oferece suporte ao uso de SparkSQL ou Dataset DataFrame. APIs | | Quando a substituição da partição dinâmica não é acionada | O protocolo de confirmação só otimiza os casos de substituição de partição dinâmica. Para outros casos, consulte [Usar o confirmador otimizado para EMRFS S3](emr-spark-s3-optimized-committer.md). | Os exemplos de Scala a seguir demonstram algumas situações adicionais que o protocolo de confirmação otimizado para EMRFS S3 delega para `SQLHadoopMapReduceCommitProtocol`. **Example – Modo de substituição de partição dinâmica com local de partição personalizado** Neste exemplo, os programas Scala sobrescrevem duas partições no modo de substituição dinâmica de partição. Uma partição tem um local de partição personalizado. A outra partição usa o local de partição padrão. O protocolo de confirmação otimizado para EMRFS S3 só melhora a partição que usa o local de partição padrão. ``` val table = "dataset" val inputView = "tempView" val location = "s3://bucket/table" spark.sql(s""" CREATE TABLE $table (id bigint, dt date) USING PARQUET PARTITIONED BY (dt) LOCATION '$location' """) // Add a partition using a custom location val customPartitionLocation = "s3://bucket/custom" spark.sql(s""" ALTER TABLE $table ADD PARTITION (dt='2019-01-28') LOCATION '$customPartitionLocation' """) // Add another partition using default location spark.sql(s"ALTER TABLE $table ADD PARTITION (dt='2019-01-29')") def asDate(text: String) = lit(text).cast("date") spark.range(0, 10) .withColumn("dt", when($"id" > 4, asDate("2019-01-28")).otherwise(asDate("2019-01-29"))) .createTempView(inputView) // Set partition overwrite mode to 'dynamic' spark.sql(s"SET spark.sql.sources.partitionOverwriteMode=dynamic") spark.sql(s"INSERT OVERWRITE TABLE $table SELECT * FROM $inputView") ``` O código Scala cria os seguintes objetos do Amazon S3: ``` custom/part-00001-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet custom_$folder$ table/_SUCCESS table/dt=2019-01-29/part-00000-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet table/dt=2019-01-29_$folder$ table_$folder$ ``` Gravar em locais de partição personalizados em versões anteriores do Spark pode resultar em perda de dados. Neste exemplo, a partição `dt='2019-01-28'` seria perdida. Para obter mais detalhes, consulte [SPARK-35106](https://issues.apache.org/jira/browse/SPARK-35106). Isso foi corrigido nas versões 5.33.0 e posteriores do Amazon EMR, excluindo 6.0.x e 6.1.x. Ao gravar em partições em locais personalizados, o Spark usa um algoritmo de confirmação semelhante ao exemplo anterior, que é descrito abaixo. Como no exemplo anterior, o algoritmo resulta em renomeações sequenciais, o que pode afetar negativamente a performance. O algoritmo no Spark 2.4.0 segue estas etapas: 1. Ao gravar a saída em uma partição em um local personalizado, as tarefas gravam em um arquivo no diretório de preparação do Spark, que é criado no local de saída final. O nome do arquivo inclui um UUID aleatório para evitar colisões de nomes de arquivos. A tentativa de tarefa controla cada arquivo junto com o caminho de saída final desejado. 1. Quando uma tarefa é concluída com êxito, ela fornece o driver com os arquivos e os caminhos desejados de saída final. 1. Depois que todas as tarefas forem concluídas, a fase de confirmação do trabalho renomeará sequencialmente todos os arquivos que foram gravados para partições em locais personalizados em seus caminhos de saída final. 1. O diretório de preparação é excluído antes de a fase de confirmação de trabalho ser concluída. # O protocolo de confirmação otimizado para EMRFS S3 e carregamentos multipart Para usar a otimização para substituição de partições dinâmicas no protocolo de confirmação otimizado para EMRFS S3, os carregamentos multipart devem ser habilitados no Amazon EMR. Multipart uploads são habilitados por padrão. Você pode habilitá-los novamente, se necessário. Para obter mais informações, consulte [Configure multipart upload for Amazon S3](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#Config_Multipart) (Configurar o carregamento fracionado no Amazon S3) no *Guia de gerenciamento do Amazon EMR*. Durante a substituição de partições dinâmicas, o protocolo de confirmação otimizado para EMRFS S3 usa as características semelhantes a transações de carregamentos multipart para garantir que os arquivos gravados por tentativas de tarefas apareçam apenas no local de saída do trabalho após a confirmação do trabalho. Ao usar carregamentos multipart dessa maneira, o protocolo de confirmação melhora a performance de confirmação de trabalhos em relação ao padrão `SQLHadoopMapReduceCommitProtocol`. Quando o protocolo de confirmação otimizado para EMRFS S3 é usado, há algumas diferenças fundamentais com relação ao comportamento tradicional de carregamentos multipart a serem consideradas: + Os multipart uploads são sempre executados, independentemente do tamanho do arquivo. Isso é diferente do comportamento padrão do EMRFS, em que a propriedade `fs.s3n.multipart.uploads.split.size` controla o tamanho do arquivo no qual multipart uploads são acionados. + Os multipart uploads são deixados incompletos por um período mais longo até que a tarefa seja confirmada ou cancelada. Isso é diferente do comportamento padrão do EMRFS no qual um multipart upload é concluído quando uma tarefa é concluída ao gravar um determinado arquivo. Devido a essas diferenças, se uma JVM do executor do Spark apresenta falha ou é eliminada enquanto as tarefas estão executando e gravando dados no Amazon S3 ou se uma JVM do executor do Spark apresenta falha ou é eliminada enquanto um trabalho está sendo executado, é mais provável que os carregamentos multipart sejam abandonados. Por esse motivo, quando você usa o protocolo de confirmação otimizado para EMRFS S3, não deixe de seguir as práticas recomendadas para gerenciar carregamentos multipart que apresentaram falha. Para obter mais informações, consulte [Práticas recomendadas](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#emr-bucket-bestpractices) para trabalhar com buckets do Amazon S3 no *Guia de gerenciamento do Amazon EMR*. # Considerações sobre ajuste de tarefas Em executores do Spark, o protocolo de confirmação otimizado para EMRFS S3 consome uma pequena quantidade de memória para cada arquivo gravado por uma tentativa de tarefa até que a tarefa seja confirmada ou cancelada. Na maioria dos trabalhos, a quantidade de memória consumida é insignificante. Nos drivers do Spark, o protocolo de confirmação otimizado para EMRFS S3 exige memória para armazenar informações de metadados de cada arquivo confirmado até que o trabalho seja confirmado ou cancelado. Na maioria dos trabalhos, a configuração padrão de memória do driver do Spark é insignificante. Para trabalhos que têm tarefas de execução prolongada que gravam um grande número de arquivos, a memória que o confirmador consome pode ser perceptível e precisar de ajustes na memória alocada para executores do Spark. Você pode ajustar a memória usando a `spark.driver.memory` propriedade dos drivers do Spark e a `spark.executor.memory` propriedade dos executores do Spark. Como diretriz, uma única tarefa gravando 100.000 arquivos normalmente exigiria 100 MB adicionais de memória. Para obter mais informações, consulte [Propriedades da aplicação](https://spark.apache.org/docs/latest/configuration.html#application-properties) na documentação de configuração do Apache Spark. # Tentar novamente solicitações do Amazon S3 com EMRFS Este tópico fornece informações sobre as estratégias de novas tentativas que você pode usar ao fazer solicitações ao Amazon S3 com o EMRFS. Quando a taxa de solicitação aumenta, o S3 tenta escalar para suportar a nova taxa. Durante esse processo, o S3 pode controlar a utilização das solicitações e retornar um erro `503 Slow Down`. Para melhorar a taxa de sucesso das solicitações do S3, você pode ajustar sua estratégia de novas tentativas configurando propriedades na sua configuração `emrfs-site`. Você pode ajustar sua estratégia de novas tentativas das maneiras a seguir. + Aumente o limite máximo de novas tentativas para a estratégia padrão de novas tentativas de recuo exponencial. + Habilite e configure a estratégia de novas tentativas de additive-increase/multiplicative-decrease (AIMD - aumento aditivo/diminuição multiplicativa). O AIMD é compatível com versões 6.4.0 e posteriores do Amazon EMR. ## Usar a estratégia padrão de recuo exponencial Por padrão, o EMRFS usa uma estratégia de recuo exponencial para tentar novamente solicitações do Amazon S3. O limite padrão de novas tentativas do EMRFS é 15. Para evitar um erro `503 Slow Down` do S3, você pode aumentar o limite de novas tentativas ao criar um novo cluster, em um cluster em execução ou no runtime da aplicação. Para aumentar o limite de novas tentativas, você deve alterar o valor de `fs.s3.maxRetries` na sua configuração `emrfs-site`. O exemplo de configuração a seguir define `fs.s3.maxRetries` como um valor personalizado de 30. ``` [ { "Classification": "emrfs-site", "Properties": { "fs.s3.maxRetries": "30" } } ] ``` Para obter mais informações sobre como trabalhar com objetos de configuração, consulte [Configurar aplicações](emr-configure-apps.md). ## Usar a estratégia de novas tentativas do AIMD Com as versões 6.4.0 e posteriores do Amazon EMR, o EMRFS é compatível com uma estratégia alternativa de novas tentativas baseada em um modelo de aumento aditivo/diminuição multiplicativa (AIMD). A estratégia de novas tentativa do AIMD é especialmente útil quando você trabalha com grandes clusters do Amazon EMR. O AIMD calcula uma taxa de solicitação personalizada usando dados sobre solicitações recentes bem-sucedidas. Essa estratégia diminui o número de solicitações submetidas a controle de utilização e o total de tentativas necessárias por solicitação. Para habilitar a estratégia de novas tentativas do AIMD, você deve definir a propriedade `fs.s3.aimd.enabled` como `true` em sua configuração `emrfs-site`, como no exemplo a seguir. ``` [ { "Classification": "emrfs-site", "Properties": { "fs.s3.aimd.enabled": "true" } } ] ``` Para obter mais informações sobre como trabalhar com objetos de configuração, consulte [Configurar aplicações](emr-configure-apps.md). ## Configurações avançadas de novas tentativas do AIMD Você pode configurar as propriedades listadas na tabela a seguir para refinar o comportamento de novas tentativas quando usar a estratégia de novas tentativas do AIMD. Para a maioria dos casos de uso, recomendamos que você use os valores padrão. **Propriedades avançadas da estratégia de novas tentativas do AIMD** | Propriedade | Valor padrão | Description | | --- | --- | --- | | fs.s3.aimd.increaseIncrement | 0.1 | Controla a rapidez com que a taxa de solicitações aumenta quando solicitações consecutivas são bem-sucedidas. | | fs.s3.aimd.reductionFactor | 2 | Controla a rapidez com que a taxa de solicitação diminui quando o Amazon S3 retorna uma resposta 503. O fator padrão de 2 reduz a taxa de solicitação pela metade. | | fs.s3.aimd.minRate | 0.1 | Define o limite inferior da taxa de solicitações quando as solicitações sofrem controle de utilização sustentado pelo S3. | | fs.s3.aimd.initialRate | 5500 | Define a taxa de solicitação inicial, que sofre alterações de acordo com os valores que você especifica para fs.s3.aimd.increaseIncrement e fs.s3.aimd.reductionFactor.A taxa inicial também é usada para solicitações GET e é escalada proporcionalmente (3500/5500) para solicitações PUT. | | fs.s3.aimd.adjustWindow | 2 | Controla a frequência com que a taxa de solicitação é ajustada, medida em número de respostas. | | fs.s3.aimd.maxAttempts | 100 | Define o número máximo de tentativas para testar uma solicitação. |