Realtime Compute for Apache Flink:Java UDTF'leri

Ön koşullar

Başlamadan önce aşağıdakilere sahip olduğunuzdan emin olun:

• Yerel makinenizde kurulu olan Apache Maven

• Yerel makinenizde kurulu olan IntelliJ IDEA

• Realtime Compute for Apache Flink'te yapılandırılmış bir Flink SQL iş ortamı

Bir UDTF Geliştirme

1. ASI_UDX_Demo örnek projesini yerel makinenize indirin ve sıkıştırılmış dosyayı açın. Sıkıştırma işleminden sonra aşağıdaki dosyaları içeren bir ASI_UDX-main klasörü oluşturulur:

   • pom.xml: Projenin koordinatlarını, bağımlılıklarını ve derleme ayarlarını açıklayan Maven proje yapılandırma dosyası.

   • \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDTF.java: UDTF için örnek Java kodu.

2. IntelliJ IDEA'da File > Open seçeneklerine tıklayın ve sıkıştırması açılan ASI_UDX-main klasörünü açın.

3. \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDTF.java dosyasını açın ve UDTF mantığınızı uygulayın. Bir UDTF, çıktı türünün T olduğu TableFunction<T> sınıfından türetilir. eval() yöntemi void döndürmelidir; bir return deyimi kullanmak yerine çıktı satırlarını yaymak için collect() yöntemini çağırın. Bu durum, return kullanan skaler işlevlerden farklıdır. Aşağıdaki örnek, bir dizeyi dikey çubuk (|) ayırıcısını kullanarak iki sütuna böler:

   package ASI_UDTF;
   
   import org.apache.flink.api.java.tuple.Tuple2;
   import org.apache.flink.table.functions.TableFunction;
   
   public class ASI_UDTF extends TableFunction<Tuple2<String, String>> {
       public void eval(String str) {
           String[] split = str.split("\\|");
           String name = split[0];
           String place = split[1];
           // emit one row with two columns
           collect(Tuple2.of(name, place));
       }
   }

   Çıktı türlerini bildirme Flink, bir UDTF'nin çıktı şemasını bildirmek için iki yaklaşımı destekler:

   • @FunctionHint ve @DataTypeHint ile Satır türü (önerilen): Alan adlarını ve türlerini açıkça bildirerek çıktı şemasının hem Flink hem de okuyucular için net olmasını sağlar.

     @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
     public static class SplitFunction extends TableFunction<Row> {
         public void eval(String str) {
             for (String s : str.split(" ")) {
                 // use collect(...) to emit a row
                 collect(Row.of(s, s.length()));
             }
         }
     }

   • Tuple türü (tür genel türlerden çıkarılır): Yazılması daha basittir ancak sütun adları açık değildir.

     TableFunction<Tuple2<String, Integer>>

   Not

   TableFunction tarafından desteklenen veri türleri ve tür çıkarsama mekanizması, ana Flink sürümleri arasında değişiklik gösterir. Yukarıdaki örnekler Flink 1.15 belgelerini takip eder (Veri türleri ve Tür çıkarsama). Ververica Runtime (VVR) sürümünüzle eşleşen Flink belgelerini kontrol edin. Flink sürümünüzü bulmak için bkz. Depolama yönetimi ve işlemleri.

4. ASI_UDX-main klasöründeki pom.xml dosyasını açın ve bağımlılıklarınızı yapılandırın. Örnek proje, Flink 1.11 için aşağıdaki Maven bağımlılıklarını içerir:

   <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>

   UDTF'nizin listelenenlerin dışında ek bağımlılığı yoksa bir sonraki adıma geçin. Aksi takdirde, gerekli bağımlılık girişlerini pom.xml dosyasına ekleyin.

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 başarılı olursa \ASI_UDX-main\target\ dizininde ASI_UDX-1.0-SNAPSHOT.jar dosyası oluşturulur. UDTF geliştirme işlemi tamamlanmıştır.

