Create and use an EMR Spark node - DataWorks: Central de Documentação do Alibaba Cloud

O Spark é um mecanismo de análise unificada para big data, reconhecido pelo alto desempenho, facilidade de uso e versatilidade. Ele oferece suporte a computação complexa em memória, sendo ideal para criar aplicações de análise de dados em grande escala e com baixa latência. O DataWorks permite usar nós EMR Spark para desenvolver e agendar periodicamente tarefas do Spark. Este tópico explica como configurar e usar nós EMR Spark e fornece exemplos de aplicação.

Pré-requisitos

Antes de iniciar o desenvolvimento do nó, se você precisar de um ambiente de componentes personalizado, crie uma imagem personalizada com base na imagem oficial dataworks_emr_base_task_pod e use-a no Data Studio. Para mais informações, consulte Criar uma imagem personalizada e Usar imagens no desenvolvimento de dados.

Por exemplo, ao criar uma imagem personalizada, substitua pacotes JAR do Spark ou adicione dependências de libraries, files ou JAR packages específicos.
Crie um cluster Alibaba Cloud EMR e registre-o no DataWorks. Para mais informações, consulte Novo Data Studio: Anexar um recurso de computação EMR.
(Opcional, para usuários RAM) Adicione o usuário do Resource Access Management (RAM) responsável pelo desenvolvimento da tarefa ao workspace e atribua a ele a função Development ou Workspace Administrator (esta função inclui permissões amplas e deve ser concedida com cautela). Para mais informações, consulte Adicionar membros ao workspace.

Se você estiver usando uma conta raiz, pule esta etapa.
Se suas tarefas exigirem um ambiente de desenvolvimento específico, crie uma imagem personalizada. Para mais informações, consulte imagem personalizada.

Limitações

Os nós EMR Shell só podem ser executados em um grupo de recursos serverless (recomendado) ou em um grupo de recursos exclusivo para agendamento. O uso de imagem personalizada no desenvolvimento de dados requer um grupo de recursos serverless.
Para gerenciar metadados de um cluster DataLake ou personalizado no DataWorks, configure primeiro o EMR-HOOK no cluster. Para mais informações, consulte Configurar EMR-HOOK para Spark SQL.

Nota
Sem o EMR-HOOK configurado no cluster, o DataWorks não consegue exibir metadados em tempo real, gerar logs de auditoria, mostrar linhagem de dados ou executar tarefas de governança relacionadas ao EMR.
Os clusters Spark do EMR on ACK não oferecem suporte à visualização de linhagem de dados, mas os clusters EMR Serverless Spark oferecem esse recurso.
Os clusters Spark do EMR on ACK e do EMR Serverless Spark suportam apenas a referência a recursos OSS via OSS REF e o upload de recursos para o OSS, mas não permitem o upload de recursos para o HDFS.
Clusters DataLake e clusters personalizados permitem referenciar recursos OSS via OSS REF, fazer upload de recursos para o OSS e também para o HDFS.

Observações

Se você ativou o controle de acesso Ranger para Spark no cluster EMR associado ao workspace atual:

Tarefas Spark que usam a imagem padrão suportam automaticamente este recurso.
Para executar uma tarefa Spark com uma imagem personalizada, envie um ticket ao suporte técnico solicitando a atualização da imagem.

Desenvolver um job Spark e obter o pacote JAR

Antes de agendar jobs EMR Spark com o DataWorks, desenvolva o job no EMR e compile o código em um pacote JAR. Para mais informações sobre o desenvolvimento de jobs EMR Spark, consulte Visão geral do Spark.

Nota

Em seguida, faça o upload do pacote JAR para o DataWorks para agendar jobs EMR Spark recorrentes.

Procedimento

Na página de edição do nó EMR Spark, desenvolva o job.

Desenvolver um job Spark

Escolha uma opção conforme o seu cenário:

Opção 1: Upload e referência

O DataWorks permite fazer upload de recursos da sua máquina local para o DataStudio e referenciá-los posteriormente. Após compilar o job EMR Spark, obtenha o pacote jar compilado. O método de armazenamento recomendado para o pacote jar depende do seu tamanho.

Faça o upload do pacote jar e crie-o como um recurso EMR do DataWorks; em seguida, envie-o. Também é possível armazená-lo diretamente no HDFS do cluster EMR. Clusters EMR on ACK e EMR Serverless Spark não suportam upload de recursos para o HDFS.

JAR < 500 MB

Crie um recurso JAR do EMR.

