Realtime Compute for Apache Flink:Java UDSF'leri

Nasıl çalışır

UDSF, Apache Flink'in ScalarFunction sınıfını genişletir ve bir veya daha fazla eval() yöntemi uygular. Flink, her giriş satırı için eval() yöntemini bir kez çağırır ve dönüş değerini çıkış skaleri olarak kullanır.

public class SubstringFunction extends ScalarFunction {
    public String eval(String s, Integer begin, Integer end) {
        return s.substring(begin, end);
    }
}

eval() yöntemi gereksinimleri:

• public olarak bildirilmelidir

• Yöntem aşırı yüklemeyi destekler — farklı giriş türleri için birden fazla eval() imzası tanımlayabilirsiniz

• Değişken sayıda bağımsız değişkeni (varargs) destekler (örneğin, eval(Integer...))

• NULL girişlerini doğru işlemek için kutulanmış ilkel türler kullanın (örneğin, int yerine Integer)

Ön koşullar

Başlamadan önce şunlara sahip olduğunuzdan emin olun:

• IntelliJ IDEA yüklü olmalı

• Apache Maven yüklü olmalı

• Realtime Compute for Apache Flink çalışma alanına erişiminiz olmalı

Eşzamansız skaler işlevler (AsyncScalarFunction)

Normal bir skaler işlevin aksine, özel bir eşzamansız skaler işlev AsyncScalarFunction sınıfını genişletmeli ve sonuçları bir CompletableFuture aracılığıyla döndürmelidir.

İşlev, harici bir veritabanını sorgulamak veya bir HTTP arayüzünü çağırmak gibi G/Ç bekleme süresi içeren işlemler yürüttüğünde eşzamansız skaler işlev kullanın. İşlev, birden çok kaydı eşzamanlı olarak işleyerek bekleme süresinin iş aktarım hızı üzerindeki etkisini azaltır. Dize kısaltma ve sayısal hesaplama gibi harici hizmetler için bekleme gerektirmeyen işlemler için genellikle eşzamanlı bir skaler işlev yeterlidir.

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

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;

/**
 * Queries a product name through an asynchronous HTTP interface.
 *
 * <p>Interface contract: 200 returns a UTF-8 product name, and 404 indicates that the product does not exist.
 * This example suits read-only queries that return short text. Other status codes are treated as invocation failures and handed over to the engine.
 */
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() {
        // Product information may change. The same ID may return different results at different times.
        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);
        }
    }
}

UDSF geliştirme

Flink, geliştirme ortamını önceden yapılandıran UDF örnekleri sunar. Örnekler UDSF'leri, kullanıcı tanımlı toplama işlevlerini (UDAF) ve kullanıcı tanımlı tablo değerli işlevleri (UDTF) kapsar.

1. ASI_UDX_Demo örneğini indirin ve yerel makinenizde açın. Açma işleminden sonra aşağıdaki yapıya sahip ASI_UDX-main klasörü oluşturulur:

   • pom.xml: Maven proje yapılandırma dosyası. Projenin koordinatlarını, bağımlılıklarını, derleme kurallarını ve ilgili meta verilerini tanımlar.

   • \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDF.java: UDSF'nin örnek Java uygulaması.

   ASI_UDX_Demo üçüncü taraf bir web sitesinde barındırılmaktadır. Erişim hataları veya gecikmeler yaşayabilirsiniz.

2. IntelliJ IDEA'da File > Open seçeneğine tıklayın ve ASI_UDX-main klasörünü seçin.

3. \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDF.java dosyasını açın ve eval() yöntemini özel mantığınızla güncelleyin. Örnek uygulama, her giriş dizesinin begin konumundan end konumuna kadar karakterleri çıkarır:

   package ASI_UDF;
   
   import org.apache.flink.table.functions.ScalarFunction;
   
   public class ASI_UDF extends ScalarFunction {
       public String eval(String s, Integer begin, Integer end) {
           return s.substring(begin, end);
       }
   }

4. \ASI_UDX-main\pom.xml dosyasını açın ve Flink sürümünüz için Maven bağımlılıklarını yapılandırın. Aşağıdaki örnek, Flink 1.11 için ana JAR paketi bağımlılıklarını gösterir. UDSF'niz ek JAR paketlerine bağımlı değilse bu adımı atlayın.

   <dependencies>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-streaming-java_2.12</artifactId>
           <version>1.11.0</version>
           <!--<scope>provided</scope>-->
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-table</artifactId>
           <version>1.11.0</version>
           <type>pom</type>
           <!--<scope>provided</scope>-->
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-core</artifactId>
           <version>1.11.0</version>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-table-common</artifactId>
           <version>1.11.0</version>
       </dependency>
   </dependencies>

5. pom.xml dosyasını içeren dizinde projeyi paketlemek için aşağıdaki komutu çalıştırın:

   mvn package -Dcheckstyle.skip

   Derleme tamamlandıktan sonra \ASI_UDX-main\target\ dizininde ASI_UDX-1.0-SNAPSHOT.jar dosyası oluşturulur.

UDSF kaydetme

JAR paketini UDSF olarak kaydetmek için bkz. Kullanıcı tanımlı işlevleri (UDF) yönetme.

UDSF kullanma

Kayıttan sonra UDSF'yi bir Flink SQL işinde çağırın.

1. Bir Flink SQL işi oluşturun. Rehberlik için bkz. İş geliştirme haritası. Aşağıdaki örnek SQL, ASI_UDSF_Source tablosunun a alanındaki dizenin ikinci konumundan dördüncü konumuna kadar karakterleri çıkarır ve sonuçları ASI_UDSF_Sink tablosuna yazar:

   CREATE TEMPORARY TABLE ASI_UDSF_Source (
     a VARCHAR,
     b INT,
     c INT
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDSF_Sink (
     a VARCHAR
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDSF_Sink
   SELECT ASI_UDSF(a, 2, 4)
   FROM ASI_UDSF_Source;

2. Operation Center > Job O&M sayfasında işinizi bulun ve Actions sütununda Start seçeneğine tıklayın. İş başladıktan sonra, ASI_UDSF_Source tablosundaki her satırın a alanının ikinci konumundan dördüncü konumuna kadar karakterler ASI_UDSF_Sink tablosuna eklenir.