Eşzamansız tablo işlevleri (AsyncTableFunction)

Eşzamansız bir tablo işlevi (AsyncTableFunction), Flink SQL veya Tablo API'sinde harici hizmetlere eşzamansız olarak erişir ve her çağrı için sıfır, bir veya birden çok satır döndürür. Bu, bir tür kullanıcı tanımlı tablo değerli işlevdir (UDTF). Tipik senaryolar arasında ürün kimliğine göre birden çok etiketi sorgulama veya bir arama koşulu için birden çok eşleşen kaydı döndürme yer alır. Eşzamansız bir tablo işlevi, SQL'de LATERAL TABLE veya Tablo API'sinde joinLateral tarafından çağrılan ve eşzamansız bir ilişkilendirme işlemi oluşturan bir kullanıcı tanımlı işlev olabilir. Ayrıca, eşzamansız arama birleştirmesi için sorgu uygulaması olarak bir boyut tablosu bağlayıcısı tarafından sağlanabilir.

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;

/**
 * Queries product tags through an asynchronous HTTP interface.
 *
 * <p>Interface contract: 200 returns UTF-8 text with one tag per line; 204 and 404 indicate no tags.
 * This example suits read-only queries that return short text. Other status codes are treated as invocation failures and handed over to the engine.
 */
@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("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<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(
                                                "Failed to query the product. HTTP status code: " + response.statusCode()));
                            })
                    .whenComplete(
                            (rows, error) -> {
                                if (error != null) {
                                    result.completeExceptionally(error);
                                } else {
                                    result.complete(rows);
                                }
                            });
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }
    }
}

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) {
        // ...same logic as the preceding development example: dispatch the asynchronous task,
        // and then complete `future` in the callback thread.
    }

    // The parameter list must align with eval(): the same CompletableFuture generic type,
    // and the same types and order for the remaining parameters (Integer waitMax).
    // The method body must complete `future` synchronously. Do not start another asynchronous call
    // or a new thread to complete it asynchronously. When this callback is triggered,
    // the operator has already given up on this record.
    public void timeout(CompletableFuture<Collection<Long>> future, Integer waitMax) {
        future.complete(Collections.singletonList(-1L));
        // You can also use a custom business exception instead of the default TimeoutException:
        // future.completeExceptionally(new RuntimeException(waitMax));
    }
}

Bir UDTF Kaydetme

Bir UDTF'nin nasıl kaydedileceğine ilişkin talimatlar için bkz. Kullanıcı tanımlı işlevleri (UDF'ler) yönetme.

Bir UDTF Kullanma

UDTF'yi kaydettikten sonra bunu bir Flink SQL işinde kullanın.

1. Bir Flink SQL işi geliştirin. Daha fazla bilgi için bkz. İş geliştirme haritası. Aşağıdaki SQL, ASI_UDTF_Source içindeki message alanını | ayırıcısını kullanarak iki sütuna böler ve ardından sonuçları ASI_UDTF_Sink içine ekler. LATERAL TABLE sözdizimi, her girdi satırını UDTF tarafından yayılan satırlarla çapraz birleştirir.

   CREATE TEMPORARY TABLE ASI_UDTF_Source (
     `message` VARCHAR
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDTF_Sink (
     name  VARCHAR,
     place VARCHAR
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDTF_Sink
   SELECT name, place
   FROM ASI_UDTF_Source, LATERAL TABLE(ASI_UDTF(`message`)) AS T(name, place);

2. Operation Center > Job O&M sayfasında hedef işi bulun ve Actions sütunundaki Start seçeneğine tıklayın. İş başladıktan sonra ASI_UDTF_Source içindeki message alanındaki dizeler | ayırıcısına göre bölünür ve elde edilen sütunlar ASI_UDTF_Sink içine eklenir.

Sonraki adımlar

• Kullanıcı tanımlı işlevleri (UDF'ler) yönetme

• İş geliştirme haritası

• Depolama yönetimi ve işlemleri