例子

示例

为了更容易创建 InLong Sort 作业，这里我们列出了一些数据流配置示例。下面将介绍 InLong Sort 的 SQL、Dashboard、Manager 客户端工具的使用。

环境要求

  • JDK 1.8.x
  • Flink 1.13.5
  • MySQL
  • Kafka
  • Hadoop
  • Hive 3.x

准备 InLong Sort 和 Connectors

你可以通过参考部署指引准备 InLong Sort 和数据节点 Connectors。

使用 SQL API 方式

示例构建了 MySQL —> Kafka —> Hive 的数据流，为了便于理解流程执行过程进行了拆解。

读 MySQL 写 Kafka

单表同步配置示例如下：

  1. ./bin/flink run -c org.apache.inlong.sort.Entrance FLINK_HOME/lib/sort-dist-[version].jar \
  2. --sql.script.file /YOUR_SQL_SCRIPT_DIR/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 写 Hive

注意: 首先需要在 hive 中创建 user 表。

  1. ./bin/flink run -c org.apache.inlong.sort.Entrance FLINK_HOME/lib/sort-dist-[version].jar \
  2. --sql.script.file /YOUR_SQL_SCRIPT_DIR/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`;

备注：以上过程所有的 SQL 可以放在一个文件中提交执行。

使用 Inlong Dashboard 方式

目前 Dashboard 支持文件采集同步的方式，以上数据源可视化配置方式正在开发中。

使用 Manager Client Tools 方式

TODO: 未来发布的版本将会支持。