# Redact a Column with a User-Defined Function This example builds a scalar Flink user-defined function (UDF), `mask`, that keeps the last few characters of a string and replaces the rest with asterisks. You upload it as an artifact, register it in an environment catalog, and call it in a query to redact an email column. For the reference material behind each step, see [User-Defined Functions in Confluent Manager for Apache Flink](../user-defined-functions.md#cmf-sql-udf). ## Prerequisites Complete the [examples setup](setup.md#cmf-sql-examples-setup) first. It provides the `test` environment, the `pool` compute pool, artifact management, and the JDK and Maven you use to build the function JAR. ## Build the function JAR The function extends `ScalarFunction` and needs only `flink-table-common`, which is already on the CMF and Flink classpath, so declare that dependency as `provided`. Compile for Java 11 to match the Flink cluster runtime. `MaskFunction.java`: ```java package io.confluent.examples.flink.udf; import org.apache.flink.table.functions.ScalarFunction; public class MaskFunction extends ScalarFunction { public String eval(String input, Integer visible) { if (input == null) { return null; } int keep = (visible == null || visible < 0) ? 0 : Math.min(visible, input.length()); int hide = input.length() - keep; return "*".repeat(hide) + input.substring(hide); } } ``` Package it with this `pom.xml`. Because `flink-table-common` is `provided`, the JAR carries only your class and needs no shading: ```xml 4.0.0 io.confluent.examples.flink udfs 1.0.0 jar UTF-8 11 1.19.1 org.apache.flink flink-table-common ${flink.version} provided udfs ``` Build it: ```bash mvn package ``` The build writes `target/udfs.jar`. ## Upload the JAR as an artifact A UDF JAR is referenced by a `cmf://` URI, so it does not need the connector or format labels. Put the metadata in `artifact.json`: ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "Artifact", "metadata": { "name": "udfs.jar" }, "spec": {} } ``` Upload it: ```bash curl -X POST http://cmf:8080/cmf/api/v1/environments/test/artifacts \ -F 'artifact=@artifact.json;type=application/json' \ -F 'file=@target/udfs.jar' ``` The response has `status.version` of `1`. ## Register the function Register `mask` in the environment catalog with `CREATE FUNCTION`, pointing `USING JAR` at the artifact. This is metadata only and completes immediately. ```bash confluent flink statement create udf-create-mask \ --url http://cmf:8080 \ --environment test \ --compute-pool pool \ --catalog _env_test \ --database default \ --sql "CREATE FUNCTION mask AS 'io.confluent.examples.flink.udf.MaskFunction' USING JAR 'cmf://test/udfs.jar';" ``` The statement reaches phase `COMPLETED` with `sqlKind` of `CREATE_FUNCTION`. ## Call the function Call `mask` like any built-in function. The unqualified name resolves only when `_env_test` and `default` are the current catalog and database, so start the SQL shell with the same `--catalog` and `--database` flags: ```bash confluent flink shell \ --url http://cmf:8080 \ --environment test \ --compute-pool pool \ --catalog _env_test \ --database default ``` This query redacts customer emails from the `examples.marketplace.customers` datagen table, keeping the last four characters: ```sql SELECT customer_id, email, mask(email, 4) AS masked_email FROM examples.marketplace.customers; ``` The statement runs on the compute pool, and the shell shows results such as: ```text customer_id email masked_email 3133 eufemia.jenkins@yahoo.com *********************.com 3130 anabel.sipes@gmail.com ******************.com 3006 billy.tillman@gmail.com *******************.com ``` ## Clean up The shell prints the statement name when it submits the query. For example, `cli-2026-10-19-120056-c98ff916`. Stop the query by deleting that statement, and drop the function if you no longer need it: ```bash confluent flink statement delete \ --url http://cmf:8080 --environment test confluent flink statement create udf-drop-mask \ --url http://cmf:8080 --environment test --compute-pool pool \ --catalog _env_test --database default \ --sql "DROP FUNCTION IF EXISTS mask;" ``` ## Related content - [User-Defined Functions in Confluent Manager for Apache Flink](../user-defined-functions.md#cmf-sql-udf) - [Manage Artifacts in Confluent Manager for Apache Flink](../../../configure/artifacts.md#cmf-artifacts) - [Flink SQL Examples in Confluent Manager for Apache Flink](index.md#cmf-sql-examples)