Overview

Overview

Load Nodes is a set of Sink Connectors based on Apache Flink® for loading data to different storage systems.

Supported Load Nodes

Load NodeVersionDriver
KafkaKafka: 0.10+None
HBaseHBase: 2.2.xNone
PostgreSQLPostgreSQL: 9.6, 10, 11, 12JDBC Driver: 42.2.12
OracleOracle: 11, 12, 19Oracle Driver: 19.3.0.0
MySQLMySQL: 5.6, 5.7, 8.0.x
RDS MySQL: 5.6, 5.7, 8.0.x
PolarDB MySQL: 5.6, 5.7, 8.0.x
Aurora MySQL: 5.6, 5.7, 8.0.x
MariaDB: 10.x
PolarDB X: 2.0.1
JDBC Driver: 8.0.21
TDSQL-PostgreSQLTDSQL-PostgreSQL: 10.17JDBC Driver: 42.2.12
GreenplumGreenplum: 4.x, 5.x, 6.xJDBC Driver: 42.2.12
ElasticsearchElasticsearch: 6.x, 7.xNone
ClickHouseClickHouse: 20.7+JDBC Driver: 0.3.1
HiveHive: 1.x, 2.x, 3.xNone
SQLServerSQLServer: 2012, 2014, 2016, 2017, 2019JDBC Driver: 7.2.2.jre8
HDFSHDFS: 2.x, 3.xNone
IcebergIceberg: 0.13.1+None

The following table shows the version mapping between InLong® Load Nodes and Flink®:

InLong® Load Nodes VersionFlink® Version
1.2.01.13.5

Usage for SQL API

We need several steps to setup a Flink cluster with the provided connector.

  1. Setup a Flink cluster with version 1.13.5 and Java 8+ installed.
  2. Download and decompress the Sort Connectors jars from the Downloads page (or build yourself).
  3. Put the Sort Connectors jars under FLINK_HOME/lib/.
  4. Restart the Flink cluster.

The example shows how to create a MySQL Load Node in Flink SQL Client and load data to it.

  1. -- Creates a MySQL Extract Node
  2. CREATE TABLE mysql_extract_node (
  3. id INT NOT NULL,
  4. name STRING,
  5. age INT,
  6. weight DECIMAL(10,3),
  7. PRIMARY KEY(id) NOT ENFORCED
  8. ) WITH (
  9. 'connector' = 'mysql-cdc-inlong',
  10. 'hostname' = 'YourHostname',
  11. 'port' = '3306',
  12. 'username' = 'YourUsername',
  13. 'password' = 'YourPassword',
  14. 'database-name' = 'YourDatabaseName',
  15. 'table-name' = 'YourTableName'
  16. );
  17. -- Creates a MySQL Load Node
  18. CREATE TABLE mysql_load_node (
  19. id INT NOT NULL,
  20. name STRING,
  21. age INT,
  22. weight DECIMAL(10,3),
  23. PRIMARY KEY(id) NOT ENFORCED
  24. ) WITH (
  25. 'connector' = 'jdbc-inlong',
  26. 'url' = 'jdbc:mysql://YourHostname:3306/YourDatabaseName',
  27. 'username' = 'YourUsername',
  28. 'password' = 'YourPassword',
  29. 'table-name' = 'YourTableName'
  30. );
  31. INSERT INTO mysql_load_node SELECT id, name, age, weight FROM mysql_extract_node;