Se o pacote jar tiver menos de 500 MB, faça o upload dele da sua máquina local como um recurso JAR EMR do DataWorks. Isso facilita o gerenciamento visual no console do DataWorks. Após criar o recurso, envie-o. Para mais informações, consulte Criar e usar recursos EMR.
1. Faça o upload do pacote jar para o diretório de armazenamento de recursos JAR usando o método de upload Local. Para mais informações, consulte Gerenciamento de recursos.
2. Clique em Click Upload para enviar o recurso JAR.
3. Selecione um Storage Path, Data Sources e Resource Group.
4. Clique em Save.
Referencie o recurso JAR do EMR.
1. Abra o nó EMR Spark criado e acesse a página de edição de código.
2. Na seção Resource Management do painel de navegação à esquerda, localize o recurso desejado, clique com o botão direito nele e selecione Insert Resource Path.
3. Após selecionar a referência, uma instrução de referência é adicionada automaticamente ao editor de código do nó EMR Spark. Isso indica que a referência foi bem-sucedida.
```
##@resource_reference{"spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jar"}
spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jar
```
  Neste código, spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jar corresponde ao nome do recurso JAR EMR enviado por você.
4. Edite o código do nó EMR Spark para incluir o comando spark-submit. O código abaixo serve como exemplo.
  
  Nota
  O editor do nó EMR Spark não suporta comentários. Escreva o código do job conforme o exemplo a seguir, sem comentários. Caso contrário, a execução do nó falhará.
```
##@resource_reference{"spark-examples_2.11-2.4.0.jar"}
spark-submit --class org.apache.spark.examples.SparkPi --master yarn  spark-examples_2.11-2.4.0.jar 100
```
  Nota
  - org.apache.spark.examples.SparkPi: A classe principal no pacote jar compilado.
  - spark-examples_2.11-2.4.0.jar: O nome do recurso JAR EMR enviado.
  - Use os valores de exemplo para os demais parâmetros. Alternativamente, execute o comando spark-submit --help para visualizar informações de uso e modifique o comando spark-submit conforme necessário.
  - Para usar parâmetros simplificados no comando spark-submit dentro de um nó Spark, adicione-os ao seu código, por exemplo: --executor-memory 2G.
  - Nós Spark suportam apenas o envio de jobs para o YARN no modo cluster.
  - Para jobs enviados com spark-submit, defina o modo de implantação como cluster, e não como client.

JAR >= 500 MB

Crie um recurso JAR do EMR.

Se o pacote jar tiver 500 MB ou mais, não será possível fazer o upload dele da sua máquina local como um recurso do DataWorks. Recomendamos armazenar o pacote jar diretamente no HDFS do cluster EMR e registrar seu caminho de armazenamento. Depois, referencie esse caminho ao agendar o job Spark no DataWorks.
1. Faça o upload do pacote jar para o diretório de armazenamento de recursos JAR usando o método de upload Local. Para mais informações, consulte Gerenciamento de recursos.
2. Clique em Click Upload para enviar o recurso JAR.
3. Selecione um Storage Path, Data Sources e Resource Group.
4. Clique em Save.
Referencie o recurso JAR do EMR.

Se o pacote jar estiver armazenado no HDFS, referencie-o especificando seu caminho no código do nó EMR Spark.
1. Clique duas vezes no nó EMR Spark criado para abrir sua página de edição de código.
2. Escreva o comando spark-submit. Veja um exemplo abaixo.
```
spark-submit --master yarn
--deploy-mode cluster
--name SparkPi
--driver-memory 4G
--driver-cores 1
--num-executors 5
--executor-memory 4G
--executor-cores 1
--class org.apache.spark.examples.JavaSparkPi
hdfs:///tmp/jars/spark-examples_2.11-2.4.8.jar 100
```
  Nota
  - hdfs:///tmp/jars/spark-examples_2.11-2.4.8.jar: O caminho real do pacote jar no HDFS.
  - org.apache.spark.examples.JavaSparkPi: A classe principal no pacote jar compilado.
  - Configure os outros parâmetros de acordo com as definições do seu cluster EMR. Você também pode executar o comando spark-submit --help para ver as informações de uso do spark-submit e modificar o comando conforme necessário.
  - Para usar parâmetros simplificados no comando spark-submit em um nó Spark, adicione-os ao código, por exemplo: --executor-memory 2G.
  - Nós Spark suportam apenas o envio de jobs para o YARN no modo cluster.
  - Para jobs enviados com spark-submit, defina o modo de implantação como cluster, e não como client.

