Example

To make it easier for you to create InLong Sort jobs, here we list some data stream configuration examples. The following will introduce SQL, Dashboard, Manager Client Tools methods to use Inlong Sort.

Environment Requirements

  • Apache Flink 1.13.5
  • MySQL
  • Apache Kafka
  • Apache Hadoop
  • Apache Hive 3.x

Prepare InLong Sort And Connectors

You can prepare InLong Sort and Data Node Connectors by referring to Deployment Guide.

Usage for SQL API

This example defines the data flow for a single table(mysql—>kafka—>hive).

MySQL to Kafka

Single table sync example:

  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-kafka.sql
  • mysql-to-kafka.sql
  1. CREATE TABLE `table_1`(
  2.     PRIMARY KEY (`id`) NOT ENFORCED,
  3.     `id` BIGINT,
  4.     `name` STRING,
  5.     `age` INT,
  6.     `salary` FLOAT,
  7.     `ts` TIMESTAMP(2),
  8.     `event_type` STRING)
  9.     WITH (
  10.     'append-mode' = 'true',
  11.     'connector' = 'mysql-cdc-inlong',
  12.     'hostname' = 'localhost',
  13.     'username' = 'root',
  14.     'password' = 'password',
  15.     'database-name' = 'dbName',
  16.     'table-name' = 'tableName'
  17. );
  19. CREATE TABLE `table_2`(
  20.     `id` BIGINT,
  21.     `name` STRING,
  22.     `age` INT,
  23.     `salary` FLOAT,
  24.     `ts` TIMESTAMP(2))
  25.     WITH (
  26.     'topic' = 'topicName',-- Your kafka topic
  27.     'properties.bootstrap.servers' = 'localhost:9092',
  28.     'connector' = 'kafka',
  29.     'json.timestamp-format.standard' = 'SQL',
  30.     'json.encode.decimal-as-plain-number' = 'true',
  31.     'json.map-null-key.literal' = 'null',
  32.     'json.ignore-parse-errors' = 'true',
  33.     'json.map-null-key.mode' = 'DROP',
  34.     'format' = 'json',
  35.     'json.fail-on-missing-field' = 'false'
  36. );
  38. INSERT INTO `table_2` 
  39.     SELECT 
  40.     `id` AS `id`,
  41.     `name` AS `name`,
  42.     `age` AS `age`,
  43.     CAST(NULL as FLOAT) AS `salary`,
  44.     `ts` AS `ts`
  45.     FROM `table_1`;

Kafka to Hive

Example - 图1caution

First you need to create user table in Hive.

  1. ./bin/flink run -c org.apache.inlong.sort.Entrance apache-inlong-[version]-bin/inlong-sort/sort-dist-[version].jar \
  2. --sql.script.file kafka-to-hive.sql
  • kafka-to-hive.sql
  1. CREATE TABLE `table_1`(
  2.     `id` BIGINT,
  3.     `name` STRING,
  4.     `age` INT,
  5.     `salary` FLOAT,
  6.     `ts` TIMESTAMP(2)
  7.     WITH (
  8.     'topic' = 'topicName',-- Your kafka topic
  9.     'properties.bootstrap.servers' = 'localhost:9092',
  10.     'connector' = 'kafka',
  11.     'scan.startup.mode' = 'earliest-offset',
  12.     'json.timestamp-format.standard' = 'SQL',
  13.     'json.encode.decimal-as-plain-number' = 'true',
  14.     'json.map-null-key.literal' = 'null',
  15.     'json.ignore-parse-errors' = 'true',
  16.     'json.map-null-key.mode' = 'DROP',
  17.     'format' = 'json',
  18.     'json.fail-on-missing-field' = 'false',
  19.     'properties.group.id' = 'groupId'-- Your group id
  20. );
  22. CREATE TABLE `user`(
  23.     `id` BIGINT,
  24.     `name` STRING,
  25.     `age` INT,
  26.     `salary` FLOAT,
  27.     `ts` TIMESTAMP(9))
  28.     WITH (
  29.     'connector' = 'hive',
  30.     'default-database' = 'default',
  31.     'hive-version' = '3.1.2',
  32.     'hive-conf-dir' = 'hdfs://ip:9000/.../hive-site.xml' -- Put your hive-site.xml into HDFS
  33. );
  35. INSERT INTO `user` 
  36.     SELECT 
  37.     `id` AS `id`,
  38.     `name` AS `name`,
  39.     `age` AS `age`,
  40.     CAST(NULL as FLOAT) AS `salary`,
  41.     `ts` AS `ts`
  42.     FROM `table_1`;

Other Connectors

there are lots of supported Extract Node and Load Node , you can use them directly.