Realtime Compute for Apache Flink:UDSF de Java

Funcionamiento

Una UDSF extiende la clase ScalarFunction de Apache Flink e implementa uno o más métodos eval(). Flink llama a eval() una vez por cada fila de entrada y usa el valor devuelto como el escalar de salida.

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

Requisitos del método eval():

• Debe declararse como public

• Admite sobrecarga de métodos: defina varias firmas de eval() para distintos tipos de entrada

• Admite argumentos variables (por ejemplo, eval(Integer...))

• Use tipos primitivos envueltos (por ejemplo, Integer en lugar de int) para gestionar correctamente las entradas NULL

Requisitos previos

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

• IntelliJ IDEA instalado

• Apache Maven instalado

• Acceso a un espacio de trabajo de Realtime Compute for Apache Flink

Funciones escalares asíncronas (AsyncScalarFunction)

A diferencia de una función escalar común, una función escalar asíncrona personalizada debe extender AsyncScalarFunction y devolver los resultados mediante un CompletableFuture.

Utilice una función escalar asíncrona cuando esta ejecute operaciones que impliquen esperas de E/S, como consultar una base de datos externa o llamar a una interfaz HTTP. La función procesa varios registros de forma concurrente, lo cual reduce el impacto de las esperas en el rendimiento del trabajo. Para operaciones que no dependen de servicios externos, como el recorte de cadenas o los cálculos numéricos, suele ser suficiente una función escalar síncrona.

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

Desarrollar una UDSF

Flink proporciona ejemplos de UDF con el entorno de desarrollo preconfigurado. Estos ejemplos abarcan UDSF, funciones de agregación definidas por el usuario (UDAF) y funciones con valores de tabla definidas por el usuario (UDTF).

1. Descargue y descomprima el ejemplo ASI_UDX_Demo en su computadora local. Tras la descompresión, se creará la carpeta ASI_UDX-main con la siguiente estructura:

   • pom.xml: archivo de configuración del proyecto de Maven. Define las coordenadas, dependencias, reglas de compilación y metadatos relacionados del proyecto.

   • \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDF.java: implementación de muestra en Java de la UDSF.

   ASI_UDX_Demo está alojado en un sitio web de terceros. Es posible que experimente errores o retrasos al acceder a él.

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

3. Abra \ASI_UDX-main\src\main\java\ASI_UDF\ASI_UDF.java y actualice el método eval() con su lógica personalizada. La implementación de muestra extrae caracteres desde la posición begin hasta la posición end de cada cadena de entrada:

   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. Abra \ASI_UDX-main\pom.xml y configure las dependencias de Maven correspondientes a su versión de Flink. El siguiente ejemplo muestra las dependencias principales del paquete JAR para Flink 1.11. Si su UDSF no depende de paquetes JAR adicionales, omita este paso.

   <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. En el directorio que contiene el archivo pom.xml, ejecute el siguiente comando para compilar el proyecto:

   mvn package -Dcheckstyle.skip

   Una vez finalizada la compilación, se generará el archivo ASI_UDX-1.0-SNAPSHOT.jar en el directorio \ASI_UDX-main\target\.

Registrar una UDSF

Para registrar el paquete JAR como una UDSF, consulte Manage user-defined functions (UDFs).

Usar una UDSF

Después de registrarla, llame a la UDSF en un trabajo de Flink SQL.

1. Cree un trabajo de Flink SQL. Para obtener instrucciones, consulte Job development map. El siguiente código SQL de ejemplo extrae los caracteres de la segunda a la cuarta posición de la cadena en el campo a de ASI_UDSF_Source y escribe los resultados en ASI_UDSF_Sink:

   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. En la página Operation Center > Job O&M, busque su trabajo y haga clic en Start dentro de la columna Actions. Una vez iniciado el trabajo, los caracteres de la segunda a la cuarta posición del campo a de cada fila en ASI_UDSF_Source se insertarán en ASI_UDSF_Sink.