MaxCompute:UDAF Java

Estrutura do código UDAF

// Package the defined Java class in org.alidata.odps.udaf.examples.
package org.alidata.odps.udaf.examples;
// Import the required base classes.
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import com.aliyun.odps.io.DoubleWritable;
import com.aliyun.odps.io.Writable;
import com.aliyun.odps.udf.Aggregator;
import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.annotation.Resolve;
// Define the custom Java class.
// Specify the @Resolve annotation.
@Resolve("double->double")
public class AggrAvg extends Aggregator {
// Implement the methods for the Java class.
  private static class AvgBuffer implements Writable {
    private double sum = 0;
    private long count = 0;
    @Override
    public void write(DataOutput out) throws IOException {
      out.writeDouble(sum);
      out.writeLong(count);
    }
    @Override
    public void readFields(DataInput in) throws IOException {
      sum = in.readDouble();
      count = in.readLong();
    }
  }
  private DoubleWritable ret = new DoubleWritable();
  @Override
  public Writable newBuffer() {
    return new AvgBuffer();
  }
  @Override
  public void iterate(Writable buffer, Writable[] args) throws UDFException {
    DoubleWritable arg = (DoubleWritable) args[0];
    AvgBuffer buf = (AvgBuffer) buffer;
    if (arg != null) {
      buf.count += 1;
      buf.sum += arg.get();
    }
  }
  @Override
  public Writable terminate(Writable buffer) throws UDFException {
    AvgBuffer buf = (AvgBuffer) buffer;
    if (buf.count == 0) {
      ret.set(0);
    } else {
      ret.set(buf.sum / buf.count);
    }
    return ret;
  }
  @Override
  public void merge(Writable buffer, Writable partial) throws UDFException {
    AvgBuffer buf = (AvgBuffer) buffer;
    AvgBuffer p = (AvgBuffer) partial;
    buf.sum += p.sum;
    buf.count += p.count;
  }
}

Limites

• Acesso à Internet via UDFs

  Por padrão, o MaxCompute não permite acesso à Internet por meio de UDFs. Se precisar acessar a Internet usando UDFs, preencha o formulário de solicitação de conexão de rede conforme suas necessidades de negócio e envie a solicitação. A equipe de suporte técnico do MaxCompute entrará em contato prontamente para habilitar a conectividade de rede. Para mais detalhes sobre como preencher o formulário de solicitação de conexão de rede, consulte Network connection process.

• Acesso a VPC via UDFs

  Por padrão, o MaxCompute não permite o acesso a recursos em VPCs por meio de UDFs. Para acessar recursos em uma VPC usando UDFs, estabeleça uma conexão de rede entre o MaxCompute e a VPC. Para mais informações sobre as operações relacionadas, consulte Access VPC resources from a UDF.

• Leitura de dados de tabelas via UDFs, UDAFs ou UDTFs

  Não use UDFs, UDAFs ou UDTFs para ler dados dos seguintes tipos de tabelas:

  • Tabelas com evolução de schema

  • Tabelas com tipos de dados complexos

  • Tabelas com tipos de dados JSON

  • Tabelas transacionais

Observações de uso

Anotação @Resolve

@Resolve(<signature>)

'arg_type_list -> type'

A seguir, exemplos válidos de anotações @Resolve.

Tipos de dados

Os tipos de dados suportados pelo MaxCompute variam conforme a edição de tipos de dados. A partir do MaxCompute 2.0, novos tipos estão disponíveis, incluindo tipos complexos como ARRAY, MAP e STRUCT. Para mais informações, consulte Data type editions.

Sua UDAF Java deve usar tipos de dados compatíveis com os do MaxCompute. A tabela a seguir descreve esses mapeamentos.

Uso

Para um exemplo completo de desenvolvimento e chamada de uma UDAF Java usando o MaxCompute Studio, consulte Example.

Exemplo

Este exemplo demonstra como usar o MaxCompute Studio para desenvolver uma UDAF chamada AggrAvg, que calcula um valor médio. A figura a seguir ilustra a lógica envolvida.

1. Fatiamento dos dados de entrada: O MaxCompute segue o fluxo de processamento MapReduce para dividir os dados de entrada em fatias que um worker possa processar eficientemente.

   Configure o tamanho da fatia por meio do parâmetro odps.stage.mapper.split.size.

2. Primeira fase do cálculo da média: Cada worker conta o número de registros de dados e calcula sua soma dentro de sua fatia. A contagem e a soma de cada fatia constituem um resultado intermediário.

3. Segunda fase do cálculo da média: Os resultados intermediários de cada fatia da primeira fase são agregados.

4. Saída final: r.sum/r.count representa a média de todos os dados de entrada.

As etapas a seguir descrevem como desenvolver e chamar a UDAF Java:

