MaxCompute:Java UDAF

UDAF Kod Yapısı

// 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;
  }
}

Sınırlamalar

• UDF'leri Kullanarak İnternet'e Erişim

  Varsayılan olarak MaxCompute, UDF'ler kullanarak İnternet'e erişmenize izin vermez. UDF'ler aracılığıyla İnternet'e erişmek istiyorsanız, iş gereksinimlerinize göre ağ bağlantısı başvuru formunu doldurun ve başvuruyu gönderin. MaxCompute teknik destek ekibi, ağ bağlantısını etkinleştirmek için sizinle hızla iletişime geçecektir. Ağ bağlantısı başvuru formunun nasıl doldurulacağı hakkında daha fazla bilgi için bkz. Ağ Bağlantı Süreci.

• UDF'leri Kullanarak VPC'ye Erişim

  Varsayılan olarak MaxCompute, UDF'ler kullanarak VPC'lerdeki kaynaklara erişmenize izin vermez. UDF'leri kullanarak bir VPC'deki kaynaklara erişmek için MaxCompute ile VPC arasında bir ağ bağlantısı kurmanız gerekir. İlgili işlemler hakkında daha fazla bilgi için bkz. UDF'den VPC Kaynaklarına Erişim.

• UDF'ler, UDAF'ler veya UDTF'ler Kullanarak Tablo Verilerini Okuma

  Aşağıdaki tablo türlerinden veri okumak için UDF'leri, UDAF'leri veya UDTF'leri kullanamazsınız:

  • Şema evriminin uygulandığı tablo

  • Karmaşık veri türleri içeren tablo

  • JSON veri türleri içeren tablo

  • İşlemsel tablo

Kullanım Notları

@Resolve Ek Açıklaması

@Resolve(<signature>)

'arg_type_list -> type'

Aşağıda geçerli @Resolve ek açıklamalarına örnekler verilmiştir.

Veri Türleri

MaxCompute tarafından desteklenen veri türleri veri türü sürümüne göre değişir. MaxCompute 2.0'dan başlayarak, ARRAY, MAP ve STRUCT gibi karmaşık türler dahil olmak üzere ek veri türleri kullanılabilir. Daha fazla bilgi için bkz. Veri Türü Sürümleri.

Java UDAF'iniz, MaxCompute'daki türlerle eşleşen veri türlerini kullanmalıdır. Aşağıdaki tabloda bu eşlemeler açıklanmaktadır.

Kullanım

MaxCompute Studio kullanarak bir Java UDAF'in nasıl geliştirileceği ve çağrılacağına ilişkin tam bir örnek için bkz. Örnek.

Örnek

Bu örnek, ortalama bir değeri hesaplayan AggrAvg adlı bir UDAF geliştirmek için MaxCompute Studio'nun nasıl kullanılacağını gösterir. Aşağıdaki şekil mantığı görselleştirir.

1. Giriş verisi dilimleme: MaxCompute, bir çalışanın verimli bir şekilde işleyebileceği dilimlere giriş verilerini bölmek için MapReduce işleme iş akışını takip eder.

   Dilim boyutunu odps.stage.mapper.split.size parametresini kullanarak yapılandırmanız gerekir.

2. Ortalama hesaplamanın ilk aşaması: Her çalışan, kendi dilimindeki veri kayıtlarının sayısını sayar ve toplamlarını hesaplar. Her dilimden elde edilen sayım ve toplam, bir ara sonuç olarak kabul edilir.

3. Ortalama hesaplamanın ikinci aşaması: İlk aşamadaki her dilimden gelen ara sonuçlar toplanır.

4. Son çıktı: r.sum/r.count tüm giriş verilerinin ortalamasıdır.

Aşağıdaki adımlarda Java UDAF'in nasıl geliştirileceği ve çağrılacağı açıklanmaktadır:

1. Ortamı hazırlayın.

   MaxCompute Studio'da bir UDF geliştirmeden ve hata ayıklamadan önce MaxCompute Studio'yu yüklemeniz ve bir MaxCompute projesine bağlamanız gerekir. Daha fazla bilgi için aşağıdaki konulara bakın:

   1. MaxCompute Studio'yu Yükleme

   2. Bir MaxCompute Projesine Bağlanma

   3. Bir MaxCompute Java Modülü Oluşturma

2. UDAF kodunu yazın

   1. Project gezgininde, modülün kaynak kodu dizinine (src > main > java) sağ tıklayın ve New > MaxCompute Java öğelerini seçin.

   2. Create new MaxCompute java class iletişim kutusunda UDAF seçeneğine tıklayın, Name alanına bir ad girin ve Enter tuşuna basın. Bu örnek için Java sınıfının adını AggrAvg olarak belirleyin.

      Name, MaxCompute Java sınıfının adıdır. Bir paket oluşturmadıysanız, adı packagename.classname formatında girin. Bir paket otomatik olarak oluşturulur.

   3. Kod düzenleyicide aşağıdaki UDAF kodunu yapıştırın.

      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. UDAF'i yerel olarak hata ayıklayın

   Daha fazla hata ayıklama işlemi için bkz. UDF'leri Yerel Olarak Çalıştırarak Hata Ayıklama.

   Projenizdeki AggrAvg.java dosyasına sağ tıklayın ve Run 'AggrAvg.main()' seçeneğini belirleyin. Görünen Run/Debug Configurations iletişim kutusunda aşağıdaki parametreleri yapılandırın: MaxCompute project için local, MaxCompute table için kmeans_in, Table columns için dim1,dim2, Download Record limit için 100 ve Data Column Separator için virgül. Ardından OK düğmesine tıklayın.

   Not

   Çalıştırma parametreleri için referans olarak şekilde gösterilen verileri kullanabilirsiniz.

4. Oluşturduğunuz UDAF'i bir JAR dosyasında paketleyin, JAR dosyasını bir MaxCompute projesine yükleyin ve işlevi kaydedin. Örneğin, işlevin adı user_udaf olsun.

   Paketleme işlemleri hakkında daha fazla bilgi için bkz. Adımlar.

   IntelliJ IDEA'da, UDAF'i içeren Java dosyasına sağ tıklayın ve Deploy to server... seçeneğini belirleyin. Package a jar, submit resource and register function iletişim kutusunda MaxCompute project, Resource name, Main class (UDAF sınıf adını girin) ve Function name seçeneklerini yapılandırın. Force update if already exists onay kutusunu seçin ve dağıtımı tamamlamak için OK düğmesine tıklayın.

5. MaxCompute Studio'nun sol gezinme bölmesinde Project Explorer seçeneğine tıklayın, hedef MaxCompute projesine sağ tıklayın, MaxCompute istemcisini başlatın ve yeni oluşturulan UDAF'i çağırmak için bir SQL komutu yürütün.

   Hedef tablo my_table için şema ve verilerin aşağıdaki gibi olduğunu varsayalım.

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

   UDAF'i çağırmak için aşağıdaki SQL ifadesini çalıştırın.

   select user_udaf(col0) as c0 from my_table;

   Aşağıdaki sonuç döndürülür.

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