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;
/**
 * プロジェクト: example_project 
 * テーブル: wc_in2 
 * パーティション: p1=2,p2=1 
 * 列: 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 コード例」セクションのコードを使用して、UDTF Java クラス「UDTFResource」を作成します。

2. MaxCompute Java モジュールの warehouse リソースの内容に基づいて、実行パラメーターを設定します。

   説明

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

   • 実行中、コードはローカルの file_resource.txt ファイル、wc_in1 テーブル (table_resource1 リソースにマッピング) 、および wc_in2 テーブルの p1=2, p2=1 パーティション (table_resource2 リソースにマッピング) からデータを読み取ります。

   

3. UDTFResource クラス名を右クリックし、実行 を選択します。

   

   

クライアントサイドテスト

1. IDEA の左上隅にある [Project Explorer] をクリックし、[Add Resource] を選択します。

   

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 テーブルを、table_resource1 リソースと table_resource2 リソースにマッピングします。

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

   

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

   

5. UDTF を JAR パッケージ化し、MaxCompute プロジェクトにアップロードして、関数「my_udtf」として登録します。デプロイするには、UDTFResource クラス名を右クリックし、[Deploy to Server...] を選択します。 表示されるダイアログボックスの [Extra resources] セクションに、必要な file_resource.txt、table_resource1、および table_resource2 リソースを追加します。

   

6. [Project Explorer] をクリックします。 ターゲットの MaxCompute プロジェクトを右クリックし、[Open Console] を選択して MaxCompute クライアントを起動します。 次に、SQL コマンドを実行して UDTF を呼び出します。

   

   

   以下はサンプルコマンドです：

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