Migrate Table/SQL & DataStream JAR jobs-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

This topic describes how to migrate JAR jobs from an open source, self-managed Flink cluster to Realtime Compute for Apache Flink.

Background information

Note

This topic provides a sample JAR job. For more information, see the flink2vvp project on GitHub.

This job uses a business scenario that calculates the total number and amount of orders within a 5-minute window. It consumes data from ApsaraMQ for Kafka and writes the results to ApsaraDB RDS. You can deploy the job to Realtime Compute for Apache Flink using SQL or a JAR file. The following figure shows the migration scenario.

Prerequisites

You have set up the migration environment. For more information, see Set up the basic environment.

Notes

Set the dependency scope for non-connector dependencies in the job to `provided` to avoid conflicts with cloud dependencies.
State data from jobs cannot be migrated.

When using multiple connectors, merge the META-INF directories. To do this, add the following code to the maven-shade-plugin plug-in in the pom.xml file:

<transformers>
  <!-- The service transformer is needed to merge META-INF/services files -->
  <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
  <transformer implementation="org.apache.maven.plugins.shade.resource.ApacheNoticeResourceTransformer">
    <projectName>Apache Flink</projectName>
    <encoding>UTF-8</encoding>
  </transformer>
</transformers>

Preparations

Create an ApsaraDB RDS for MySQL database and table

Create a database.
Create the test_db sample database for the target instance.

Create the sink table rds_new_table.

On the details page for the target instance, click Log On To Database.

Double-click the target database and create a sink table in the SQL execution window.

In this topic, you will create a table in the test_db database.

