Migrating flows to Confluent Platform for Apache Flink

This topic covers the migration tools and steps for migrating flows created in the Event Processing low-code canvas to Confluent Platform for Apache Flink.

Note: Confluent provides two methods to create and manage Flink jobs, either as applications or SQL statements. In this topic, your Flink job from Event Processing is migrated to be executed as a Flink application.

Overview

IBM announced the Support lifecycle transition and software ordering completion for IBM Event Automation, stating that IBM intends to provide migration tools, services, and entitlement flexibility to assist with migrating Event Streams and Event Processing deployments to IBM Confluent Platform.

The following table describes the methods available in Event Processing for creating Flink workloads and their migration process:

Resources to assist with migrating your flows are available in the IBM Event Automation GitHub repository.

Limitations

• Event Processing flows that use the detect patterns node or the deduplicate node cannot be migrated directly to Confluent Platform for Apache Flink.

• In many cases, pattern detection nodes can be replaced with a Free SQL node that uses Flink’s MATCH_RECOGNIZE syntax. In some cases, it is necessary to rewrite those flows as Flink Java applications that use the Complex Event Processing API.
• Deduplicate nodes can be replaced with a Free SQL node that uses Flink’s ROW_NUMBER() AS rownum ... WHERE rownum=1 syntax (see the Flink SQL cookbook).

Prerequisites

This migration process applies to Event Processing flows that have at least one connected destination node. Flows that contain the detect patterns node cannot be migrated by using this process.

Before you begin, complete the following steps:

1. Export each of your flows from Event Processing as an SQL file.
2. Restore redacted credentials and amend the SQL for the target environment.
3. Install Confluent Manager for Apache Flink and Confluent for Kubernetes in your target environment.

How deployed flows are structured

Flows deployed in Event Processing usually consists of:

• An image containing the Apache Flink runtime, and custom Event Processing libraries that support Event Processing flows.
• The exported flow.sql or flow.json file.
• The config.yaml file (only available if the flow was exported as JSON + Configuration YAML).

Migrating to Confluent Platform for Apache Flink changes the following:

• The Java dependencies used at compilation and runtime.
• The base Apache Flink image.
• The source of Flink connectors.
• The source of user-defined functions (UDFs) and process table functions (PTFs).
• The deployment mechanism.

The flow SQL exported from Event Processing requires custom libraries at runtime:

• CREATE TABLE statements use the TO_TIMESTAMP_LTZ UDF. This function smoothes over timestamp processing differences between the SQL and ISO-8601 standards, and between the Avro and JSON formats.

Migrated flows therefore assemble the following components:

• Confluent’s Apache Flink runtime image.
• A JAR file containing the TO_TIMESTAMP_LTZ UDF and the deduplicate node PTF.
• The exported flow.sql file.
• Connectors supported by Confluent, from Confluent’s Maven repository.
• Connectors not supported by Confluent, either from IBM or the Apache Flink project.
• An Apache Flink SQL Runner JAR file.

Step 1: Build the application Docker image

Your exported flow must be packaged as a custom docker image as described in the Confluent documentation.

A Dockerfile is provided that packages all required components. It compiles the Apache Flink SQL Runner JAR file with Confluent Maven coordinates, downloads the required Confluent connectors and the IBM UDF+PTF JAR file, and assembles everything alongside the exported flow.sql into a single deployable image.

Run the following command to build the image:

docker build --build-arg FLOW_SQL_PATH=/path/to/exported.sql -t my/image -f /path/to/Dockerfile

After building the image, push it to a registry accessible from your Kubernetes cluster. If your registry requires authentication, configure the image pull secret. For example, in OpenShift, you can use the global cluster pull secret.

The following sections describe the components of the image in more detail.

New application base: Apache Flink SQL runner

The Flink Kubernetes Operator SQL Example forms the base for migrated applications. To use the Flink dependencies provided as part of Confluent Platform for Apache Flink, you must add the Confluent Maven repository and change the Maven group ID and version for supported components in the POM file for your project, as described in the Confluent documentation.

The provided Dockerfile follows this approach: the Apache Flink Operator repository is cloned, the Maven coordinates are patched to point to Confluent’s libraries, and the application is compiled to produce the SQL runner JAR file that serves as the entry point for the migrated application.

Common dependencies

The following dependencies are required by all migrated flows. See the Confluent documentation for more information.

Connectors from Confluent

Confluent-supported connectors are listed in the Confluent documentation.

Note: All connectors and supporting libraries must be stored in /opt/flink/lib in the packaged deployment.

Connectors from open source

Event Processing flows that use the API enrichment or watsonx nodes require an HTTP connector:

• Event Processing 1.5.5 and later uses the Apache Flink HTTP connector at org.apache.flink:flink-sql-connector-http. For more information, see the GitHub repository and Maven Central.
• Flows that were deployed with Event Processing versions earlier than 1.5.5 must use com.getindata:flink-http-connector. For more information, see the GitHub repository and Maven Central.