Opção 2: Referenciar um recurso OSS

Referencie recursos OSS diretamente a partir do nó atual usando OSS REF. Ao executar um nó EMR, o DataWorks carrega automaticamente os recursos OSS especificados no código para uso local. Esse método é comum em cenários onde jobs EMR exigem dependências JAR ou dependem de scripts.

Desenvolva o recurso JAR.

Prepare as dependências de código.

Visualize as dependências de código necessárias no caminho /usr/lib/emr/spark-current/jars/ no nó mestre do seu cluster EMR. O exemplo a seguir usa o Spark 3.4.2. Abra um projeto existente no IntelliJ IDEA, adicione as dependências do Project Object Model (POM) e referencie os plug-ins relevantes.

Dependências POM

<dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.4.2</version>
        </dependency>
        <!-- Apache Spark SQL -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.4.2</version>
        </dependency>
</dependencies>

Plug-ins

<build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.7.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <configuration>
                    <recompileMode>incremental</recompileMode>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

Código de exemplo

package com.aliyun.emr.example.spark
import org.apache.spark.sql.SparkSession
object SparkMaxComputeDemo {
  def main(args: Array[String]): Unit = {
    // Create a SparkSession.
    val spark = SparkSession.builder()
      .appName("HelloDataWorks")
      .getOrCreate()
    // Print the Spark version.
    println(s"Spark version: ${spark.version}")
  }
}

Após editar o código Scala, gere um pacote jar.

O pacote jar de exemplo chama-se SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar.

Faça o upload do recurso JAR.
1. Após concluir o desenvolvimento do código, faça login no console do OSS. No painel de navegação à esquerda da região, clique em Bucket list.
2. Clique no nome do bucket de destino para abrir a página File Management.
  
  Este tópico usa o bucket onaliyun-bucket-2 como exemplo.
3. Clique em Create Directory para criar um diretório onde armazenar o recurso JAR.
  
  Defina Catalog Name como emr/jars para criar o diretório de armazenamento do recurso JAR.
4. Envie o recurso JAR para o diretório de armazenamento.
  
  Acesse o diretório e clique em Upload File. Na área Files to Upload, clique em Scan Files, adicione o arquivo SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar ao bucket e, em seguida, clique em Upload File.

Referencie o recurso JAR.

Edite o código para referenciar o recurso JAR.

Na página de edição do nó EMR Spark criado, edite o código para referenciar o recurso JAR.

spark-submit --class com.aliyun.emr.example.spark.SparkMaxComputeDemo --master yarn ossref://onaliyun-bucket-2/emr/jars/SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar

Descrição dos parâmetros:

Parâmetro	Descrição
`class`	O nome totalmente qualificado da classe principal a ser executada.
`master`	O modo de execução da aplicação Spark.
Caminho do arquivo `ossref`	O formato é `ossref://{endpoint}/{bucket}/{object}`. Endpoint: O endpoint do serviço OSS. Se este parâmetro for deixado em branco, o recurso OSS deve estar na mesma região do cluster EMR. Bucket: Um contêiner no OSS para armazenar objetos. Cada bucket possui um nome exclusivo. Faça login no console do OSS para visualizar todos os buckets da sua conta. Object: Um objeto específico, como um nome de arquivo ou caminho, armazenado em um bucket.

Execute o job do nó EMR Spark.

Após concluir a edição, clique no ícone e selecione o grupo de recursos Serverless criado para executar o nó EMR Spark. Quando o job terminar, registre os applicationIds no console, por exemplo: application_1730367929285_xxxx.

Visualize o resultado.

Crie um nó EMR Shell e execute o comando yarn logs -applicationId application_1730367929285_xxxx no nó para ver o resultado da execução:

yarn logs -applicationId application_xxx
Process Output>>> LogLastModifiedTime:Fri Nov 01 16:27:46 +0800 2024
Process Output>>> LogLength:21
Process Output>>> LogContents:
Process Output>>> Spark version: 3.4.2
Process Output>>>

(Opcional) Configurar parâmetros avançados

Configure os seguintes parâmetros de propriedade exclusivos no painel Scheduling Settings, no lado direito do nó, em EMR Node Parameters > DataWorks parameters.

Nota

Os parâmetros avançados disponíveis variam conforme o tipo de cluster EMR, conforme mostrado nas tabelas a seguir.
Para outras propriedades do Spark open-source, configure-as no painel Scheduling Settings em EMR Node Parameters > Spark parameter.

