MaxCompute:Java UDTFを使用したMaxComputeリソースの読み取り

UDTFコードの例

次のサンプルコードはJava UDTFです。

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

次のコードは、ローカルテスト用にpom.xmlファイルに必要な依存関係を示しています。

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

手順

ローカルテスト

1. MaxCompute StudioでUDTFタイプの新しいJavaプログラムを作成します。 たとえば、JavaクラスにUDTFResourceという名前を付け、UDTFコード例のプログラムコードを使用します。

2. Javaモジュールのwarehouseリソースに基づいてランタイムパラメータを設定します。

   説明

   • 入力パラメータは、ローカルリソース内のwc_in2テーブルのパーティションp1=2、p2=1の各行の第1列と第3列の値です。

   • コードを実行すると、ローカルリソースfile_resource.txtからデータ、テーブルwc_in1に対応するtable_resource1からデータ、およびパラメーターp1=2およびp2=1のテーブルwc_in2に対応するtable_resource2からデータが取得されます。

   

3. UDTFResourceクラスを右クリックし、[実行] を選択してプログラムを実行します。 結果が表示されます。

   

   

クライアントテスト

1. IDEAの左上隅にある [Project Explorer] をクリックし、[リソースの追加] を選択します。

   

2. MaxComputeインスタンス情報に基づいて、file_resource.txtファイルを追加します。

   

3. MaxComputeプロジェクトにサンプルデータテーブルwc_in1と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. MaxComputeで作成されたwc_in1およびwc_in2テーブルを、Resourceのtable_resource1およびtable_resource2にマッピングします。

   wc_in1リソースを追加します。

   

   wc_in2リソースを追加します。

   

5. 作成したUDTFをJARファイルにパッケージ化し、MaxComputeプロジェクトにアップロードして、関数を登録します。 たとえば、関数名はmy_udtfです。 UDTFResourceクラスを右クリックし、[サーバーにデプロイ] を選択して、パッケージ化とアップロードのインターフェイスを入力します。 [Extra resources] セクションにfile_resource.txt、table_resource1、table_resource2などの必要なリソースを含めます。

   

6. IDEAの左上隅にある [Project Explorer] をクリックし、対象のMaxComputeプロジェクトを右クリックし、[コンソールを開く] を選択してMaxComputeクライアントを起動し、SQLコマンドを実行して新しく作成したUDTFを呼び出します。 結果が表示されます。

   

   

   サンプルSQLコマンド:

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