1. Prepare o ambiente.

   Antes de desenvolver e depurar uma UDF no MaxCompute Studio, instale o MaxCompute Studio e conecte-o a um projeto MaxCompute. Para mais informações, consulte os seguintes tópicos:

   1. Install MaxCompute Studio

   2. Connect to a MaxCompute project

   3. Create a MaxCompute Java module

2. Escreva o código da UDAF

   1. No explorador Project, clique com o botão direito no diretório de código-fonte do módulo (src > main > java) e selecione New > MaxCompute Java.

   2. Na caixa de diálogo Create new MaxCompute java class, clique em UDAF, insira um nome no campo Name e pressione Enter. Neste exemplo, nomeie a classe Java como AggrAvg.

      O campo Name refere-se ao nome da classe Java do MaxCompute. Se ainda não tiver criado um pacote, insira o nome no formato packagename.classname. Um pacote será gerado automaticamente.

   3. No editor de código, cole o seguinte código da UDAF.

      import java.io.DataInput;
      import java.io.DataOutput;
      import java.io.IOException;
      import com.aliyun.odps.io.DoubleWritable;
      import com.aliyun.odps.io.Writable;
      import com.aliyun.odps.udf.Aggregator;
      import com.aliyun.odps.udf.UDFException;
      import com.aliyun.odps.udf.annotation.Resolve;
      @Resolve("double->double")
      public class AggrAvg extends Aggregator {
        private static class AvgBuffer implements Writable {
          private double sum = 0;
          private long count = 0;
          @Override
          public void write(DataOutput out) throws IOException {
            out.writeDouble(sum);
            out.writeLong(count);
          }
          @Override
          public void readFields(DataInput in) throws IOException {
            sum = in.readDouble();
            count = in.readLong();
          }
        }
        private DoubleWritable ret = new DoubleWritable();
        @Override
        public Writable newBuffer() {
          return new AvgBuffer();
        }
        @Override
        public void iterate(Writable buffer, Writable[] args) throws UDFException {
          DoubleWritable arg = (DoubleWritable) args[0];
          AvgBuffer buf = (AvgBuffer) buffer;
          if (arg != null) {
            buf.count += 1;
            buf.sum += arg.get();
          }
        }
        @Override
        public Writable terminate(Writable buffer) throws UDFException {
          AvgBuffer buf = (AvgBuffer) buffer;
          if (buf.count == 0) {
            ret.set(0);
          } else {
            ret.set(buf.sum / buf.count);
          }
          return ret;
        }
        @Override
        public void merge(Writable buffer, Writable partial) throws UDFException {
          AvgBuffer buf = (AvgBuffer) buffer;
          AvgBuffer p = (AvgBuffer) partial;
          buf.sum += p.sum;
          buf.count += p.count;
        }
      }

3. Depure a UDAF localmente

   Para mais operações de depuração, consulte Debug UDFs by running them locally.

   Clique com o botão direito no arquivo AggrAvg.java em seu projeto e escolha Run 'AggrAvg.main()'. Na caixa de diálogo Run/Debug Configurations exibida, configure os seguintes parâmetros: MaxCompute project como local, MaxCompute table como kmeans_in, Table columns como dim1,dim2, Download Record limit como 100 e Data Column Separator como vírgula. Em seguida, clique em OK.

   Nota

   Use os dados mostrados na figura como referência para os parâmetros de execução.

4. Empacote a UDAF criada em um arquivo JAR, faça upload do arquivo JAR para um projeto MaxCompute e registre a função. Por exemplo, suponha que a função se chame user_udaf.

   Para mais informações sobre operações de empacotamento, consulte Steps.

   No IntelliJ IDEA, clique com o botão direito no arquivo Java que contém a UDAF e escolha Deploy to server.... Na caixa de diálogo Package a jar, submit resource and register function, configure MaxCompute project, Resource name, Main class (insira o nome da classe da UDAF) e Function name. Marque a caixa de seleção Force update if already exists e clique em OK para concluir a implantação.

5. Na barra de navegação à esquerda do MaxCompute Studio, clique em Project Explorer, clique com o botão direito no projeto MaxCompute desejado, inicie o cliente MaxCompute e execute um comando SQL para chamar a UDAF recém-criada.

   Suponha que a tabela alvo my_table tenha o seguinte schema e dados.

   +------------+------------+
   | col0       | col1       |
   +------------+------------+
   | 1.2        | 2.0        |
   | 1.6        | 2.1        |
   +------------+------------+

   Execute a seguinte instrução SQL para chamar a UDAF.

   select user_udaf(col0) as c0 from my_table;

   O resultado retornado é o seguinte.

   +----+
   | c0 |
   +----+
   | 1.4|
   +----+