Deployment

InLong Sort is based on Apache Flink, you need to set up an Apache Flink Environment.

InLong Sort relies on Apache Flink 1.13.5 or 1.15.4. Chose flink-1.13.5-bin-scala_2.11.tgz or flink-1.15.4-bin-scala_2.12.tgz when downloading package.

Prepare installation files

  • InLong Sort file, Download apache-inlong-[version]-bin.tar.gz
  • Data Nodes Connectors, Download apache-inlong-[version]-sort-connectors.tar.gz

Deployment - 图1caution

Please put required Connectors jars into under FLINK_HOME/lib/ after download.
Put mysql-connector-java:8.0.21.jar to FLINK_HOME/lib/ when you use mysql-cdc-inlong connector.
If you change the Apache Flink version, you also need to change the connectors of the corresponding version.

Start an InLong Sort Job

  1. ./bin/flink run -c org.apache.inlong.sort.Entrance apache-inlong-[version]-bin/inlong-sort/sort-dist-[version].jar \
  2. --sql.script.file [souce-to-sink].sql

Deployment - 图2note

--sql.script.file add a SQL script file includes multi Flink SQL statements that can be separated by semicolon, support CREATE TABLE, CRETAE VIEW, INSERT INTO etc.

MySQL to PostgreSQL

We can write following SQL script if we want to read data from MySQL and write into PostgreSQL.

  • Prepare mysql-to-postgresql.sql
  1.  CREATE TABLE `table_1`(
  2.     `age` INT,
  3.     `name` STRING)
  4.     WITH (
  5.     'connector' = 'mysql-cdc-inlong',
  6.     'hostname' = 'localhost',
  7.     'port' = '3306',
  8.     'username' = 'root',
  9.     'password' = 'inlong',
  10.     'database-name' = 'test',
  11.     'scan.incremental.snapshot.enabled' = 'false',
  12.     'server-time-zone' = 'GMT+8',
  13.     'table-name' = 'user'
  14. );
  15. CREATE TABLE `table_2`(
  16.     PRIMARY KEY (`name`) NOT ENFORCED,
  17.     `name` STRING,
  18.     `age` INT)
  19.     WITH (
  20.     'connector' = 'jdbc',
  21.     'url' = 'jdbc:postgresql://localhost:5432/postgres',
  22.     'username' = 'postgres',
  23.     'password' = 'inlong',
  24.     'table-name' = 'public.user'
  25. );
  26. INSERT INTO `table_2` 
  27.     SELECT 
  28.     `name` AS `name`,
  29.     `age` AS `age`
  30.     FROM `table_1`;
  • Summit job
  1. ./bin/flink run -c org.apache.inlong.sort.Entrance apache-inlong-[version]-bin/inlong-sort/sort-dist-[version].jar \
  2. --sql.script.file mysql-to-postgresql.sql

其它完整使用示例，可以参考 Example