Cluster Datalake/cluster personalizado: EMR on ECS

Parâmetro avançado	Descrição
queue	A fila de agendamento do job. O padrão é `default`. Se você configurou uma YARN Resource Queue no nível do workspace ao registrar o cluster EMR no workspace do DataWorks: Se você selecionar a caixa de seleção Global Settings Take Precedence e defini-la como Yes, o job Spark usará a fila configurada durante o registro do cluster EMR. Se você não selecionar a caixa de seleção, o job Spark usará a fila configurada para o nó EMR Spark. Para mais informações sobre o EMR YARN, consulte Configuração básica de fila. Para mais detalhes sobre a configuração de fila ao registrar um cluster EMR, consulte Definir a fila global de recursos YARN.
priority	A prioridade. O valor padrão é 1.
FLOW_SKIP_SQL_ANALYZE	O modo de execução para instruções SQL. Valores válidos: `true`: Executa várias instruções SQL simultaneamente. `false` (padrão): Executa uma instrução SQL por vez. Nota Este parâmetro é suportado apenas para execuções de teste no ambiente de desenvolvimento de dados.
Outros	Adicione parâmetros personalizados do Spark nas Configurações Avançadas. Por exemplo, se você adicionar `spark.eventLog.enabled : false` , o DataWorks adicionará automaticamente o parâmetro ao código enviado ao cluster EMR no formato `--conf key=value`. Também é possível configurar parâmetros globais do Spark. Para mais informações, consulte Definir parâmetros globais do Spark. Nota Para ativar o controle de acesso Ranger, adicione a configuração `spark.hadoop.fs.oss.authorization.method=ranger` em Definir parâmetros globais do Spark para garantir que o controle de acesso Ranger tenha efeito.

Cluster Spark: EMR on ACK

Parâmetro avançado

Descrição

FLOW_SKIP_SQL_ANALYZE

O modo de execução para instruções SQL. Valores válidos:

true: Executa várias instruções SQL simultaneamente.
false: Executa uma instrução SQL por vez.

Nota

Este parâmetro é suportado apenas para execuções de teste no ambiente de desenvolvimento de dados.

Outros

Adicione parâmetros personalizados do Spark nas Configurações Avançadas. Por exemplo: spark.eventLog.enabled: false. O DataWorks adiciona automaticamente o parâmetro ao código enviado ao cluster EMR no formato --conf key=value.
Também é possível configurar parâmetros globais do Spark. Para mais informações, consulte Definir parâmetros globais do Spark.

Cluster Hadoop: EMR on ECS

Parâmetro avançado	Descrição
queue	A fila de agendamento para a qual os jobs são enviados. Valor padrão: default. Se você configurar uma YARN resource queue no nível do workspace ao registrar um cluster EMR em um workspace do DataWorks, aplicam-se as seguintes regras: Se você selecionar a caixa de seleção Global Settings Take Precedence e definir o valor como Yes, a configuração da fila especificada durante o registro do cluster EMR terá precedência na execução de um job Spark. Se você desmarcar a caixa de seleção, a configuração da fila especificada para o nó EMR Spark terá precedência na execução de um job Spark. Para mais informações sobre o EMR YARN, consulte Configurações básicas de fila. Para mais detalhes sobre como configurar uma fila ao registrar um cluster EMR, consulte Definir a fila global de recursos YARN.
priority	A prioridade. O valor padrão é 1.
FLOW_SKIP_SQL_ANALYZE	O modo de execução para instruções SQL. Valores válidos: `true`: Executa várias instruções SQL simultaneamente. `false`: Executa uma instrução SQL por vez. Nota Este parâmetro é suportado apenas para execuções de teste no ambiente de desenvolvimento de dados.
USE_GATEWAY	Define se o job deste nó deve ser enviado por meio de um cluster Gateway. Valores válidos: `true`: Envia o job através do cluster Gateway. `false`: Não envia o job através de um cluster Gateway. Por padrão, o job é enviado para o nó header. Nota Se você definir este parâmetro como `true`, mas o cluster do nó não estiver associado a um cluster Gateway, os envios subsequentes de jobs EMR falharão.
Outros	Adicione parâmetros personalizados do Spark diretamente nas Configurações Avançadas. Para uma entrada como `spark.eventLog.enabled : false` , o DataWorks a adiciona automaticamente ao código enviado ao cluster EMR no formato `--conf key=value`. Também é possível configurar parâmetros globais do Spark. Para mais informações, consulte Definir parâmetros globais do Spark. Nota Para ativar o controle de acesso Ranger, adicione a configuração `spark.hadoop.fs.oss.authorization.method=ranger` em Definir parâmetros globais do Spark para garantir que o controle de acesso Ranger tenha efeito.

