# Create a User-Defined Function with Confluent Cloud for Apache Flink
A [user-defined function (UDF)](../concepts/user-defined-functions.md#flink-sql-udfs) extends the capabilities
of Confluent Cloud for Apache Flink® and enables you to implement custom logic beyond what SQL
supports. For example, you can implement functions like encoding and
decoding a string, performing geospatial calculations, encrypting and decrypting
fields, or reusing an existing library or code from a third-party supplier.
Confluent Cloud for Apache Flink supports UDFs written in Java and Python.
An artifact is the compiled JAR for Java or packaged distribution for Python
that contains your UDF code and its dependencies. You upload the artifact to
Confluent Cloud so that Flink can run it.
- **Java UDFs**: Package your custom function and its
dependencies into a JAR file and upload it as an artifact to Confluent Cloud.
Register the function in a Flink database by using the
[CREATE FUNCTION](../reference/statements/create-function.md#flink-sql-create-function) statement, and invoke your
UDF in Flink SQL or the [Table API](../reference/table-api.md#flink-table-api). Confluent Cloud
provides the infrastructure to run your code.
- **Python UDFs**: Package your custom function and its
dependencies into a Python package and upload it as an artifact to Confluent Cloud.
Register the function in a Flink database by using the
[CREATE FUNCTION](../reference/statements/create-function.md#flink-sql-create-function) statement, and invoke your
UDF in Flink SQL. Confluent Cloud provides the infrastructure to run your code.
For a list of cloud service providers and regions that support UDFs, see
[UDF regional availability](../concepts/user-defined-functions.md#flink-sql-udfs-availability).
The following steps show how to implement a simple
[user-defined scalar function](../concepts/user-defined-functions.md#flink-sql-udfs-scalar-functions), upload
it to Confluent Cloud, and use it in a Flink SQL statement.
- [Step 0: Create a connection (optional)](#flink-sql-create-udf-create-connection)
- [Step 1: Build the artifact](#flink-sql-create-udf-create-artifact)
- [Step 2: Upload the UDF as a Flink artifact](#flink-sql-create-udf-upload-artifact)
- [Step 3: Register the UDF](#flink-sql-create-udf-register)
- [Step 4: Use the UDF in a Flink SQL query](#flink-sql-create-udf-use)
- [Step 5: Implement UDF logging (optional)](#flink-sql-implement-udf-log)
- [Step 6: Delete the UDF](#flink-sql-create-udf-delete)
After you build and run the scalar function, try
[building a table function](#flink-sql-implement-udtf-function).
For more code examples, see
[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),
which include a vectorized (`func_type="pandas"`) scalar function example.
## Prerequisites
You need the following prerequisites to use the Flink SQL shell.
- Access to Confluent Cloud.
- The organization ID and environment ID for your organization.
- The cloud provider and region where you run your Flink SQL statements.
Default compute pools are scoped to an environment and a region, so you
must specify both unless you name a compute pool.
- (Optional) A compute pool ID, if you want to use a specific compute pool instead
of the default pool. For more information, see [Compute Pools](../concepts/compute-pools.md#flink-sql-compute-pools).
- The FlinkDeveloper role is granted by default to all users in an environment.
To create compute pools manually for workload isolation or cost control,
you need the OrganizationAdmin, EnvironmentAdmin, or FlinkAdmin role.
If you don’t have the appropriate role, contact your OrganizationAdmin
or EnvironmentAdmin.
- The Confluent CLI. To use the Flink SQL shell, update to the latest
version of the Confluent CLI by running the following command:
```bash
confluent update --yes
```
If you used Homebrew to install the Confluent CLI, update the CLI by
using the `brew upgrade` command, instead of `confluent update`.
For more information, see [Confluent CLI](https://docs.confluent.io/confluent-cli/current/overview.html).
- Sufficient permissions to upload and invoke UDFs in Confluent Cloud. For more
information, see [Flink RBAC](../operate-and-deploy/flink-rbac.md#flink-rbac).
### Java
- Apache Maven, a tool for managing software projects. For details, see
[Installing Apache Maven](https://maven.apache.org/install.html).
- Java 11 to Java 21.
- If using the Table API only, Flink supports `flink-table-api-java`
versions 1.18.x through 2.1.0, the version used in the
`pom.xml` example in [Step 1: Build the artifact](#flink-sql-create-udf-create-artifact).
### Python
- Python 3.11.
- The [uv](https://docs.astral.sh/uv/) package manager to manage your
Python versions and environments. Confluent Cloud supports only Python version
3.11.
## Step 0: Create a connection (optional)
If your UDF requires external connectivity, you must first create a connection
object by using the [CREATE CONNECTION](../reference/statements/create-connection.md#flink-sql-create-connection)
statement.
```sql
-- Example: Create a connection to an external REST service.
CREATE CONNECTION my_external_service
WITH (
'type' = 'REST',
'endpoint' = 'https://api.example.com/v1/resource',
'token' = 'my-token'
);
```
## Step 1: Build the artifact
### Build the uber JAR
In this section, you compile a simple Java class named
`TShirtSizingIsSmaller` into a JAR file. The project is based on the
`ScalarFunction` class in the Flink Table API. The
`TShirtSizingIsSmaller.java` class has an `eval` function that
compares two T-shirt sizes and returns the smaller size.
1. Copy the following project object model into a file named `pom.xml`.
#### IMPORTANT
You can’t use your own Flink-related JARs. If you package Flink core
dependencies as part of the JAR, you can break the dependency.
Also, this example shows how to capture all dependencies greedily,
possibly including more than needed. As an alternative, you can
optimize on artifact size by listing all dependencies and including
their transitive dependencies.
### pom.xml
```xml
4.0.0
example
udf_example
1.0
11
11
UTF-8
org.apache.flink
flink-table-api-java
2.1.0
provided
./example
org.apache.maven.plugins
maven-shade-plugin
3.6.0
*:*
*
META-INF/*.SF
META-INF/*.DSA
META-INF/*.RSA
package
shade
```
2. Create a directory named `example`.
```bash
mkdir example
```
3. In the `example` directory, create a file named
`TShirtSizingIsSmaller.java`.
```bash
touch example/TShirtSizingIsSmaller.java
```
4. Copy the following code into `TShirtSizingIsSmaller.java`.
```java
package com.example.my;
import org.apache.flink.table.functions.ScalarFunction;
import java.util.Arrays;
import java.util.List;
import java.util.stream.IntStream;
/** TShirt sizing function for demo. */
public class TShirtSizingIsSmaller extends ScalarFunction {
public static final String NAME = "IS_SMALLER";
private static final List ORDERED_SIZES =
Arrays.asList(
new Size("X-Small", "XS"),
new Size("Small", "S"),
new Size("Medium", "M"),
new Size("Large", "L"),
new Size("X-Large", "XL"),
new Size("XX-Large", "XXL"));
public boolean eval(String shirt1, String shirt2) {
int size1 = findSize(shirt1);
int size2 = findSize(shirt2);
// If either can't be found just say false rather than throw an error
if (size1 == -1 || size2 == -1) {
return false;
}
return size1 < size2;
}
private int findSize(String shirt) {
return IntStream.range(0, ORDERED_SIZES.size())
.filter(
i -> {
Size s = ORDERED_SIZES.get(i);
return s.name.equalsIgnoreCase(shirt)
|| s.abbreviation.equalsIgnoreCase(shirt);
})
.findFirst()
.orElse(-1);
}
private static class Size {
private final String name;
private final String abbreviation;
public Size(String name, String abbreviation) {
this.name = name;
this.abbreviation = abbreviation;
}
}
}
```
5. Optionally, access external connections by overriding the `open()`
method and using `context.getJobParameter` to retrieve secrets.
```java
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.FunctionContext;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.URI;
import java.time.Duration;
public class ExternalEnrichmentUDF extends ScalarFunction {
private transient HttpClient httpClient;
private transient String endpoint;
private transient String token;
@Override
public void open(FunctionContext context) throws Exception {
// Retrieve connection details using the connection name defined in SQL
this.endpoint = context.getJobParameter("my_external_service.endpoint", null);
this.token = context.getJobParameter("my_external_service.token", null);
// Initialize HttpClient
this.httpClient = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(5))
.build();
}
public String eval(String id) {
try {
// Construct and send request. The request timeout bounds how
// long a slow endpoint can block the task thread. Choose a
// value that suits your endpoint.
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(this.endpoint + "?id=" + id))
.header("Authorization", "Bearer " + this.token)
.timeout(Duration.ofSeconds(2))
.GET()
.build();
HttpResponse response = this.httpClient.send(request, HttpResponse.BodyHandlers.ofString());
return response.body();
} catch (Exception e) {
return "Error: " + e.getMessage();
}
}
```
6. Run the following command to build the JAR file.
```bash
mvn clean package
```
7. Run the following command to check the contents of your JAR.
```bash
jar -tf target/udf_example-1.0.jar | grep -i TShirtSizingIsSmaller
```
Your output should resemble:
```none
com/example/my/TShirtSizingIsSmaller$Size.class
com/example/my/TShirtSizingIsSmaller.class
```
### Build the Python package
1. Create a new library project with your desired name and Python
version. Confluent Cloud supports only Python version 3.11.
```bash
uv init -p 3.11 --lib example_udf
cd example_udf
```
2. Append the following to `example_udf/pyproject.toml`. This ensures
that your local dependencies are compatible with those available in
the Confluent Cloud environment.
```toml
[tool.uv]
constraint-dependencies = [
"apache-flink==2.0.0",
"click>=8.2.0",
"confluent-function-runtime-core>=0.181.0",
"grpc-interceptor>=0.15.0",
"grpcio-health-checking>=1.65.0",
"grpcio-reflection>=1.65.0",
"grpcio>=1.65.0",
"protobuf>=5.29.1",
"psutil>=5.9.0",
"pyparsing>=3.2.5",
"typing_extensions>=4.4.0",
]
```
Add a dependency on `apache-flink` to access the PyFlink UDF API.
Confluent Cloud supports only version 2.0.0 of the Apache Flink Python API.
```bash
uv add 'apache-flink==2.0.0'
```
3. Create a new file named `example_udf/tshirt_sizing.py` and add the
following code:
```python
from pyflink.table import DataTypes
from pyflink.table.types import DataType
from pyflink.table.udf import udf
# Dictionary mapping both full names and abbreviations to numeric size values
_SIZE_MAP = {
"x-small": 0, "xs": 0,
"small": 1, "s": 1,
"medium": 2, "m": 2,
"large": 3, "l": 3,
"x-large": 4, "xl": 4,
"xx-large": 5, "xxl": 5,
}
def _get_size_value(shirt: str) -> int:
"""
Returns the numeric size value for a given shirt size string.
Returns -1 if the size is not found.
"""
if shirt is None:
return -1
return _SIZE_MAP.get(shirt.strip().lower(), -1)
def _f_is_smaller(shirt1: str, shirt2: str) -> bool:
"""
Returns True if shirt1 is a smaller size than shirt2 based on standard T-shirt sizes.
If a size cannot be found, returns False.
"""
size1 = _get_size_value(shirt1)
size2 = _get_size_value(shirt2)
if size1 == -1 or size2 == -1:
return False
return size1 < size2
_is_smaller_inp_types: list[DataType] = [
DataTypes.STRING(),
DataTypes.STRING(),
]
is_smaller = udf(
_f_is_smaller,
input_types=_is_smaller_inp_types,
result_type=DataTypes.BOOLEAN(),
)
```
4. Build the sdist using uv.
```bash
uv build --sdist
```
Your output should resemble:
```text
Successfully built dist/example_udf-0.1.0.tar.gz
```
5. Re-package the sdist into a zip.
```bash
zip -r dist/example_udf-0.1.0.zip dist/example_udf-0.1.0.tar.gz
```
Your output should resemble:
```text
.gz
adding: dist/example_udf-0.1.0.tar.gz (stored 0%)
```
## Step 2: Upload the UDF as a Flink artifact
You can use the Confluent Cloud Console, the Confluent CLI, or the REST API to
upload your UDF.
### Confluent Cloud Console
1. Log in to Confluent Cloud.
2. In the navigation menu, click **Environments**, and click the tile
for the environment where you want to run the UDF.
3. In the environment details page, click **Flink**.
4. In the **Flink** page, click **Artifacts**.
5. Click **Upload artifact** to open the upload pane.
6. In the **Choose the type of UDF artifact** section, select **Java** or
**Python**.
7. In the **Cloud provider** dropdown, select **AWS**, and in the
**Region** dropdown, select the cloud region.
#### NOTE
The **Region** dropdown lists all regions available for the
selected cloud provider, not only regions where you already have
a compute pool. You can upload an artifact to a region before
creating a compute pool there.
8. Click **Upload your artifact** and navigate to the location of your
JAR or ZIP file, which in the current example is
`target/udf_example-1.0.jar` or `dist/example_udf-0.1.0.zip`.
9. After you upload your JAR or ZIP file, it appears in the
**Artifacts** list. In the list, click the row for your UDF artifact
to open the details pane.
### Confluent CLI
1. Log in to Confluent Cloud.
```bash
confluent login --organization ${ORG_ID} --prompt
```
2. Run the following command to upload the JAR to Confluent Cloud.
### Java
```bash
confluent flink artifact create udf_example \
--artifact-file target/udf_example-1.0.jar \
--cloud ${CLOUD_PROVIDER} \
--region ${CLOUD_REGION} \
--environment ${ENV_ID}
```
Your output should resemble:
```text
+--------------------+-------------+
| ID | cfa-ldxmro |
| Name | udf_example |
| Version | ver-81vxm5 |
| Cloud | aws |
| Region | us-east-1 |
| Environment | env-z3q9rd |
| Content Format | JAR |
| Description | |
| Documentation Link | |
+--------------------+-------------+
```
### Python
Python UDF artifacts have a 100 MB size limit. If you need
more space, contact Confluent support at
[https://support.confluent.io](https://support.confluent.io).
```bash
confluent flink artifact create udf_example \
--artifact-file dist/example_udf-0.1.0.zip \
--runtime-language python \
--cloud ${CLOUD_PROVIDER} \
--region ${CLOUD_REGION} \
--environment ${ENV_ID}
```
Your output should resemble:
```text
+--------------------+-------------+
| ID | cfa-ldxmro |
| Name | udf_example |
| Version | ver-81vxm5 |
| Cloud | aws |
| Region | us-east-1 |
| Environment | env-z3q9rd |
| Content Format | ZIP |
| Description | |
| Documentation Link | |
+--------------------+-------------+
```
Note the artifact ID and version of your UDF, which in this example
are `cfa-ldxmro` and `ver-81vxm5`, because you use them later to
register the UDF in Flink SQL and to manage the artifact.
3. Run the following command to view all of the available UDFs.
```bash
confluent flink artifact list \
--cloud ${CLOUD_PROVIDER} \
--region ${CLOUD_REGION}
```
Your output should resemble:
```none
ID | Name | Cloud | Region | Environment
-------------+-------------+-------+-----------+--------------
cfa-ldxmro | udf_example | AWS | us-east-1 | env-z3q9rd
```
4. Run the following command to view the details of your UDF. You can use
the artifact ID from the previous step or the artifact name to specify
your UDF.
```bash
# use the artifact ID
confluent flink artifact describe \
cfa-ldxmro \
--cloud ${CLOUD_PROVIDER} \
--region ${CLOUD_REGION}
# use the artifact name
confluent flink artifact describe \
udf_example \
--cloud ${CLOUD_PROVIDER} \
--region ${CLOUD_REGION}
```
Your output should resemble:
```text
+--------------------+-------------+
| ID | cfa-ldxmro |
| Name | udf_example |
| Version | ver-81vxm5 |
| Cloud | aws |
| Region | us-east-1 |
| Environment | env-z3q9rd |
| Content Format | JAR |
| Description | |
| Documentation Link | |
+--------------------+-------------+
```
### REST API
You can upload your JAR or ZIP file by requesting a presigned upload URL,
then uploading the file by using the presigned URL information. For more
information, see [Create a Flink artifact](../operate-and-deploy/flink-rest-api.md#flink-rest-api-create-artifact).
## Step 3: Register the UDF
You register UDFs inside a Flink database, so you must specify
the Confluent Cloud environment, which is the Flink catalog, and Apache Kafka® cluster,
which is the Flink database, where you want to use the UDF.
You can use the Confluent Cloud Console, the Confluent CLI, the Confluent
Terraform provider, or the REST API to register your UDF.
### Confluent Cloud Console
1. In the navigation menu, click **Flink**.
2. In the Flink page, click **Compute pools**.
3. In the tile for the compute pool where you want to run the UDF, click
**Open SQL workspace**.
4. In the **Use catalog** dropdown, select the environment where you want
to run the UDF.
5. In the **Use database** dropdown, select the Kafka cluster where you
want to run the UDF.
### Confluent CLI
1. Run the following command to start the Flink shell.
```bash
# The compute-pool parameter is optional and you can omit it if
# you're using the default compute pool.
confluent flink shell --environment ${ENV_ID} --compute-pool ${COMPUTE_POOL_ID}
```
2. Run the following statements to specify the catalog and database.
```sql
-- Specify your catalog. This example uses the default.
USE CATALOG default;
```
Your output should resemble:
```none
+---------------------+---------+
| Key | Value |
+---------------------+---------+
| sql.current-catalog | default |
+---------------------+---------+
```
Specify the database you want to use, for example, `cluster_0`.
```sql
-- Specify your database. This example uses cluster_0.
USE cluster_0;
```
Your output should resemble:
```none
+----------------------+-----------+
| Key | Value |
+----------------------+-----------+
| sql.current-database | cluster_0 |
+----------------------+-----------+
```
### Terraform
You can register a previously uploaded UDF by using the Confluent
Terraform provider. For more information, see
[confluent_flink_artifact Resource](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs/resources/confluent_flink_artifact).
### REST API
You can register a UDF by sending a POST request to the
[Create Artifact endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/createArtifactV1FlinkArtifact).
For more information, see [Create a Flink artifact](../operate-and-deploy/flink-rest-api.md#flink-rest-api-create-artifact).
- In Cloud Console or the Confluent CLI, run the
: [CREATE FUNCTION](../reference/statements/create-function.md#flink-sql-create-function) statement to register
your UDF in the current catalog and database. Substitute your
UDF’s value for ``.
### Java
```sql
-- If you have an external connection, the connection name in the
-- USING CONNECTIONS clause must match the prefix used in getJobParameter,
-- for example, my_external_service.endpoint.
CREATE FUNCTION is_smaller
AS 'com.example.my.TShirtSizingIsSmaller'
USING JAR 'confluent-artifact://'
USING CONNECTIONS ('my_external_service');
```
Your output should resemble:
```none
Function 'is_smaller' created.
```
### Python
```sql
CREATE FUNCTION is_smaller AS 'example_udf.tshirt_sizing.is_smaller'
LANGUAGE PYTHON
USING JAR 'confluent-artifact://';
```
Your output should resemble:
```none
Function 'is_smaller' created.
```
## Step 4: Use the UDF in a Flink SQL query
After it’s registered, your UDF is available to use in queries. In this step,
you create a `sizes` table that contains the data for the `is_smaller` UDF,
and you run the UDF by using a SELECT statement.
1. Run the following statement to view the UDFs in the current database.
```sql
SHOW USER FUNCTIONS;
```
Your output should resemble:
```none
+---------------+
| Function Name |
+---------------+
| is_smaller |
+---------------+
```
2. Run the following statement to create a `sizes` table.
```sql
CREATE TABLE sizes (
`size_1` STRING,
`size_2` STRING
);
```
3. Run the following statement to populate the `sizes` table with values.
```sql
INSERT INTO sizes VALUES
('XL', 'L'),
('small', 'L'),
('M', 'L'),
('XXL', 'XL');
```
4. Run the following statement to view the rows in the `sizes` table.
```sql
SELECT * FROM sizes;
```
Your output should resemble:
```none
size_1 size_2
XL L
small L
M L
XXL XL
```
5. Run the following statement to execute the `is_smaller` function on the
data in the `sizes` table.
```sql
SELECT size_1, size_2, is_smaller (size_1, size_2)
AS is_smaller
FROM sizes;
```
Your output should resemble:
```none
size_1 size_2 is_smaller
XL L FALSE
small L TRUE
M L TRUE
XXL XL FALSE
```
## Step 5: Implement UDF logging (optional)
If you want to log UDF status messages, follow the steps in
[Log Debug Messages in UDFs](enable-udf-logging.md#flink-sql-enable-udf-logging).
## Step 6: Delete the UDF
When you’re finished using the UDF, you can delete it from the current database.
You can use the Confluent Cloud Console, the Confluent CLI, the Confluent
Terraform provider, or the REST API to delete your UDF.
### Drop the function
Use the following command to remove the function from the current database.
#### WARNING
Follow the safe procedure in [Update a UDF safely](#flink-sql-update-udf) before you drop a
function or delete its artifact. Dropping and recreating a function can
permanently break statements that depend on it but aren’t currently
running, because a statement’s compiled plan pins the UDF to the artifact
version that existed when the statement was submitted.
```sql
DROP FUNCTION is_smaller;
```
Your output should resemble:
```none
Function 'is_smaller' dropped.
```
Currently running statements are unaffected.
### Delete the artifact
### Confluent Cloud Console
1. In the navigation menu, click **Environments**, and click the tile
for the environment where your UDF is registered.
2. In the environment details page, click **Flink**.
3. In the **Flink** page, click **Artifacts**.
4. In the artifacts list, find the UDF you want to delete.
5. In the **Actions** column, click the icon, and in the context menu,
select **Delete artifact**.
6. In the confirmation dialog, type `udf_example`, and click
**Confirm**. The “Artifact deleted successfully” message appears.
### Confluent CLI
1. Run the following command to delete the artifact from the environment.
```bash
confluent flink artifact delete \
\
--cloud ${CLOUD_PROVIDER} \
--region ${CLOUD_REGION}
```
You receive a warning about breaking Flink statements that use the
artifact. Type `y` at the prompt to proceed.
Your output should resemble:
```none
Deleted Flink artifact "".
```
### Terraform
You can delete a UDF by using the Confluent Terraform provider. For
more information, see
[confluent_flink_artifact Resource](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs/resources/confluent_flink_artifact).
### REST API
You can delete a UDF by sending a DELETE request to the
[Delete Artifact endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/deleteArtifactV1FlinkArtifact).
For more information, see [Delete an artifact](../operate-and-deploy/flink-rest-api.md#flink-rest-api-delete-artifact).
## Update a UDF safely
UDFs are immutable. Flink has no `ALTER FUNCTION` statement, so you can’t
change a registered function’s code, class, or artifact in place. To update a
UDF, you drop the existing function and create a new one that points to the
updated artifact.
Update a UDF with care, because a statement’s compiled plan pins the UDF to the
artifact version that existed when the statement was submitted. Dropping a
function that other statements still depend on can permanently break those
statements. For more information, see
[DROP FUNCTION](../reference/statements/drop-function.md#flink-sql-drop-function).
To update a UDF safely, use the following rollout order:
1. Upload the updated code as a new artifact.
Build your updated JAR or ZIP
file and upload it with `confluent flink artifact create`, as described in
[Step 2: Upload the UDF as a Flink artifact](#flink-sql-create-udf-upload-artifact). Each upload produces a new
artifact with its own ID, so don’t delete the old artifact yet.
2. Recreate the function.
Drop the function and create it again with the
same name, pointing to the new artifact, as described in
[CREATE FUNCTION Statement](../reference/statements/create-function.md#flink-sql-create-function). The function name now resolves to the
updated code. Currently running statements are unaffected, because their
compiled plans still point to the old artifact.
3. Recreate the dependent statements.
Drop and recreate the statements that
use the UDF so that they recompile and bind to the new artifact. You can’t
swap the UDF underneath a running or stopped statement in place.
4. Delete the old artifact.
Only after no statement references the old
artifact, delete it, as described in
[Delete the artifact](#flink-sql-create-udf-delete-artifact).
### Recover a statement pinned to a deleted artifact
To recover a statement whose pinned artifact was deleted, force the statement
to recompile against the current function definition by recreating it:
- In Terraform, run `terraform apply -replace=''` to destroy
and recreate the statement.
- Outside Terraform, drop and recreate the statement manually.
This situation occurs when you drop a function and delete its artifact while a
stopped or failed statement still depends on it. The statement can’t resume,
because its compiled plan points to an artifact that no longer exists, and
resuming fails with an artifact-not-found error. In Terraform, this appears as an
`apply` command that hangs with `Still modifying...` and eventually fails
with a context-deadline-exceeded error.
## Implement a user-defined table function (Java only)
Confluent Cloud for Apache Flink also supports
[user-defined table functions (UDTFs)](../concepts/user-defined-functions.md#flink-sql-udfs-table-functions),
which take multiple scalar values as input arguments and return multiple rows
as output, instead of a single value. UDTFs build on the scalar UDF you
implemented in the previous steps.
#### NOTE
Python does not support user-defined table functions.
The following steps show how to implement a simple UDTF, upload it to Confluent Cloud,
and use it in a Flink SQL statement.
- [Step 1: Build the uber JAR](#flink-sql-create-udtf-create-jar)
- [Step 2: Upload the UDTF JAR as a Flink artifact](#flink-sql-create-udtf-upload-jar)
- [Step 3: Register the UDTF](#flink-sql-create-udtf-register)
- [Step 4: Use the UDTF in a Flink SQL query](#flink-sql-create-udtf-use)
### Step 1: Build the uber JAR
In this section, you compile a simple Java class named `SplitFunction` into
a JAR file, similar to the previous section. The class is based on the
`TableFunction` class in the Flink Table API. The `SplitFunction.java` class
has an `eval` function that uses the Java `split` method to break up a
string into words and returns the words as columns in a row.
1. In the `example` directory, create a file named `SplitFunction.java`.
```bash
touch example/SplitFunction.java
```
2. Copy the following code into `SplitFunction.java`.
```java
package com.example.my;
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 class SplitFunction extends TableFunction {
public void eval(String str, String delimiter) {
for (String s : str.split(delimiter)) {
// use collect(...) to emit a row
collect(Row.of(s));
}
}
}
```
3. Run the following command to build the JAR file. You can use the POM file
from the [previous section](#flink-sql-create-udf-create-artifact).
```bash
mvn clean package
```
4. Run the following command to check the contents of your JAR.
```bash
jar -tf target/udf_example-1.0.jar | grep -i SplitFunction
```
Your output should resemble:
```none
com/example/my/SplitFunction.class
```
### Step 2: Upload the UDTF JAR as a Flink artifact
### Confluent Cloud Console
1. Log in to Confluent Cloud.
2. In the navigation menu, click **Environments**, and click the tile
for the environment where you want to run the UDF.
3. In the environment details page, click **Flink**.
4. In the **Flink** page, click **Artifacts**.
5. Click **Upload artifact** to open the upload pane.
6. In the **Cloud provider** dropdown, select **AWS**, and in the
**Region** dropdown, select the cloud region.
7. Click **Upload your JAR file** and navigate to the location of your
JAR file, which in the current example is
`target/udf_example-1.0.jar`.
8. After you upload your JAR file, it appears in the **Artifacts** list.
In the list, click the row for your UDF artifact to open the details
pane.
### Confluent CLI
1. Log in to Confluent Cloud.
```bash
confluent login --organization ${ORG_ID} --prompt
```
2. Run the following command to upload the JAR to Confluent Cloud.
```bash
confluent flink artifact create udf_table_example \
--artifact-file target/udf_example-1.0.jar \
--cloud ${CLOUD_PROVIDER} \
--region ${CLOUD_REGION} \
--environment ${ENV_ID}
```
Your output should resemble:
```text
+--------------------+-------------------+
| ID | cfa-l5xp82 |
| Name | udf_table_example |
| Version | ver-0x37m2 |
| Cloud | aws |
| Region | us-east-1 |
| Environment | env-z3q9rd |
| Content Format | JAR |
| Description | |
| Documentation Link | |
+--------------------+-------------------+
```
Note the artifact ID and version of your UDTF, which in this example
are `cfa-l5xp82` and `ver-0x37m2`, because you use them later to
register the UDTF in Flink SQL and to manage it.
### Step 3: Register the UDTF
1. In the Flink shell or the Cloud Console, specify the catalog and
database (environment and cluster) where you want to use the UDTF, as you did
in the [previous section](#flink-sql-create-udf-register).
2. Run the [CREATE FUNCTION](../reference/statements/create-function.md#flink-sql-create-function) statement to
register your UDTF in the current catalog and database. Substitute your
UDTF’s value for ``.
```sql
CREATE FUNCTION split_string
AS 'com.example.my.SplitFunction'
USING JAR 'confluent-artifact://';
```
Your output should resemble:
```none
Function 'split_string' created.
```
### Step 4: Use the UDTF in a Flink SQL query
After it’s registered, your UDTF is available to use in queries.
1. Run the following statement to view the UDFs in the current database.
```sql
SHOW USER FUNCTIONS;
```
Your output should resemble:
```none
+---------------+
| Function Name |
+---------------+
| split_string |
+---------------+
```
2. Run the following statement to execute the `split_string` function.
```sql
SELECT * FROM (VALUES 'A;B', 'C;D;E;F') as T(f), LATERAL TABLE(split_string(f, ';'));
```
Your output should resemble:
```none
f word
A;B A
A;B B
C;D;E;F C
C;D;E;F D
C;D;E;F E
C;D;E;F F
```
3. When you’re done with the example UDTF, drop the function and delete the
JAR artifact as you did in [Step 6: Delete the UDF](#flink-sql-create-udf-delete).
## Error handling best practices
The
[error-handling.mode](../reference/statements/create-table.md#flink-sql-create-table-with-error-handling-mode)
table property handles deserialization errors at the source, but it does not
catch exceptions thrown inside UDFs. If an unhandled exception occurs in a UDF,
the Flink statement fails.
To build resilient pipelines that use UDFs, handle errors inside your UDF code.
### Catch exceptions in the eval method
Wrap your UDF logic in a try-catch block to prevent unhandled exceptions from
failing the statement.
### Java
```java
public class SafeParseDouble extends ScalarFunction {
public Double eval(String value) {
try {
return Double.parseDouble(value);
} catch (Exception e) {
// Return a default value instead of failing the job.
return null;
}
}
}
```
### Python
```python
@udf(result_type=DataTypes.DOUBLE())
def safe_parse_double(value: str):
try:
return float(value)
except Exception:
# Return a default value instead of failing the job.
return None
```
### Use a flag column to route errors
If you need to identify which rows failed UDF processing, add a boolean column
to indicate errors. You can then filter on this column downstream to route
errors to a separate topic.
```java
public class DecryptField extends ScalarFunction {
// Returns ROW
public Row eval(String encrypted) {
try {
String result = decrypt(encrypted);
return Row.of(result, false);
} catch (Exception e) {
return Row.of(null, true);
}
}
}
```
Use the error flag in a query to route failed records to a separate topic:
```sql
-- Process successful records
INSERT INTO output_table
SELECT decrypted_value
FROM (SELECT decrypt_field(raw_data) AS result FROM source_table)
WHERE result.has_error = FALSE;
-- Route failed records to an error topic
INSERT INTO error_records
SELECT raw_data
FROM (SELECT raw_data, decrypt_field(raw_data) AS result FROM source_table)
WHERE result.has_error = TRUE;
```
### Handle external-call failures explicitly
UDFs that call external endpoints have failure modes that aren’t always
surfaced in the statement’s `Status Detail` field. Handle them inside the
UDF so the pipeline doesn’t drop rows silently. This pattern extends the
external-connection UDF shown in [Step 1: Build the artifact](#flink-sql-create-udf-create-artifact),
which initializes `httpClient`, `endpoint`, and `token` in the `open()`
method.
The following failure modes are common:
- **Platform timeout**: The platform terminates a batched invocation that
doesn’t complete in time, so one hung call can fail the records batched with
it. Set a request timeout (`.timeout()`) so slow calls fail predictably
instead of blocking the task thread up to the platform timeout. Set a short
`connectTimeout()` on the client so DNS and TCP problems fail fast. For how
call latency affects throughput, see
[External connectivity call latency and throughput](../concepts/user-defined-functions.md#flink-sql-udfs-external-connectivity-latency).
- **5xx responses**: Returned to the UDF as normal `HttpResponse` objects.
Check `response.statusCode()` and treat anything outside the 2xx range as
an error, rather than passing the body through unchanged.
- **DNS or network errors**: Cause `HttpClient.send()` to throw
`IOException` (for example, `UnknownHostException`). Catch these
explicitly so the UDF can emit a sentinel row instead of failing the
statement.
URL-encode any untrusted input concatenated into the request URI. Otherwise,
characters like spaces, `&`, or `#` cause `URI.create()` to throw
`IllegalArgumentException` or produce a malformed request.
The following pattern captures the status code and exception type in the output
row, which is then routable downstream using a flag column or a sentinel
prefix:
```java
public String eval(String id) {
try {
String encodedId = URLEncoder.encode(id, StandardCharsets.UTF_8);
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(this.endpoint + "?id=" + encodedId))
.header("Authorization", "Bearer " + this.token)
.timeout(Duration.ofSeconds(2))
.GET()
.build();
HttpResponse response =
this.httpClient.send(request, HttpResponse.BodyHandlers.ofString());
if (response.statusCode() >= 200 && response.statusCode() < 300) {
return "OK:" + response.body();
}
return "ERR:HTTP_" + response.statusCode();
} catch (java.net.http.HttpTimeoutException e) {
return "ERR:TIMEOUT";
} catch (java.net.UnknownHostException e) {
return "ERR:DNS:" + e.getMessage();
} catch (InterruptedException e) {
// Restore the interrupt flag so Flink can cancel the task cleanly.
Thread.currentThread().interrupt();
return "ERR:INTERRUPTED";
} catch (Exception e) {
return "ERR:" + e.getClass().getSimpleName() + ":" + e.getMessage();
}
}
```
Catch `InterruptedException` separately and restore the interrupt flag with
`Thread.currentThread().interrupt()`. Flink uses thread interrupts to cancel
tasks during savepoints and shutdown, so a UDF that swallows the interrupt can
prevent the task from stopping cleanly.
The sentinel-prefix approach pairs well with the
[flag-column pattern](#flink-sql-create-udf-flag-column): convert the
prefix to a `boolean` error column in a follow-on query so failed rows can be
routed to a separate topic.
## Related content
- [Enable UDF Logging](enable-udf-logging.md#flink-sql-enable-udf-logging)
- [confluent flink artifact create](https://docs.confluent.io/confluent-cli/current/command-reference/flink/artifact/confluent_flink_artifact_create.html)
- [CREATE FUNCTION Statement](../reference/statements/create-function.md#flink-sql-create-function)
- [Artifacts endpoints](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1))
- [Flink UDF Java Examples](https://github.com/confluentinc/flink-udf-java-examples)
#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).