Overview

Overview

Extract Nodes is a set of Source Connectors based on Apache Flink® for extracting data from different source systems.

Supported Extract Nodes

Extract NodeVersionDriver
KafkaKafka: 0.10+None
PulsarPulsar: 2.8.x+None
MongoDB-CDCMongoDB: 3.6, 4.x, 5.0None
MySQL-CDCMySQL: 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
Oracle-CDCOracle: 11, 12, 19Oracle Driver: 19.3.0.0
PostgreSQL-CDCPostgreSQL: 9.6, 10, 11, 12JDBC Driver: 42.2.12
SqlServer-CDCSQLServer: 2012, 2014, 2016, 2017, 2019None

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

InLong® Extract 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 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 Extract Node in Flink SQL Client and execute queries on 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. );
  18. SELECT id, name, age, weight FROM mysql_extract_node;