MaxCompute:UDAF Java

Structure du code de l'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

• Accéder à Internet via des UDF

  Par défaut, MaxCompute n'autorise pas l'accès à Internet via des UDF. Si vous souhaitez accéder à Internet via des UDF, remplissez le formulaire de demande de connexion réseau en fonction de vos besoins métier et soumettez la demande. L'équipe de support technique MaxCompute vous contactera rapidement pour activer la connectivité réseau. Pour plus d'informations sur la manière de remplir le formulaire de demande de connexion réseau, consultez la section Processus de connexion réseau.

• Accéder à un VPC via des UDF

  Par défaut, MaxCompute n'autorise pas l'accès aux ressources des VPC via des UDF. Pour utiliser des UDF afin d'accéder aux ressources d'un VPC, vous devez établir une connexion réseau entre MaxCompute et le VPC. Pour plus d'informations sur les opérations associées, consultez la section Accéder aux ressources VPC depuis une UDF.

• Lire les données de table via des UDF, UDAF ou UDTF

  Vous ne pouvez pas utiliser des UDF, UDAF ou UDTF pour lire les données des types de tables suivants :

  • Table sur laquelle une évolution de schéma est effectuée

  • Table contenant des types de données complexes

  • Table contenant des types de données JSON

  • Table transactionnelle

Remarques relatives à l'utilisation

Annotation @Resolve

@Resolve(<signature>)

'arg_type_list -> type'

Voici des exemples d'annotations @Resolve valides.

Types de données

Les types de données pris en charge par MaxCompute varient selon l'édition de type de données. À partir de MaxCompute 2.0, des types de données supplémentaires sont disponibles, notamment des types complexes tels que ARRAY, MAP et STRUCT. Pour plus d'informations, consultez la section Éditions de types de données.

Votre UDAF Java doit utiliser des types de données qui correspondent à ceux de MaxCompute. Le tableau suivant décrit ces correspondances.

Utilisation

Pour un exemple complet de développement et d'appel d'une UDAF Java à l'aide de MaxCompute Studio, consultez la section Exemple.

Exemple

Cet exemple montre comment utiliser MaxCompute Studio pour développer une UDAF nommée AggrAvg qui calcule une valeur moyenne. La figure suivante illustre la logique.

1. Découpage des données d'entrée : MaxCompute suit le flux de traitement MapReduce pour diviser les données d'entrée en tranches qu'un nœud de travail peut traiter efficacement.

   Configurez la taille de la tranche à l'aide du paramètre odps.stage.mapper.split.size.

2. Première phase du calcul de la moyenne : chaque nœud de travail compte le nombre d'enregistrements de données et calcule leur somme au sein de sa tranche. Le comptage et la somme de chaque tranche sont considérés comme un résultat intermédiaire.

3. Deuxième phase du calcul de la moyenne : les résultats intermédiaires de chaque tranche de la première phase sont agrégés.

4. Sortie finale : r.sum/r.count représente la moyenne de toutes les données d'entrée.

Les étapes suivantes décrivent comment développer et appeler l'UDAF Java :

1. Préparez l'environnement.

   Avant de pouvoir développer et déboguer une UDF dans MaxCompute Studio, installez MaxCompute Studio et connectez-le à un projet MaxCompute. Pour plus d'informations, consultez les rubriques suivantes :

   1. Installer MaxCompute Studio

   2. Se connecter à un projet MaxCompute

   3. Créer un module Java MaxCompute

2. Écrire le code UDAF

   1. Dans l'explorateur Project, cliquez avec le bouton droit sur le répertoire source du module (src > main > java) et sélectionnez New > MaxCompute Java.

   2. Dans la boîte de dialogue Create new MaxCompute java class, cliquez sur UDAF, saisissez un nom dans le champ Name et appuyez sur Entrée. Pour cet exemple, nommez la classe Java AggrAvg.

      Le Name correspond au nom de la classe Java MaxCompute. Si vous n'avez pas créé de package, saisissez le nom au format packagename.classname. Un package est automatiquement généré.

   3. Dans l'éditeur de code, collez le code UDAF suivant.

      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. Déboguer l'UDAF localement

   Pour plus d'opérations de débogage, consultez la section Déboguer les UDF en les exécutant localement.

   Cliquez avec le bouton droit sur le fichier AggrAvg.java dans votre projet et choisissez Run 'AggrAvg.main()'. Dans la boîte de dialogue Run/Debug Configurations qui s'affiche, configurez les paramètres suivants : MaxCompute project sur local, MaxCompute table sur kmeans_in, Table columns sur dim1,dim2, Download Record limit sur 100 et Data Column Separator sur une virgule. Ensuite, cliquez sur OK.

   Remarque

   Vous pouvez utiliser les données présentées dans la figure comme référence pour les paramètres d'exécution.

4. Empaquetez l'UDAF que vous avez créée dans un fichier JAR, téléchargez le fichier JAR dans un projet MaxCompute et enregistrez la fonction. Par exemple, la fonction est nommée user_udaf.

   Pour plus d'informations sur les opérations d'empaquetage, consultez la section Étapes.

   Dans IntelliJ IDEA, cliquez avec le bouton droit sur le fichier Java contenant l'UDAF et choisissez Deploy to server.... Dans la boîte de dialogue Package a jar, submit resource and register function, configurez MaxCompute project, Resource name, Main class (saisissez le nom de la classe UDAF) et Function name. Cochez la case Force update if already exists et cliquez sur OK pour terminer le déploiement.

5. Dans le volet de navigation de gauche de MaxCompute Studio, cliquez sur Project Explorer, cliquez avec le bouton droit sur le projet MaxCompute cible, démarrez le client MaxCompute et exécutez une commande SQL pour appeler l'UDAF nouvellement créée.

   Supposons que la table cible my_table possède le schéma et les données suivants.

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

   Exécutez l'instruction SQL suivante pour appeler l'UDAF.

   select user_udaf(col0) as c0 from my_table;

   Le résultat suivant est renvoyé.

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