Realtime Compute for Apache Flink:Java UDTF

設定パラメーター

次のパラメーターは、非同期テーブル関数のランタイム動作を制御します。

開発例

AsyncTableFunction を実装する場合、public で非 static の eval メソッドを宣言します。最初のパラメーターは、結果を送信するために使用される CompletableFuture<Collection<T>> であり、残りのパラメーターは SQL 呼び出しによって渡されるビジネスパラメーターです。

関数は、非同期リクエストを送信した後、すぐに戻り、完了コールバックで result.complete(rows) を呼び出して結果を送信するか、result.completeExceptionally(error) を呼び出して例外を送信します。1 回の呼び出しで得られる複数の行は、1 つのコレクションとして送信されます。結果がない場合は、空のコレクションを送信します。

package com.example.udf;

import org.apache.flink.table.functions.AsyncTableFunction;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.types.Row;
import org.apache.flink.table.functions.FunctionContext;

import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.stream.Collectors;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

/**
 * 非同期 HTTP インターフェイスを介して製品タグをクエリします。
 *
 * <p>インターフェイスの仕様： 200 は 1 行に 1 つのタグを含む UTF-8 テキストを返し、204 と 404 はタグがないことを示します。
 * この例は、短いテキストを返す読み取り専用クエリに適しています。他のステータスコードは呼び出し失敗として扱われ、エンジンに渡されます。
 */
@FunctionHint(output = @DataTypeHint("ROW<tag STRING>"))
public class AsyncProductTags extends AsyncTableFunction<Row> {

    private transient HttpClient client;
    private transient URI serviceBaseUri;

    @Override
    public void open(FunctionContext context) {
        String baseUrl = context.getJobParameter("product.service.base-url", null);
        if (baseUrl == null || baseUrl.isBlank()) {
            throw new IllegalArgumentException("product.service.base-url を設定してください。");
        }
        serviceBaseUri = URI.create(baseUrl);
        if (!("https".equalsIgnoreCase(serviceBaseUri.getScheme())
                        || "http".equalsIgnoreCase(serviceBaseUri.getScheme()))
                || serviceBaseUri.getHost() == null
                || serviceBaseUri.getRawQuery() != null
                || serviceBaseUri.getRawFragment() != null
                || !baseUrl.endsWith("/")) {
            throw new IllegalArgumentException("サービス URL は、スラッシュ (/) で終わる HTTP(S) ベース URL である必要があります。");
        }
        client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
    }

    @Override
    public boolean isDeterministic() {
        // 製品情報は変更される可能性があります。同じ ID が異なる時点で異なる結果を返す可能性があります。
        return false;
    }

    public void eval(CompletableFuture<Collection<Row>> result, Long productId) {
        if (productId == null) {
            result.complete(Collections.emptyList());
            return;
        }

        try {
            HttpRequest request =
                    HttpRequest.newBuilder(
                                    serviceBaseUri.resolve("products/" + productId + "/tags"))
                            .timeout(Duration.ofSeconds(5))
                            .header("Accept", "text/plain")
                            .GET()
                            .build();

            client.sendAsync(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8))
                    .thenApply(
                            response -> {
                                if (response.statusCode() == 200) {
                                    return response.body().lines()
                                            .map(String::strip)
                                            .filter(tag -> !tag.isEmpty())
                                            .map(tag -> Row.of(tag))
                                            .collect(Collectors.toList());
                                }
                                if (response.statusCode() == 204 || response.statusCode() == 404) {
                                    return Collections.<Row>emptyList();
                                }
                                throw new CompletionException(
                                        new IOException(
                                                "製品のクエリに失敗しました。HTTP ステータスコード： " + response.statusCode()));
                            })
                    .whenComplete(
                            (rows, error) -> {
                                if (error != null) {
                                    result.completeExceptionally(error);
                                } else {
                                    result.complete(rows);
                                }
                            });
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }
    }
}

HttpClient.sendAsync はネットワークリクエストを非同期で実行するため、関数は独自のスレッドプールを作成する必要はありません。コールバックは短いテキストのみを解析し、結果を送信します。別のクライアントを使用する場合は、そのクライアントが提供する非同期インターフェイスを優先的に使用してください。JDK クライアントの動作については、「HttpClient API」をご参照ください。

製品タグは時間の経過とともに変更される可能性があるため、この例では isDeterministic() を false に設定しています。同じ入力に対して常に同じ結果を返す関数の場合は、デフォルトの動作を維持してください。

カスタムタイムアウト処理