CREATE TABLE `rds_new_table`
( 
  `window_start` timestamp NOT NULL, 
  `window_end` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `order_type` varchar(8) NOT NULL,
  `order_number` bigint NULL,
  `order_value_sum` double NULL,
   PRIMARY KEY ( `window_start`, `window_end`, `order_type` )
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;

Click Execute, and then click Execute Directly.

Prepare Kafka test data

Create a topic named kafka-order and a group named demo-group. For more information, see Create a topic and a group.

Write DDL and DML statements to write test data to Kafka.

In the Realtime Compute for Apache Flink console, find the target workspace and click Console in the Operation column to open the project.

On the Data Studio > ETL page, click New, select Blank Stream Draft, and then copy the following code to the SQL editor.

CREATE TEMPORARY TABLE data_in (
  id VARCHAR, 
  order_value FLOAT
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '100',
  'fields.id.length' = '10',
  'fields.order_value.min' = '1.0',
  'fields.order_value.max' = '100.0'
);
CREATE TEMPORARY TABLE kafka_order (
  order_id VARCHAR,
  order_time TIMESTAMP,
  order_type VARCHAR,
  order_value FLOAT
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafka-order',
  'properties.bootstrap.servers' = 'alikafka-pre-cn-0dw3yxao****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-0dw3yxao****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-0dw3yxao****-3-vpc.alikafka.aliyuncs.com:9092',
  'format' = 'csv'
);
INSERT INTO kafka_order
SELECT id as order_id,
  CURRENT_TIMESTAMP as order_time,
  CASE
    WHEN (substring (id, 1, 1) <= 'z' AND substring (id, 1, 1) >= 'a') and (substring (id, 2, 1) <= 'z' AND substring (id, 2, 1) >= 'a') THEN 'typeA'
    WHEN (substring (id, 1, 1) <= 'z' AND substring (id, 1, 1) >= 'a') and (substring (id, 2, 1) <= '9' AND substring (id, 2, 1) >= '0') THEN 'typeB'
    WHEN (substring (id, 1, 1) <= '9' AND substring (id, 1, 1) >= '0') and (substring (id, 2, 1) <= '9' AND substring (id, 2, 1) >= '0') THEN 'typeC'
    ELSE 'typeD'
  END as order_type,
  order_value
FROM
  data_in;

This topic uses the datagen connector to randomly generate data and write it to Kafka in real time. For more information about WITH parameters, see Datagen and ApsaraMQ for Kafka.

In the upper-right corner, click Deploy to deploy the job.
In the navigation pane on the left, choose Operation Center > Job O&M. Find the target job and click Start in the Operation column. Select Stateless Start and then click Start.

Migrate a self-managed Flink job to Alibaba Cloud Flink

You can migrate a self-managed Flink job by deploying an SQL job or a JAR job in Realtime Compute for Apache Flink. This topic describes how to use built-in connectors to create a Flink SQL or JAR job, migrate a self-managed Flink job, and validate the migration result.

Migrate the job.

Migrate using the SQL method

Extract the SQL statements from the Table/SQL code and deploy them as an SQL job.

In the Realtime Compute for Apache Flink console, find the target workspace and click Console in the Operation column to open the project.
In the navigation pane on the left, choose Data Studio > ETL.
Click , and then click New Stream Job. Enter a File Name, select a Database Engine Version, and click Create.

Write DDL and DML statements.

Extract the SQL statements from the TableJobKafka2Rds.java code in flink2vvp. The following code is a sample.

CREATE TEMPORARY TABLE kafkatable (
  order_id varchar,
  order_time timestamp (3),
  order_type varchar,
  order_value float,
  WATERMARK FOR order_time AS order_time - INTERVAL '2' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafka-order',
  'properties.bootstrap.servers' = 'alikafka-pre-cn-0dw3yxao****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-0dw3yxao****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-0dw3yxao****-3-vpc.alikafka.aliyuncs.com:9092',
  'properties.group.id' = 'demo-group',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE rdstable (
  window_start timestamp,
  window_end timestamp,
  order_type varchar,
  order_number bigint,
  order_value_sum double,
  PRIMARY KEY (window_start, window_end, order_type) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = 'rm-**********.mysql.rds.aliyuncs.com',
  'port'='3306',
  'database-name' ='test_db',
  'table-name'= 'rds_new_table',
  'username' = 'flink****',
  'password' = '${secret_values.rdspw}'
);

INSERT INTO rdstable
SELECT TUMBLE_START (order_time, INTERVAL '5' MINUTE) as window_start,
  TUMBLE_END (order_time, INTERVAL '5' MINUTE) as window_end,
  order_type,
  COUNT (1) as order_number,
  SUM (order_value) as order_value_sum
FROM
  kafkatable
GROUP BY TUMBLE (order_time, INTERVAL '5' MINUTE), order_type;

Modify the Kafka and ApsaraDB RDS parameter settings.

Category	Parameter	Description
Kafka	topic	The name of the Kafka topic. This example uses kafka-order.
	properties.bootstrap.servers	The endpoint of the Kafka broker. The format is `host:port,host:port,host:port`. Separate multiple endpoints with commas (,). You can view the endpoint on the Access Point Information section of the ApsaraMQ for Kafka instance product page. For more information, see View endpoints.
	properties.group.id	The ID of the Kafka consumer group. This example uses demo-group. Note To avoid consumer group ID conflicts, create a new consumer group in the Kafka console and use the new ID here.
RDS	connector	Use the built-in MySQL connector. The value is fixed to mysql. It supports all databases that are compatible with the MySQL protocol, such as ApsaraDB RDS for MySQL, PolarDB for MySQL, or self-managed MySQL. For more information, see MySQL.
	hostname	The IP address or hostname of the MySQL database. We recommend that you enter the VPC endpoint. The VPC endpoint of ApsaraDB RDS is a private network endpoint. For more information, see View and manage instance endpoints and ports.
	port	The port number of the MySQL database service.
	database-name	The name of the MySQL database. This example uses test_db.
	table-name	The name of the MySQL table. This example uses rds_new_table.
	username	The username for the MySQL database service.
	password	The password for the MySQL database service. To avoid exposing passwords and other information in plaintext, this topic uses variables. For more information, see Manage variables.

(Optional) If you use an open source connector JAR package or other related dependencies, click More Configurations on the right side of the page and upload the files in the Additional Dependency Files section.
Note
The dependency file version must match the engine version of the Flink job.
(Optional) In the upper-right corner of the page, click Validate and Debug to check the syntax and validate the job logic.
The debug feature simulates job execution, which lets you verify output results and validate the logic of SELECT or INSERT statements. However, you must create a session cluster. For more information, see Debug a job.
In the upper-right corner of the page, click Deploy.

For more information about the SQL job workflow, see Quick Start for Flink SQL jobs.

Migrate using the Table/SQL JAR method

Compile and package the Table/SQL code to generate a JAR file, and then deploy the JAR file.

Modify the following configuration information in Maven and build a new JAR file.

In IntelliJ IDEA, choose File > Open, and open the downloaded and decompressed flink2vvp-main package.

Double-click to open TableJobKafka2Rds and modify the WITH parameters for the built-in Kafka and ApsaraDB RDS connectors.

Category	Parameter	Description
Kafka	topic	The name of the Kafka topic. This example uses kafka-order.
	properties.bootstrap.servers	The endpoint of the Kafka broker. The format is `host:port,host:port,host:port`. Separate multiple endpoints with commas (,). You can view the endpoint on the Access Point Information section of the ApsaraMQ for Kafka instance product page. For more information, see View endpoints.
	properties.group.id	The ID of the Kafka consumer group. This example uses demo-group. Note To avoid consumer group ID conflicts, create a new consumer group in the Kafka console and use the new ID here.
RDS	connector	Use the built-in MySQL connector. The value is fixed to mysql. It supports all databases that are compatible with the MySQL protocol, such as ApsaraDB RDS for MySQL, PolarDB for MySQL, or self-managed MySQL. For more information, see MySQL.
	hostname	The IP address or hostname of the MySQL database. We recommend that you enter the VPC endpoint. The VPC endpoint of ApsaraDB RDS is a private network endpoint. For more information, see View and manage instance endpoints and ports.
	port	The port number of the MySQL database service.
	database-name	The name of the MySQL database. This example uses test_db.
	table-name	The name of the MySQL table. This example uses rds_new_table.
	username	The username for the MySQL database service.
	password	The password for the MySQL database service. To avoid exposing passwords and other information in plaintext, this topic uses variables. For more information, see Manage variables.

Modify the related configuration information in the pom.xml file.
Note
The pom dependency in the downloaded code sample is for Flink 1.13. You must modify the pom information to match your job's Flink version.
Add Properties
Add the <vvr.version> tag. The version number must be the same as the engine version of the JAR job that you want to deploy.
Modify connector dependencies
Modify the connector dependencies to use the built-in Kafka and MySQL connectors.
```
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-kafka</artifactId>
    <version>${vvr.version}</version>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>
```
Modify build configurations
Add the <excludes> section within the <configuration> tag to ignore the DataStreamJobKafka2Rds.java source file during the build.
```
<excludes>
    <exclude>**/DataStreamJobKafka2Rds.java</exclude>
</excludes>
```
Run the mvn clean package command to build the JAR file.
After the build is successful, you can find the corresponding JAR file in the target directory.

Deploy the JAR job.

In the Realtime Compute for Apache Flink console, find the target workspace and click Console in the Operation column to open the project.
In the Operation Center > Job O&M page, click Deploy Job > JAR Job.

Enter the deployment information. For more information about the parameters, see Deploy a JAR job.

Configuration Item	Description
Deployment Name	A custom job name.
Database Engine Version	Must be the same as the version of the dependency files.
JAR URI	Upload the newly compiled JAR package or enter the corresponding JAR information.
Entry Point Class	Specify the main class name as com.alibaba.realtimecompute.TableJobKafka2Rds.
Additional Dependency Files	This topic uses built-in connectors, so you do not need to upload dependency files. If you use an open source connector JAR package or other related dependencies, upload them here.

Click Deploy.

For more information about the JAR job workflow, see Quick Start for Flink JAR jobs.

Migrate using the DataStream JAR method

Compile and package the DataStream code to generate a JAR file, and then deploy the JAR file.