Realtime Compute for Apache Flink:Java

注意事項

• UDF 開発時に JAR パッケージの依存関係の競合を回避するため、以下の点にご注意ください。

  • SQL 開発ページで選択した Flink のバージョンと、Pom 依存関係で指定した Flink のバージョンが一致していることを確認してください。

  • Flink 関連の依存関係については、<scope>provided</scope> を追加して、スコープを `provided` に設定してください。

  • その他のサードパーティ依存関係は、Shade メソッドを使用してパッケージ化してください。詳細については、「Apache Maven Shade Plugin」をご参照ください。

  Flink の依存関係競合に関する詳細については、「Flink の依存関係競合を解決する方法」をご参照ください。

• SQL ジョブ内で UDF を頻繁に呼び出すことによるタイムアウトを防止するには、UDF の JAR パッケージを依存ファイルとしてアップロードし、ジョブ内で CREATE TEMPORARY FUNCTION 構文を使用して関数を宣言します。例： CREATE TEMPORARY FUNCTION 'GetJson' AS 'com.soybean.function.udf.MyGetJsonObjectFunction';

UDF の分類

UDF の登録

• グローバル UDF の登録方法については、「グローバル UDF」をご参照ください。

• ジョブレベルの UDF の登録方法については、「ジョブレベルの UDF」をご参照ください。

UDF へのパラメーターの渡し方

Flink 開発コンソールで UDF のパラメーターを設定し、UDF コード内で使用できます。これにより、コンソール上で UDF のパラメーター値を直接素早く変更できます。

名前付きパラメーター

SQL で関数を呼び出す際には、すべてのパラメーターを正しい順序で指定する必要があります。関数のパラメーター数が多い場合、パラメーターの数や順序を誤りやすくなります。また、省略可能なパラメーターも省略できません。名前付きパラメーターを使用すると、必要なパラメーターのみを指定できるため、エラーの発生確率を低減でき、より便利です。以下のユーザー定義スカラー関数 (ScalarFunction) の例では、名前付きパラメーターの使用方法を示します。

// ユーザー定義スカラー関数を実装します。最後の 2 つの入力パラメーターは省略可能 (isOptional = true) です。
public class MyFuncWithNamedArgs extends ScalarFunction  {
	private static final long serialVersionUID = 1L;

	public String eval(@ArgumentHint(name = "f1", isOptional = false, type = @DataTypeHint("STRING")) String f1,
			@ArgumentHint(name = "f2", isOptional = true, type = @DataTypeHint("INT")) Integer i2,
			@ArgumentHint(name = "f3", isOptional = true, type = @DataTypeHint("LONG")) Long l3) {

		if (i2 != null) {
			return "i2#" + i2;
		}
		if (l3 != null) {
			return "l3#" + l3;
		}
		return "default#" + f1;
	}
}

この UDF を SQL で使用する場合、最初の必須パラメーターのみを指定したり、省略可能なパラメーターを任意に選択して指定したりできます。以下のコードを例として示します。

CREATE TEMPORARY FUNCTION MyNamedUdf AS 'com.aliyun.example.MyFuncWithNamedArgs';

CREATE temporary TABLE s1 (
    a INT,
    b BIGINT,
    c VARCHAR,
    d VARCHAR,
    PRIMARY KEY(a) NOT ENFORCED
) WITH (
    'connector' = 'datagen',
    'rows-per-second'='1'
);

CREATE temporary TABLE sink (
    a INT,
    b VARCHAR,
    c VARCHAR,
    d VARCHAR
) WITH (
    'connector' = 'print'
);

INSERT INTO sink
SELECT a,
    -- 最初の必須パラメーターのみを指定
    MyNamedUdf(f1 => c) arg1_res,
    -- 最初の必須パラメーターと 2 番目の省略可能なパラメーターを指定
    MyNamedUdf(f1 => c, f2 => a) arg2_res,
    -- 最初の必須パラメーターと 3 番目の省略可能なパラメーターを指定
    MyNamedUdf(f1 => c, f3 => d) arg3_res
FROM s1;

参照

• UDAF を使用したデータのソートおよび集計。

• Java UDF の開発および使用例については、「ユーザー定義集計関数 (UDAF)」、「ユーザー定義スカラー関数 (UDSF)」、および「ユーザー定義テーブル値関数 (UDTF)」をご参照ください。

• Python UDF のデバッグおよびチューニング方法については、「概要」をご参照ください。

• Python UDF の開発および使用例については、「ユーザー定義集計関数 (UDAF)」、「ユーザー定義スカラー関数 (UDSF)」、および「ユーザー定義テーブル値関数 (UDTF)」をご参照ください。