MaxCompute:Exemplo: Ler recursos do MaxCompute com uma UDTF Java

Pré-requisitos

• Instale o MaxCompute Studio, conecte-se a um projeto do MaxCompute e crie um módulo Java do MaxCompute.

• Instale a ferramenta de desenvolvimento IntelliJ IDEA 2024 e o JDK 1.8.

• Para mais informações, consulte Instale o MaxCompute Studio, Gerencie conexões de projeto e Crie um módulo Java do MaxCompute.

• Para obter detalhes sobre os recursos do MaxCompute, consulte Resources.

Exemplo de código da UDTF

O código a seguir apresenta a implementação da UDTF Java.

package com.aliyun.odps.examples.udf;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Iterator;
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.UDTF;
import com.aliyun.odps.udf.annotation.Resolve;
/**
 * project: example_project 
 * table: wc_in2 
 * partitions: p1=2,p2=1 
 * columns: cola,colc
 */
@Resolve("string,string->string,bigint,string")
public class UDTFResource extends UDTF {
  ExecutionContext ctx;
  long fileResourceLineCount;
  long tableResource1RecordCount;
  long tableResource2RecordCount;
  @Override
  public void setup(ExecutionContext ctx) throws UDFException {
  this.ctx = ctx;
  try {
   InputStream in = ctx.readResourceFileAsStream("file_resource.txt");
   BufferedReader br = new BufferedReader(new InputStreamReader(in));
   String line;
   fileResourceLineCount = 0;
   while ((line = br.readLine()) != null) {
     fileResourceLineCount++;
   }
   br.close();
   Iterator<Object[]> iterator = ctx.readResourceTable("table_resource1").iterator();
   tableResource1RecordCount = 0;
   while (iterator.hasNext()) {
     tableResource1RecordCount++;
     iterator.next();
   }
   iterator = ctx.readResourceTable("table_resource2").iterator();
   tableResource2RecordCount = 0;
   while (iterator.hasNext()) {
     tableResource2RecordCount++;
     iterator.next();
   }
 } catch (IOException e) {
   throw new UDFException(e);
 }
}
   @Override
   public void process(Object[] args) throws UDFException {
     String a = (String) args[0];
     long b = args[1] == null ? 0 : ((String) args[1]).length();
     forward(a, b, "fileResourceLineCount=" + fileResourceLineCount + "|tableResource1RecordCount="
     + tableResource1RecordCount + "|tableResource2RecordCount=" + tableResource2RecordCount);
    }
}

A dependência pom.xml a seguir é necessária para testes locais.

<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>odps-udf-local</artifactId>
    <version>0.48.0-public</version>
</dependency>

Procedimento

Teste local

1. No MaxCompute Studio, crie uma classe Java UDTF chamada UDTFResource usando o código da seção Exemplo de código da UDTF.

2. Configure os parâmetros de execução com base no conteúdo do recurso warehouse no módulo Java do MaxCompute.

   Nota

   • Os parâmetros de entrada correspondem aos valores da primeira e da terceira colunas de cada linha na partição p1=2, p2=1 da tabela wc_in2 no recurso local.

   • Durante a execução, o código lê dados do arquivo local file_resource.txt, da tabela wc_in1 (mapeada para o recurso table_resource1) e da partição p1=2, p2=1 da tabela wc_in2 (mapeada para o recurso table_resource2).

   

3. Clique com o botão direito no nome da classe UDTFResource e selecione Run.

   

   

Teste no cliente

1. No canto superior esquerdo do IDEA, clique em Project Explorer e selecione Add Resource.

   

2. Adicione o arquivo file_resource.txt conforme as informações da sua instância do MaxCompute.

   

3. No seu projeto do MaxCompute, crie e preencha as tabelas de dados de exemplo wc_in1 e wc_in2.

   CREATE TABLE wc_in1 (
     col1 STRING,
     col2 STRING,
     col3 STRING,
     col4 STRING
   );
   
   INSERT INTO wc_in1 VALUES 
   ('A1','A2','A3','A4'),
   ('A1','A2','A3','A4'),
   ('A1','A2','A3','A4'),
   ('A1','A2','A3','A4');
   
   CREATE TABLE wc_in2 (
       cola STRING,
       colb STRING,
       colc STRING
   ) PARTITIONED BY (p1 STRING, p2 STRING);
   
   ALTER TABLE wc_in2 ADD PARTITION  (p1='2',p2='1');
   
   INSERT INTO wc_in2 PARTITION (p1='2',p2='1') VALUES 
   ('three1','three2','three3'),
   ('three1','three2','three3'),
   ('three1','three2','three3'); 

4. Mapeie as tabelas wc_in1 e wc_in2 criadas no MaxCompute para os recursos table_resource1 e table_resource2.

   Adicione o recurso wc_in1.

   

   Adicione o recurso wc_in2.

   

5. Empacote a UDTF em um arquivo JAR, faça upload para o seu projeto do MaxCompute e registre-a como uma função chamada my_udtf. Para implantar, clique com o botão direito no nome da classe UDTFResource e selecione Deploy to Server.... Na caixa de diálogo exibida, adicione os recursos necessários file_resource.txt, table_resource1 e table_resource2 na seção Extra resources.

   

6. Clique em Project Explorer. Clique com o botão direito no projeto do MaxCompute desejado e selecione Open Console para iniciar o cliente do MaxCompute. Em seguida, execute um comando SQL para chamar a UDTF.

   

   

   Exemplo de comando:

   SELECT my_udtf("10","20") AS (a, b, fileResourceLineCount);