Realtime Compute for Apache Flink:UDTF en Java

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

• Apache Maven instalado en su computadora local

• IntelliJ IDEA instalado en su computadora local

• Un entorno de trabajo de Flink SQL configurado en Realtime Compute for Apache Flink

Desarrollar una UDTF

1. Descargue y descomprima el proyecto de ejemplo ASI_UDX_Demo en su computadora local. Tras la descompresión, se crea la carpeta ASI_UDX-main con los siguientes archivos:

   • pom.xml: archivo de configuración del proyecto de Maven que describe las coordenadas, dependencias y ajustes de compilación del proyecto.

   • \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDTF.java: código Java de ejemplo de la UDTF.

2. En IntelliJ IDEA, haga clic en File > Open y abra la carpeta descomprimida ASI_UDX-main.

3. Abra \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDTF.java e implemente la lógica de su UDTF. Una UDTF extiende TableFunction<T>, donde T es el tipo de salida. El método eval() debe devolver void. En lugar de usar una instrucción return, llame a collect() para emitir las filas de salida. Este comportamiento difiere de las funciones escalares, que utilizan return. El siguiente ejemplo divide una cadena en dos columnas mediante una barra vertical (|) como delimitador:

   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));
       }
   }

   Declaración de tipos de salida Flink admite dos enfoques para declarar el esquema de salida de una UDTF:

   • Tipo Row con @FunctionHint y @DataTypeHint (recomendado): declara de forma explícita los nombres y tipos de los campos. De esta manera, el esquema de salida queda claro tanto para Flink como para quienes lean el código.

     @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()));
             }
         }
     }

   • Tipo Tuple (el tipo se infiere a partir de los genéricos): resulta más sencillo de escribir, pero los nombres de las columnas no son explícitos.

     TableFunction<Tuple2<String, Integer>>

   Nota

   Los tipos de datos y el mecanismo de inferencia de tipos compatibles con TableFunction varían entre las versiones principales de Flink. Los ejemplos anteriores siguen la documentación de Flink 1.15 (Tipos de datos e Inferencia de tipos). Revise la documentación de Flink correspondiente a su versión de Ververica Runtime (VVR). Para conocer su versión de Flink, consulte Storage management and operations.

4. Abra pom.xml en la carpeta ASI_UDX-main y configure sus dependencias. El proyecto de ejemplo incluye las siguientes dependencias de Maven para Flink 1.11:

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

   Si su UDTF no requiere dependencias adicionales a las listadas, pase al siguiente paso. De lo contrario, agregue las entradas de dependencia obligatorias en pom.xml.

5. En el directorio que contiene pom.xml, ejecute el siguiente comando para compilar el proyecto:

   mvn package -Dcheckstyle.skip

   Si la compilación es exitosa, se genera ASI_UDX-1.0-SNAPSHOT.jar en el directorio \ASI_UDX-main\target\. Con esto concluye el desarrollo de la UDTF.

Funciones de tabla asíncronas (AsyncTableFunction)

Una función de tabla asíncrona (AsyncTableFunction) accede a servicios externos de forma asíncrona en Flink SQL o en la Table API. Devuelve cero, una o varias filas por cada llamada y constituye un tipo de función con valores de tabla definida por el usuario (UDTF). Algunos escenarios típicos incluyen consultar varias etiquetas por identificador de producto o devolver múltiples registros coincidentes para una condición de búsqueda. Esta función puede invocarse mediante LATERAL TABLE en SQL o joinLateral en la Table API para formar una operación de correlación asíncrona. También puede proporcionarla un conector de tabla de dimensiones como implementación de consulta de una unión de búsqueda asíncrona.

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));
    }
}

Registrar una UDTF

Para obtener instrucciones sobre cómo registrar una UDTF, consulte Manage user-defined functions (UDFs).

Usar una UDTF

Después de registrar la UDTF, utilícela en un trabajo de Flink SQL.

1. Desarrolle un trabajo de Flink SQL. Para obtener más información, consulte Job development map. El siguiente código SQL divide el campo message de ASI_UDTF_Source en dos columnas mediante el delimitador |. Luego inserta los resultados en ASI_UDTF_Sink. La sintaxis LATERAL TABLE realiza un cross join de cada fila de entrada con las filas que emite la UDTF.

   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. En la página Operation Center > Job O&M, busque el trabajo deseado y haga clic en Start en la columna Actions. Una vez iniciado el trabajo, las cadenas del campo message de ASI_UDTF_Source se dividen con el delimitador |. Las columnas resultantes se insertan en ASI_UDTF_Sink.

Pasos siguientes

• Manage user-defined functions (UDFs)

• Job development map

• Storage management and operations