Realtime Compute for Apache Flink:Java UDSF

設定パラメーター

以下のパラメーターは、非同期スカラー関数のランタイム動作を制御します。クエリを送信したり、実行計画を生成したりする前に、これらのパラメーターを設定してください。

順序不同の出力の設定

デフォルトでは、非同期関数は同時に実行されますが、後続のレコードの完了した結果は、先行するレコードが完了した後にのみ出力されます。外部リクエストのレイテンシーが大幅に異なり、ビジネス上、順序不同の結果が許容される場合は、順序不同の出力を有効にすることで、遅いリクエストによる待機を減らすことができます。

クエリを実行する前に、次の設定を追加します。

SET 'table.exec.async-scalar.output-mode' = 'ALLOW_UNORDERED';

カスタム非同期スカラー関数の例

次の例では、製品サービスの HTTP インターフェースを介して製品名を照会します。この関数は、open で HttpClient を作成して再利用し、eval で sendAsync() を呼び出してリクエストを送信し、完了コールバックを介して結果を返します。sendAsync() は CompletableFuture を返すため、HTTP 呼び出しを非同期スカラー関数に接続するのに適しています。詳細については、「JDK HttpClient documentation」をご参照ください。

この例では、接続タイムアウトを 3 秒、リクエストタイムアウトを 5 秒に設定しており、これはレスポンスボディが小さい製品名のクエリに適しています。実際のインターフェースの応答時間に基づいて値を調整してください。

package com.example.udf;

import org.apache.flink.table.functions.AsyncScalarFunction;
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.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

/**
 * 非同期 HTTP インターフェースを介して製品名を照会します。
 *
 * <p>インターフェースの規約： 200 は UTF-8 の製品名を返し、404 は製品が存在しないことを示します。
 * この例は、短いテキストを返す読み取り専用のクエリに適しています。他のステータスコードは呼び出しの失敗として扱われ、エンジンに渡されます。
 */
public class AsyncProductName extends AsyncScalarFunction {

    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("Configure 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("The service URL must be an HTTP(S) base URL that ends with a forward slash (/).");
        }
        client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
    }

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

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

        try {
            HttpRequest request =
                    HttpRequest.newBuilder(
                                    serviceBaseUri.resolve("products/" + productId + "/name"))
                            .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();
                                }
                                if (response.statusCode() == 404) {
                                    return null;
                                }
                                throw new CompletionException(
                                        new IOException(
                                                "Failed to query the product. HTTP status code: " + response.statusCode()));
                            })
                    .whenComplete(
                            (name, error) -> {
                                if (error != null) {
                                    result.completeExceptionally(error);
                                } else {
                                    result.complete(name);
                                }
                            });
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }
    }
}

開発中は、以下の項目に注意してください。

• 結果を返すには、result.complete(value) を呼び出します。SQL の NULL を返すには、result.complete(null) を呼び出します。

• 呼び出しが失敗した場合は、result.completeExceptionally(exception) を呼び出して例外を返します。エンジンはリトライ戦略に基づいてそれを処理します。

• eval はリクエストが送信されるとすぐに戻ります。sendAsync() の完了コールバックを介して結果を渡します。ブロッキングメソッドである send()、get()、または join() を呼び出したり、eval またはコールバック内で時間のかかる計算を実行したりしないでください。

• 各関数インスタンスは、open でクライアントを初期化して再利用することで、レコードごとにクライアントを作成するのを回避します。JDK 11 および JDK 17 の HttpClient には public な close() メソッドがなく、その内部リソースは JDK によって管理されます。close メソッドを提供する別の非同期クライアントを使用する場合は、関数の close メソッドでクライアントを解放してください。

• 製品情報は時間とともに変化するため、この例では isDeterministic() を false に設定しています。更新ストリームを扱う場合は、クエリのセマンティクスに基づいて非決定的な結果を処理してください。