デフォルトでは、eval 呼び出しがタイムアウトする前に CompletableFuture を完了せず、設定されたリトライが使い果たされた場合、フレームワークは java.util.concurrent.TimeoutException をスローし、ジョブを失敗させます。AsyncTableFunction サブクラスで eval メソッドに対応する timeout(...) メソッドを宣言して、フォールバック行を返すか、カスタムタイムアウト処理を実装できます。

リモート LLM クラスターを呼び出す場合、タイムアウトは比較的に発生しやすく、通常、タイムアウトによってジョブを直接失敗させるべきではありません。timeout メソッドを宣言すると、タイムアウト時にフォールバック結果が返され、ジョブは続行されます。

timeout メソッドは、以下のすべての条件を満たす必要があります。

• メソッド宣言： メソッドは public インスタンスメソッドである必要があり、static にすることはできず、timeout という名前である必要があります。

• eval と一致するシグネチャ: パラメーターリストは、対応する eval メソッドのものと厳密に同一である必要があります。最初のパラメーターは CompletableFuture で、そのジェネリック型は eval のものと厳密に同じである必要があります。また、残りのパラメーターの型と順序も厳密に同じである必要があります。timeout はオーバーロードをサポートしています。フォールバックロジックを必要とする各 eval のオーバーロードに対して、個別の timeout メソッドを宣言できます。

• 同期完了 (必須)： このコールバックはオペレーターのメールボックススレッドで実行され、メソッドが戻る前に future を完了する必要があります。メソッドが戻った直後、フレームワークは future.isDone() をチェックします。future が完了していない場合、フレームワークは IllegalStateException で強制的に完了します。したがって：

  • メソッド本体で別のリトライや 2 回目のクエリなど、別の非同期呼び出しを開始し、そのコールバックが future を完了することを期待しないでください。コールバックがトリガーされるまでに、フレームワークはすでにレコードを放棄しています。

  • 同じ理由で、future を非同期で完了するために新しいスレッドを開始しないでください。

  • メソッド本体で行うことは 1 つだけです。定数行、NULL 行、空のコレクションなどの軽量なフォールバック結果を提供するか、カスタムビジネス例外で completeExceptionally(...) を呼び出します。

• 安全な例外スロー： メソッド本体で同期的に例外をスローすることは可能です。フレームワークは例外をダウンストリームの ResultFuture に渡します。これは、明示的に future.completeExceptionally(thrown) を呼び出すのと同等です。

これらの制約に加えて、フレームワークは次のシナリオでも安定した動作を保証します。

• timeout メソッドが宣言されていない： フレームワークのデフォルトの TimeoutException が使用され、コード生成中に失敗は発生しません。

• 互換性のないシグネチャ： timeout メソッドが有効な可視性を持つものの、そのパラメーターリストが呼び出しサイトの実際の引数型と一致しない場合、フレームワークは計画時 (コード生成) に ValidationException で失敗します。エラーメッセージには、関数の完全修飾クラス名が含まれ、期待されるシグネチャと実際のシグネチャの両方が表示されます。

• フォールバックとしての空のコレクション： future.complete(Collections.emptyList()) が呼び出されると、INNER ルックアップ結合では行が破棄され、LEFT OUTER ルックアップ結合では右テーブルのフィールドが NULL で埋められます。

次の例は、非同期テーブル関数に timeout コールバックを追加し、future を同期的に完了してフォールバック行を返す方法を示しています。

import org.apache.flink.table.functions.AsyncTableFunction;

import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;

public static class BackgroundFunctionWithTimeout extends AsyncTableFunction<Long> {

    public void eval(CompletableFuture<Collection<Long>> future, Integer waitMax) {
        // ...前の開発例と同じロジック：非同期タスクをディスパッチし、
        // コールバックスレッドで `future` を完了します。
    }

    // パラメーターリストは eval() と一致させる必要があります。同じ CompletableFuture ジェネリック型、
    // および残りのパラメーター (Integer waitMax) の同じ型と順序。
    // メソッド本体は `future` を同期的に完了する必要があります。別の非同期呼び出しを開始したり、
    // 新しいスレッドを開始して非同期で完了したりしないでください。このコールバックがトリガーされると、
    // オペレーターはすでにこのレコードを放棄しています。
    public void timeout(CompletableFuture<Collection<Long>> future, Integer waitMax) {
        future.complete(Collections.singletonList(-1L));
        // デフォルトの TimeoutException の代わりにカスタムビジネス例外を使用することもできます。
        // future.completeExceptionally(new RuntimeException(waitMax));
    }
}