# User-defined Functions in Confluent Cloud for Apache Flink
Confluent Cloud for Apache Flink® supports user-defined functions (UDFs), which are extension points
for running custom logic that you can’t express in the system-provided
Flink SQL [queries](../reference/queries/overview.md#flink-sql-queries) or with the
[Table API](../reference/table-api.md#flink-table-api).
You can implement user-defined functions in Java or Python, and you can use
third-party libraries within a UDF. Confluent Cloud for Apache Flink supports scalar functions (UDFs),
which map scalar values to a new scalar value, and table functions (UDTFs),
which map multiple scalar values to multiple output rows (Java only).
- **Create an example UDF:** [Create a User Defined Function](../how-to-guides/create-udf.md#flink-sql-create-udf)
- **Add logging to your UDFs:** [Enable Logging in a User Defined Function](../how-to-guides/enable-udf-logging.md#flink-sql-enable-udf-logging)
- **Availability:** [UDF regional availability](#flink-sql-udfs-availability)
- **Limitations:** [UDF limitations](#flink-sql-udfs-limitations)
- **Example code:** [Flink UDF Java Examples](https://github.com/confluentinc/flink-udf-java-examples)
and [Flink UDF Python Examples](https://github.com/confluentinc/flink-udf-python-examples)
- **Permanent and in-line UDFs:** [Permanent and in-line UDFs](#flink-sql-udfs-permanent-and-inline-udfs)
- **External connections:** [External connectivity](#flink-sql-udfs-external-connectivity)
## Artifacts
Artifacts are JAR files for Java, and ZIP files for Python, that contain
user-defined functions and all of the required dependencies. You upload
artifacts to Confluent Cloud and scope them to a specific region in a Confluent Cloud
environment. To use them with UDFs, artifacts must follow a few common
implementation principles, which the following sections describe.
To use a UDF, you must register one or more functions that reference the
artifact.
## Functions
Functions are SQL objects that reference a class in an artifact and that
you can use in any SQL statement or Table API program.
After you upload an artifact, you register a function by using the
[CREATE FUNCTION](../reference/statements/create-function.md#flink-sql-create-function) statement.
After you register a function, you can invoke it from any SQL statement or
Table API program.
The following example shows how to register a `TShirtSizingIsSmaller`
function and invoke it in a SQL statement.
```sql
-- Register the function.
CREATE FUNCTION is_smaller
AS 'com.example.my.TShirtSizingIsSmaller'
USING JAR 'confluent-artifact://';
-- Invoke the function.
SELECT IS_SMALLER ('L', 'M');
```
To build and upload a UDF to Confluent Cloud for Apache Flink for use in Flink SQL or the Table API,
see [Create a UDF](../how-to-guides/create-udf.md#flink-sql-create-udf).
## RBAC
To upload artifacts, register functions, and invoke functions, you must have the
FlinkDeveloper role or higher. For more information, see
[Grant Role-Based Access](../operate-and-deploy/flink-rbac.md#flink-rbac).
## Shared responsibility
Confluent supports the UDF infrastructure in Confluent Cloud only. Troubleshooting
custom UDF issues for functions you build, or that others provide to you, is
your responsibility. The following provides additional details about shared
support responsibilities.
* **Customer Managed**: You are responsible for function logic. Confluent
does not provide any support for debugging services and features within UDFs.
* **Confluent Managed**: Confluent is responsible for managing the Flink
services and custom compute platform, and provides support for these.
## Scalar functions
A user-defined scalar function maps zero, one, or multiple scalar values to a
new scalar value. You can use any data type listed in [Data Types](../reference/datatypes.md#flink-sql-datatypes)
as a parameter or return type of an evaluation method.
To define a scalar function, extend the `ScalarFunction` base class in
`org.apache.flink.table.functions` and implement one or more evaluation
methods named `eval(...)`.
The following code example shows how to define your own hash code function in
Java.
```java
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.ScalarFunction;
import static org.apache.flink.table.api.Expressions.*;
public static class HashFunction extends ScalarFunction {
// take any data type and return INT
public int eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
return o.hashCode();
}
}
```
The following example shows how to call the `HashFunction` UDF in a
Flink SQL statement.
```sql
SELECT HashFunction(myField) FROM MyTable;
```
To build and upload a UDF to Confluent Cloud for Apache Flink for use in Flink SQL, see
[Create a User Defined Function](../how-to-guides/create-udf.md#flink-sql-create-udf).
### Vectorized scalar functions
A *vectorized* scalar function is a Python scalar UDF that processes a batch of
rows at once instead of one row at a time. Instead of receiving individual
scalar values, the function receives each argument as a `pandas.Series` and
returns a `pandas.Series` of the same length. Under the hood, Flink transfers
the batch to Python in [Apache Arrow](https://arrow.apache.org/) columnar
format, so Flink amortizes the per-row overhead of the Python interpreter
across the whole batch.
Use a vectorized function when you process high-throughput streams and you can
express your logic as element-wise operations over a batch, for example with
[NumPy](https://numpy.org/) or [pandas](https://pandas.pydata.org/). For
this kind of workload, a vectorized function can be significantly faster than
the equivalent row-by-row scalar function.
To make a Python scalar function vectorized, add `func_type="pandas"` to the
`udf` definition. No other changes are required: the function signature,
artifact packaging, and `CREATE FUNCTION` registration are the same as for any
other Python UDF. The following example computes the great-circle distance
between two coordinates over a batch by using NumPy.
```python
import numpy as np
import pandas as pd
from pyflink.table import DataTypes
from pyflink.table.udf import udf
_EARTH_RADIUS_KM = 6371.0
def _great_circle_km_vec(lat1, lon1, lat2, lon2):
lat1_r = np.radians(lat1)
lat2_r = np.radians(lat2)
dlat = np.radians(lat2 - lat1)
dlon = np.radians(lon2 - lon1)
a = (np.sin(dlat / 2) ** 2 +
np.cos(lat1_r) * np.cos(lat2_r) * np.sin(dlon / 2) ** 2)
a = np.clip(a, 0.0, 1.0)
return pd.Series(_EARTH_RADIUS_KM * 2 * np.arcsin(np.sqrt(a)))
great_circle_km_vec = udf(
_great_circle_km_vec,
input_types=[DataTypes.DOUBLE()] * 4,
result_type=DataTypes.DOUBLE(),
func_type="pandas",
)
```
When you implement a vectorized function, keep the following in mind:
- The returned `pandas.Series` must have the same length as the input batch.
- SQL `NULL` values arrive as `NaN` in the input `Series`, and Flink
converts `NaN` values in the returned `Series` back to SQL `NULL`.
- `pandas` and `numpy` are available through the Flink Python runtime, so you
don’t need to add them as explicit dependencies.
To build, upload, and register a Python UDF, see
[Create a User-Defined Function](../how-to-guides/create-udf.md#flink-sql-create-udf).
## Permanent and in-line UDFs
Starting with Confluent Table API plugin version 2.1-8, you can simplify the
process of creating and managing UDFs.
- **Permanent UDFs**: Flink registers them automatically, and you can use
them in any Flink SQL or Table API program. The Table API creates a
temporary JAR file containing all transitive classes required to run the
function, uploads it to Confluent Cloud, and registers the function using the
previously uploaded artifact.
- **In-line UDFs**: You define and use them in the same Table API program.
In-line functions must be serializable by class name, so use a top-level
class or a static nested class, not an anonymous inner class.
The following example shows how to create and call a permanent UDF and an
in-line UDF.
For the full code listing, see
[Example_09_Functions.java](https://github.com/confluentinc/flink-table-api-java-examples/blob/master/src/main/java/io/confluent/flink/examples/table/Example_09_Functions.java)
in the
[flink-table-api-java-examples](https://github.com/confluentinc/flink-table-api-java-examples/)
repository.
### Implement a permanent and in-line UDF
```java
package io.confluent.flink.examples.table;
import io.confluent.flink.plugin.ConfluentSettings;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import java.util.List;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.array;
import static org.apache.flink.table.api.Expressions.call;
import static org.apache.flink.table.api.Expressions.row;
/**
* A table program example showing how to use User-Defined Functions
* (UDFs) in the Flink Table API.
*
*
The Flink Table API simplifies the process of creating and managing UDFs.
*
*
*
It helps creating a JAR file containing all required dependencies for a given UDF.
*
Uploads the JAR to Confluent artifact API.
*
Creates SQL functions for given artifacts.
*
*/
public class Example_09_Functions {
// Fill this with an environment you have write access to
static final String TARGET_CATALOG = "";
// Fill this with a Kafka cluster you have write access to
static final String TARGET_DATABASE = "";
// All logic is defined in a main() method. It can run both in an IDE or CI/CD system.
public static void main(String[] args) {
// Setup connection properties to Confluent Cloud
EnvironmentSettings settings = ConfluentSettings.fromResource("/cloud.properties");
// Initialize the session context to get started
TableEnvironment env = TableEnvironment.create(settings);
// Set default catalog and database
env.useCatalog(TARGET_CATALOG);
env.useDatabase(TARGET_DATABASE);
System.out.println("Registering a scalar function...");
// The Table API underneath creates a temporary JAR file containing all transitive classes
// required to run the function, uploads it to Confluent Cloud, and registers the function
// using the previously uploaded artifact.
env.createFunction("CustomTax", CustomTax.class, true);
// As of now, Scalar and Table functions are supported.
System.out.println("Registering a table function...");
env.createFunction("Explode", Explode.class, true);
// Once registered, the functions can be used in Table API and SQL queries.
System.out.println("Executing registered UDFs...");
env.fromValues(row("Apple", "USA", 2), row("Apple", "EU", 3))
.select(
$("f0").as("product"),
$("f1").as("location"),
$("f2").times(call("CustomTax", $("f1"))).as("tax"))
.execute()
.print();
env.fromValues(
row(1L, "Ann", array("Apples", "Bananas")),
row(2L, "Peter", array("Apples", "Pears")))
.joinLateral(call("Explode", $("f2")).as("fruit"))
.select($("f0").as("id"), $("f1").as("name"), $("fruit"))
.execute()
.print();
// Instead of registering functions permanently, you can embed UDFs directly into queries
// without registering them first. This will upload all the functions of the query as a
// single artifact to Confluent Cloud. Moreover, the functions lifecycle will be bound to
// the lifecycle of the query.
System.out.println("Executing inline UDFs...");
env.fromValues(row("Apple", "USA", 2), row("Apple", "EU", 3))
.select(
$("f0").as("product"),
$("f1").as("location"),
$("f2").times(call(CustomTax.class, $("f1"))).as("tax"))
.execute()
.print();
env.fromValues(
row(1L, "Ann", array("Apples", "Bananas")),
row(2L, "Peter", array("Apples", "Pears")))
.joinLateral(call(Explode.class, $("f2")).as("fruit"))
.select($("f0").as("id"), $("f1").as("name"), $("fruit"))
.execute()
.print();
}
/** A scalar function that calculates a custom tax based on the provided location. */
public static class CustomTax extends ScalarFunction {
public int eval(String location) {
if (location.equals("USA")) {
return 10;
}
if (location.equals("EU")) {
return 5;
}
return 0;
}
}
/** A table function that explodes an array of string into multiple rows. */
public static class Explode extends TableFunction {
public void eval(List arr) {
for (String i : arr) {
collect(i);
}
}
}
}
```
## Table functions
Confluent Cloud for Apache Flink also supports user-defined table functions (UDTFs), which take
multiple scalar values as input arguments and return multiple rows as output,
instead of a single value.
To create a user-defined table function, extend the `TableFunction` base
class in `org.apache.flink.table.functions` and implement one or more
of the evaluation methods, which are named `eval(...)`. Input and output data
types are inferred automatically by using reflection, including the generic
argument `T` of the class, for determining the output data type. Unlike
scalar functions, the evaluation method itself doesn’t have a return type.
Instead, a table function provides a `collect(T)` method that’s called within
every evaluation method to emit zero, one, or more records.
In the [Table API](../reference/table-api.md#flink-table-api), a table function is used with the
`.joinLateral(...)` or `.leftOuterJoinLateral(...)` operators. The
`joinLateral` operator cross-joins each row from the outer table (the table
on the left of the operator) with all rows produced by the table-valued
function (on the right side of the operator). The `leftOuterJoinLateral`
operator joins each row from the outer table with all rows produced by the
table-valued function and preserves outer rows, for which the table function
returns an empty table.
#### NOTE
User-defined table functions are distinct from the
[Table API](../reference/table-api.md#flink-table-api) but can be used in Table API code.
In SQL, use `LATERAL TABLE()` with `JOIN` or `LEFT JOIN`
with an `ON TRUE` join condition.
The following code example shows how to implement a simple string splitting
function in Java.
```java
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import static org.apache.flink.table.api.Expressions.*;
@FunctionHint(output = @DataTypeHint("ROW"))
public static class SplitFunction extends TableFunction {
public void eval(String str) {
for (String s : str.split(" ")) {
// use collect(...) to emit a row
collect(Row.of(s, s.length()));
}
}
}
```
The following example shows how to call the `SplitFunction` UDTF in a
Flink SQL statement.
```sql
SELECT myField, word, length
FROM MyTable
LEFT JOIN LATERAL TABLE(SplitFunction(myField)) ON TRUE;
```
To build and upload a user-defined table function to Confluent Cloud for Apache Flink for use in
Flink SQL, see
[Create a User Defined Table Function](../how-to-guides/create-udf.md#flink-sql-implement-udtf-function).
## Implementation considerations
All UDFs adhere to a few common implementation principles, which are described
in the following sections.
- [Function class](#flink-sql-udfs-function-class)
- [Evaluation methods](#flink-sql-udfs-evaluation-methods)
- [Type inference](#flink-sql-udfs-type-inference)
- [Named parameters](#flink-sql-udfs-named-parameters)
- [Scalar functions](#flink-sql-udfs-scalar-functions)
- [Table functions](#flink-sql-udfs-table-functions)
The following code example shows how to implement a simple scalar function and
how to call it in Flink SQL.
For the [Table API](../reference/table-api.md#flink-table-api), you can register the function in
code and invoke it.
For SQL queries, your UDF must be registered by using the
[CREATE FUNCTION](../reference/statements/create-function.md#flink-sql-create-function) statement. For more
information, see [Create a User-defined Function](../how-to-guides/create-udf.md#flink-sql-create-udf).
```java
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.ScalarFunction;
import static org.apache.flink.table.api.Expressions.*;
// define function logic
public static class SubstringFunction extends ScalarFunction {
public String eval(String s, Integer begin, Integer end) {
return s.substring(begin, end);
}
}
```
The following example shows how to call the `SubstringFunction` UDF in a
Flink SQL statement.
```sql
SELECT SubstringFunction('test string', 2, 5);
```
### Function class
Your implementation class must extend one of the system-provided base classes.
- Scalar functions extend the `org.apache.flink.table.functions.ScalarFunction`
class.
- Table functions extend the `org.apache.flink.table.functions.TableFunction`
class.
The class must be declared public, not abstract, and must be accessible
globally. Non-static inner or anonymous classes are not supported.
### Evaluation methods
You define the behavior of a scalar function by implementing a custom
evaluation method, named `eval`, which must be declared `public`.
You can overload evaluation methods by implementing multiple methods named
`eval`.
The evaluation method is called by code-generated operators during runtime.
Regular JVM method-calling semantics apply, so these implementation options
are available:
- You can implement overloaded methods, like `eval(Integer)` and
`eval(LocalDateTime)`.
- You can use var-args, like `eval(Integer...)`.
- You can use object inheritance, like `eval(Object)` that takes both
`LocalDateTime` and `Integer`.
- You can use combinations of these, like `eval(Object...)` that takes all
kinds of arguments.
The `ScalarFunction` base class provides a set of optional methods that you
can override, `open()`, `close()`, `isDeterministic()`, and
`supportsConstantFolding()`. You can use the `open()` method for
initialization work and the `close()` method for cleanup work.
Internally, Table API and SQL code generation works with primitive values where
possible. To reduce overhead during runtime, a user-defined scalar function
should declare parameters and result types as primitive types instead of their
boxed classes. For example, DATE/TIME is equal to `int`, and TIMESTAMP is
equal to `long`.
The following code example shows a user-defined function that has overloaded
`eval` methods.
```java
import org.apache.flink.table.functions.ScalarFunction;
// function with overloaded evaluation methods
public static class SumFunction extends ScalarFunction {
public Integer eval(Integer a, Integer b) {
return a + b;
}
public Integer eval(String a, String b) {
return Integer.valueOf(a) + Integer.valueOf(b);
}
public Integer eval(Double... d) {
double result = 0;
for (double value : d)
result += value;
return (int) result;
}
}
```
### Type inference
The Table API is strongly typed, so both function parameters and return types
must be mapped to a data type.
The Flink planner needs information about expected types, precision, and scale.
Also it needs information about how internal data structures are represented
as JVM objects when calling a user-defined function.
*Type inference* is the process of validating input arguments and deriving data
types for both the parameters and the result of a function.
User-defined functions in Flink implement automatic type-inference extraction
that derives data types from the function’s class and its evaluation methods
by using reflection. If this implicit extraction approach with reflection
fails, you can help the extraction process by annotating affected parameters,
classes, or methods with `@DataTypeHint` and `@FunctionHint`.
#### Automatic type inference
Automatic type inference inspects the function’s class and evaluation methods
to derive data types for the arguments and return value of a function. The
`@DataTypeHint` and `@FunctionHint` annotations support automatic
extraction.
For a list of classes that implicitly map to a data type, see
[Data type extraction](../reference/datatypes.md#flink-sql-data-type-extraction).
#### Data type hints
In some situations, you might need to support automatic extraction inline
for parameters and return types of a function. In these cases you can use
data type hints and the `@DataTypeHint` annotation to define data types.
The following code example shows how to use data type hints.
```java
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;
// user-defined function that has overloaded evaluation methods.
public static class OverloadedFunction extends ScalarFunction {
// No hint required for type inference.
public Long eval(long a, long b) {
return a + b;
}
// Define the precision and scale of a decimal.
public @DataTypeHint("DECIMAL(12, 3)") BigDecimal eval(double a, double b) {
return BigDecimal.valueOf(a + b);
}
// Define a nested data type.
@DataTypeHint("ROW")
public Row eval(int i) {
return Row.of(String.valueOf(i), Instant.ofEpochSecond(i));
}
// Enable wildcard input and custom serialized output.
@DataTypeHint(value = "RAW", bridgedTo = ByteBuffer.class)
public ByteBuffer eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
return MyUtils.serializeToByteBuffer(o);
}
}
```
#### Function hints
In some situations, you might want one evaluation method to handle multiple
different data types, or you might have overloaded evaluation methods with
a common result type to declare only once.
The `@FunctionHint` annotation provides a mapping from argument data types
to a result data type. It enables annotating entire function classes or
evaluation methods for input, accumulator, and result data types. You can
declare one or more annotations on a class or individually for each evaluation
method for overloading function signatures.
All hint parameters are optional. If a parameter is not defined, the default
reflection-based extraction is used. Hint parameters defined on a function
class are inherited by all evaluation methods.
The following code example shows how to use function hints.
```java
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
// User-defined function with overloaded evaluation methods
// but globally defined output type.
@FunctionHint(output = @DataTypeHint("ROW"))
public static class OverloadedFunction extends ScalarFunction {
public void eval(int a, int b) {
collect(Row.of("Sum", a + b));
}
// Overloading arguments is still possible.
public void eval() {
collect(Row.of("Empty args", -1));
}
}
// Decouples the type inference from evaluation methods.
// The type inference is entirely determined by the function hints.
@FunctionHint(
input = {@DataTypeHint("INT"), @DataTypeHint("INT")},
output = @DataTypeHint("INT")
)
@FunctionHint(
input = {@DataTypeHint("BIGINT"), @DataTypeHint("BIGINT")},
output = @DataTypeHint("BIGINT")
)
@FunctionHint(
input = {},
output = @DataTypeHint("BOOLEAN")
)
public static class OverloadedFunction extends ScalarFunction