Data Lake Formation:Acessar o DLF a partir do EMR on ECS Spark

Pré-requisitos

• Crie um cluster EMR com a versão 5.12.0 ou posterior e selecione os componentes Spark3 e Paimon. Para solicitar suporte a uma versão diferente, entre em contato com a equipe de desenvolvimento do DLF pelo grupo do DingTalk com ID 106575000021.

• Conclua as etapas descritas em Get started with DLF.

• O cluster EMR e o DLF devem estar na mesma região, e a VPC do cluster EMR deve constar na lista de permissões do DLF.

Criar um catálogo DLF

Para mais informações, consulte Get started with DLF.

Conceder permissões do DLF a uma função

1. Conceda permissões do RAM à função AliyunECSInstanceForEMRRole. Ignore esta etapa após concluir a integração entre os product EMR e DLF.

   1. Faça login no console do RAM com sua conta Alibaba Cloud ou como administrador do RAM.

   2. No painel de navegação à esquerda, escolha Identities > Roles e localize a função AliyunECSInstanceForEMRRole.

   3. Na coluna Actions, clique em Add Permissions.

   4. Na aba Permission Policies, pesquise e selecione a política AliyunDLFFullAccess e clique em OK.

2. Conceda permissões do DLF à função AliyunECSInstanceForEMRRole.

   1. Acesse o console do DLF.

   2. Na página Catalogs, clique em no nome de um catálogo para abrir a página de detalhes.

   3. Para conceder permissões em todo o catálogo, clique em na aba Permissions. Para conceder permissões em um banco de dados ou tabela específica, acesse o recurso e clique em na aba Permissions correspondente.

   4. No painel Grant Permissions, configure os parâmetros abaixo e clique em OK.

      • Principal: Selecione RAM User/RAM Role.

      • Select Principal: Escolha AliyunECSInstanceForEMRRole na lista suspensa.

        Nota

        Se AliyunECSInstanceForEMRRole não aparecer na lista suspensa, clique em Sync User/Role na página de gerenciamento de usuários.

      • Predefined Permission Type: Personalize as permissões de leitura ou selecione um tipo predefinido, como Data Reader ou Data Editor.

Atualizar dependências do Paimon no EMR

Baixe os dois arquivos JAR a seguir (versão 1,1 ou posterior) do repositório Maven: paimon-jindo-*.jar e paimon-spark-3.x-*.jar. Verifique se as dependências são compatíveis com a versão do Spark do seu cluster EMR.

1. Importe as dependências do Paimon.

   1. Carregue os dois arquivos JAR de dependência, paimon-jindo-*.jar e paimon-spark-3.x-*.jar, no Object Storage Service (OSS). Defina a lista de controle de acesso (ACL) dos arquivos como leitura pública. Para mais informações, consulte Simple upload.

   2. Modifique o script abaixo e carregue-o no OSS.

      #!/bin/bash
      echo 'clean up paimon-dlf-2.5 exists file'
      rm -rf /opt/apps/PAIMON/paimon-dlf-2.5
      rm -rf /opt/apps/PAIMON/paimon-dlf-2.5.tar.gz.*
      cd /opt/apps/PAIMON/paimon-current/lib/spark3
      mkdir -p /opt/apps/PAIMON/paimon-dlf-2.5/lib/spark3
      cd /opt/apps/PAIMON/paimon-dlf-2.5/lib/spark3
      wget ${paimon-jindo-1.1.0.jar}
      wget ${paimon-spark-3.x-1.1.0.jar}
      echo 'link paimon-current to paimon-dlf-2.5'
      rm -f /opt/apps/PAIMON/paimon-current
      ln -sf /opt/apps/PAIMON/paimon-dlf-2.5 /opt/apps/PAIMON/paimon-current

      Importante

      Substitua os placeholders ${paimon-jindo-1.1.0.jar} e ${paimon-spark-3.x-1.1.0.jar} no script pelos caminhos de download correspondentes no OSS. Por padrão, clusters EMR on ECS não têm acesso à rede pública.

      • Endpoint interno: https://{bucket}.oss-cn-hangzhou-internal.aliyuncs.com/jars/paimon-jindo-1.1.0.jar

      • Endpoint público: https://{bucket}.oss-cn-hangzhou.aliyuncs.com/jars/paimon-jindo-1.1.0.jar

2. Execute o script como um script de inicialização do cluster EMR. Para mais informações, consulte Manually execute a script.

   1. No cluster EMR, acesse a aba Script Actions > Run Script e clique em Add and Run.

   2. Na caixa de diálogo exibida, configure os seguintes parâmetros e clique em OK.

      • Name: Insira um nome personalizado para o script.

      • Script Location: Selecione o script de atualização carregado no OSS. O caminho do script deve estar no formato oss://**/*.sh.

      • Target Scope: Selecione Cluster.

3. Após a conclusão do script, reinicie o service Spark para aplicar as alterações.

Ler e gravar dados com Spark

spark-sql --master yarn \
  --conf spark.driver.memory=5g \
  --conf spark.sql.defaultCatalog=paimon \
  --conf spark.sql.catalog.paimon=org.apache.paimon.spark.SparkCatalog \
  --conf spark.sql.catalog.paimon.metastore=rest \
  --conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions \
  --conf spark.sql.catalog.paimon.uri=http://${regionID}-vpc.dlf.aliyuncs.com \
  --conf spark.sql.catalog.paimon.warehouse=${catalog} \
  --conf spark.sql.catalog.paimon.token.provider=dlf \
  --conf spark.sql.catalog.paimon.dlf.token-loader=ecs

CREATE TABLE user_samples
(
    user_id INT,             
    age INT,           
    gender_code STRING,    
    clk BOOLEAN
);
CREATE TABLE user_samples_di (
    user_id INT,             
    age INT,           
    gender_code STRING,    
    clk BOOLEAN
)
USING CSV
OPTIONS(
'path'='oss://${bucket}/user/user_samples_di'
);

INSERT INTO user_samples VALUES
(1, 25, 'M', true),
(2, 18, 'F', false);
INSERT INTO user_samples_di VALUES
(1, 25, 'M', true),
(2, 18, 'F', true),
(3, 35, 'M', true);

SELECT * FROM user_samples;
SELECT * FROM user_samples_di;

spark-sql (default)> SELECT * FROM user_samples;
1	25	M	true
2	18	F	false

spark-sql (default)> SELECT * FROM user_samples_di;
1	25	M	true
2	18	F	true
3	35	M	true

MERGE INTO user_samples
USING user_samples_di
ON user_samples.user_id = user_samples_di.user_id
WHEN MATCHED THEN
UPDATE SET
  age = user_samples_di.age,
  gender_code = user_samples_di.gender_code,
  clk = user_samples_di.clk
WHEN NOT MATCHED THEN
  INSERT (user_id, age, gender_code, clk)
  VALUES (user_samples_di.user_id, user_samples_di.age, user_samples_di.gender_code, user_samples_di.clk);

spark-sql (default)> SELECT * FROM user_samples;
1	25	M	true
2	18	F	true
3	35	M	true