Cluster EMR Serverless Spark

Para mais informações sobre as configurações de parâmetros, consulte Definir parâmetros para enviar um job Spark.

Parâmetro avançado	Descrição
queue	A fila de agendamento do job. O padrão é `dev_queue`.
priority	A prioridade. O valor padrão é 1.
FLOW_SKIP_SQL_ANALYZE	O modo de execução para instruções SQL. Valores válidos: `true`: Executa várias instruções SQL simultaneamente. `false`: Executa uma instrução SQL por vez. Nota Este parâmetro é suportado apenas para execuções de teste no ambiente de desenvolvimento de dados.
SERVERLESS_RELEASE_VERSION	A versão do mecanismo Spark. Por padrão, o job usa a Default Engine Version configurada para o cluster no Management Center, em Clusters. Configure este parâmetro para definir versões diferentes do mecanismo para jobs distintos.
SERVERLESS_QUEUE_NAME	A fila de recursos. Por padrão, o job usa a Default Resource Queue configurada para o cluster no Management Center, em Clusters. Se houver requisitos de isolamento e gerenciamento de recursos, adicione filas. Para mais informações, consulte Gerenciar filas de recursos.
Outros	Adicione parâmetros personalizados do Spark na configuração avançada. Por exemplo, se você adicionar `spark.eventLog.enabled : false` , o DataWorks adicionará automaticamente o parâmetro ao código enviado ao cluster EMR no formato `--conf key=value`. Também é possível configurar parâmetros globais do Spark. Para mais informações, consulte Definir parâmetros globais do Spark.

Executar o job Spark

No painel Run Configuration, acesse a seção Compute Resource e configure o Compute Resource e o DataWorks Resource Group.
Nota
- Defina as CUs for Scheduling conforme os recursos necessários para a tarefa. A CU padrão é 0.25.
- Para acessar uma fonte de dados pela internet pública ou em uma VPC, use um grupo de recursos de agendamento com conectividade de rede verificada à fonte de dados. Para mais informações, consulte Soluções de conectividade de rede.
Na caixa de diálogo de parâmetros da barra de ferramentas, selecione a fonte de dados correspondente e clique em Run.

Para executar o job do nó periodicamente, configure suas propriedades de agendamento. Para mais informações, consulte Configurar propriedades de agendamento para um nó.

Nota
Se precisar personalizar o ambiente de componentes, crie uma imagem personalizada com base na imagem oficial dataworks_emr_base_task_pod e use a imagem no Desenvolvimento de Dados.

Por exemplo, ao criar uma imagem personalizada, substitua pacotes JAR do Spark ou dependa de libraries, files ou JAR packages específicos.
Após configurar o job do nó, implante o nó. Para mais informações, consulte Implantar um nó ou workflow.
Depois que o job for implantado, acesse o Operation Center para visualizar o status de execução do job periódico. Para mais informações, consulte Introdução ao Operation Center.

Perguntas frequentes

Por que o erro "DlfMetaStoreClientFactory not found" ocorre ao executar spark-submit no modo YARN-Cluster em um nó EMR Spark após ativar o Kerberos no cluster EMR?
P: Como resolver um tempo limite de conexão do nó?

R: Garanta que o grupo de recursos e o cluster possam se conectar mutuamente pela rede. Acesse a lista de recursos de computação e clique em Resource Initialization. Na caixa de diálogo exibida, clique em Re-initialize. Verifique se a inicialização foi bem-sucedida.

DataWorks:Nó EMR Spark

Pré-requisitos

Limitações

Observações

Desenvolver um job Spark e obter o pacote JAR

Procedimento

Desenvolver um job Spark

Opção 1: Upload e referência

JAR < 500 MB

JAR >= 500 MB

Opção 2: Referenciar um recurso OSS

Dependências POM

Plug-ins

(Opcional) Configurar parâmetros avançados

Cluster Datalake/cluster personalizado: EMR on ECS

Cluster Spark: EMR on ACK

Cluster Hadoop: EMR on ECS

Cluster EMR Serverless Spark

Executar o job Spark

Perguntas frequentes

Tópicos relacionados