Realtime Compute for Apache Flink:Java UDSFs

Parameter konfigurasi

Parameter berikut mengontrol perilaku waktu proses fungsi skalar asinkron. Atur parameter ini sebelum Anda mengirimkan kueri atau menghasilkan rencana eksekusi.

Konfigurasi output tidak terurut

Secara default, fungsi asinkron dijalankan secara konkuren, tetapi hasil yang telah selesai untuk catatan berikutnya tetap dipancarkan hanya setelah catatan sebelumnya selesai. Ketika latensi permintaan eksternal sangat bervariasi dan bisnis Anda dapat menerima hasil yang tidak terurut, aktifkan output tidak terurut untuk mengurangi waktu tunggu akibat permintaan lambat.

Tambahkan konfigurasi berikut sebelum menjalankan kueri:

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

Contoh fungsi skalar asinkron kustom

Contoh berikut mengkueri nama produk melalui antarmuka HTTP layanan produk. Fungsi ini membuat dan menggunakan kembali HttpClient dalam open, memanggil sendAsync() dalam eval untuk mengirim permintaan, dan mengembalikan hasil melalui callback penyelesaian. sendAsync() mengembalikan CompletableFuture, yang cocok untuk menghubungkan panggilan HTTP ke fungsi skalar asinkron. Untuk informasi lebih lanjut, lihat JDK HttpClient documentation.

Contoh ini menggunakan timeout koneksi 3 detik dan timeout permintaan 5 detik, yang sesuai untuk kueri nama produk dengan respons berukuran kecil. Sesuaikan nilai-nilai tersebut berdasarkan waktu respons aktual antarmuka Anda.

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;

/**
 * Mengkueri nama produk melalui antarmuka HTTP asinkron.
 *
 * <p>Kontrak antarmuka: kode status 200 mengembalikan nama produk dalam UTF-8, dan 404 menunjukkan bahwa produk tidak ada.
 * Contoh ini cocok untuk kueri read-only yang mengembalikan teks pendek. Kode status lain dianggap sebagai kegagalan pemanggilan dan diserahkan ke mesin.
 */
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("Konfigurasikan 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 layanan harus berupa URL dasar HTTP(S) yang diakhiri dengan garis miring (/).");
        }
        client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
    }

    @Override
    public boolean isDeterministic() {
        // Informasi produk dapat berubah. ID yang sama mungkin mengembalikan hasil berbeda pada waktu berbeda.
        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(
                                                "Gagal mengkueri produk. Kode status HTTP: " + response.statusCode()));
                            })
                    .whenComplete(
                            (name, error) -> {
                                if (error != null) {
                                    result.completeExceptionally(error);
                                } else {
                                    result.complete(name);
                                }
                            });
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }
    }
}

Perhatikan hal-hal berikut selama pengembangan:

• Panggil result.complete(value) untuk mengembalikan hasil. Untuk mengembalikan NULL SQL, panggil result.complete(null).

• Saat pemanggilan gagal, panggil result.completeExceptionally(exception) untuk mengembalikan pengecualian. Mesin kemudian akan menanganinya berdasarkan strategi percobaan ulang.

• eval langsung mengembalikan setelah permintaan dikirim. Kirim hasil melalui callback penyelesaian sendAsync(). Jangan memanggil metode blocking seperti send(), get(), atau join(), atau lakukan komputasi yang memakan waktu dalam eval atau dalam callback.

• Setiap instans fungsi menginisialisasi dan menggunakan kembali klien dalam metode open untuk menghindari pembuatan klien untuk setiap catatan. HttpClient di JDK 11 dan JDK 17 tidak memiliki metode close() publik, dan sumber daya internalnya dikelola oleh JDK. Jika Anda menggunakan klien asinkron lain yang menyediakan metode close, lepaskan klien tersebut dalam metode close fungsi.

• Informasi produk berubah seiring waktu, sehingga contoh ini mengatur isDeterministic() menjadi false. Saat aliran pembaruan terlibat, tangani hasil non-deterministik berdasarkan semantik kueri.