Step 2: Migrate the state of your Flink deployment

To preserve the state of a running Flink deployment, stop the Flink deployment and take a savepoint before migrating.

Note: If your source and target deployments share a Kubernetes namespace, you can skip this step and reuse the original persistent volume. However, the IBM Operator for Apache Flink and the Confluent Platform for Apache Flink Operator cannot be configured to manage resources in the same namespace because both operators manage the same custom resources. Before installing the Confluent Platform for Apache Flink Operator, you must delete or suspend all existing FlinkDeployment custom resources and uninstall the IBM Operator for Apache Flink.

Stop the Flink deployment and take a savepoint

To stop the Flink deployment and capture a savepoint at the same time, update the spec.job.state field in the associated FlinkDeployment custom resource:

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
...
spec:
  job:
    state: suspended
    upgradeMode: savepoint

A savepoint is created and the job enters a suspended state. A new entry appears in the FlinkDeployment status, for example:

status:
  jobStatus:
    upgradeSavepointPath: file:/opt/flink/volume/flink-sp/savepoint-e574c6-638b6089cdd2

A FlinkStateSnapshot custom resource is also created that contains the same information. For example:

apiVersion: flink.apache.org/v1beta1
kind: FlinkStateSnapshot
metadata:
  labels:
    snapshot.state: COMPLETED
    snapshot.trigger-type: UPGRADE
    snapshot.type: SAVEPOINT
  name: <flink-deployment-name>-upgrade-1782304978581
spec:
  jobReference:
    kind: FlinkDeployment
    name: <flink-deployment-name>
  savepoint:
    alreadyExists: true
    disposeOnDelete: false
    formatType: CANONICAL
    path: file:/opt/flink/volume/flink-sp/savepoint-e574c6-638b6089cdd2
status:
  state: COMPLETED

Make a note the value of the spec.savepoint.path field. This file, or a copy of it, must be accessible to the migrated application.

Copy the savepoint to the target environment

In most cases, you need to copy the savepoint files from the persistent volume mounted in the original deployment to a persistent volume available in the target namespace.

A copy-savepoint.sh script is provided for this purpose. The script uses a temporary pod in each namespace to stage the copy through local disk, and works with any storage class and access mode.

# copy-savepoint.sh — copy Flink state from a PVC in one namespace to a PVC in another.
#
# Usage:
#   ./copy-savepoint.sh [options]
#
# Options (all optional — omitted values are prompted interactively):
#   --src-namespace  <ns>    Namespace containing the source PVC (e.g. event-automation)
#   --src-pvc        <pvc>   Source PVC name
#   --src-path       <path>  Path inside the source PVC to copy (e.g. /flink-sp)
#   --dst-namespace  <ns>    Namespace containing the destination PVC (e.g. confluent)
#   --dst-pvc        <pvc>   Destination PVC name
#   --dst-path       <path>  Path inside the destination PVC to write to (default: same as --src-path)
#   --dry-run                Show what would be done without executing
#
# Example:
#   ./copy-savepoint.sh \
#     --src-namespace event-automation --src-pvc basic-datagen --src-path /flink-sp \
#     --dst-namespace confluent        --dst-pvc flink-state
#
# Requires: kubectl (or oc), bash, docker, and sufficient local disk for a temporary copy of the data.

Step 3: Deploy the migrated application

Follow the Confluent Platform for Apache Flink documentation to deploy your migrated application.

Use the deploy.sh script to deploy your Flink job as an application. This script uses the Confluent CLI to deploy a FlinkApplication custom resource into the target Confluent Manager for Apache Flink environment.

# deploy.sh — deploy a migrated IBM Event Processing flow to Confluent Platform for Apache Flink, using Confluent for Kubernetes
#
# Usage:
#   ./deploy.sh [options]
#
# Options (all optional — omitted values are prompted interactively):
#   --name                <name>       Application name (Kubernetes resource name)
#   --image               <image>      Docker image containing sql-runner.jar, e.g. myregistry.io/flink-runner:v1
#   --sql                 <file>       Path to the exported flow.sql file
#   --namespace           <namespace>  Kubernetes namespace to deploy into
#   --flink-env           <env>        FlinkEnvironment custom resource name
#   --pvc                 <pvc>        PersistentVolumeClaim name for Flink state storage
#   --savepoint           <path>       Savepoint path for state migration (e.g. file:///opt/flink/...)
#                                      Omit for a fresh deployment (no state restore).
#   --cmfrestclass        <name>       CMFRestClass custom resource name (default: "default")
#   --cmfrestclass-namespace <ns>      Namespace of the CMFRestClass custom resource (default: same as --namespace)
#   --dry-run                          Print the rendered YAML instead of submitting it.

The SQL from the exported application is stored in a Kubernetes Secret and mounted into the Flink pods at runtime. For more information, see the Confluent for Kubernetes documentation.