总览

简介

Load 节点列表是一组基于 Apache Flink® 的 Sink Connectors 用于将数据加载到不同的存储系统。

支持的 Load 节点列表

支持的 Flink 版本列表

下表展示了 InLong® Load 节点 和 Flink® 版本之间的对应关系。

SQL API 的用法

我们需要几个步骤来使用提供的连接器设置 Flink 集群。

• 设置一个安装了 1.13.5 版本和 Java 8+ 的 Flink 集群。
• 从 下载 页面下载并解压 Sort Connectors jars (或者参考 如何编译 编译需要的版本)。
• 将下载并解压后的 Sort Connectors jars 放到 FLINK_HOME/lib/。
• 重启 Flink 集群。

下面例子展示了如何在 Flink SQL Client 创建一个 MySQL Load 节点并加载数据进去：

-- 创建一个 MySQL Extract 节点
CREATE TABLE mysql_extract_node (
    id INT NOT NULL,
    name STRING,
    age INT,
    weight DECIMAL(10,3),
    PRIMARY KEY(id) NOT ENFORCED
) WITH (
      'connector' = 'mysql-cdc-inlong',
      'hostname' = 'YourHostname',
      'port' = '3306',
      'username' = 'YourUsername',
      'password' = 'YourPassword',
      'database-name' = 'YourDatabaseName',
      'table-name' = 'YourTableName'
      );
-- 创建一个 MySQL Load 节点
CREATE TABLE mysql_load_node (
 id INT NOT NULL,
 name STRING,
 age INT,
 weight DECIMAL(10,3),
 PRIMARY KEY(id) NOT ENFORCED
) WITH (
 'connector' = 'jdbc-inlong',
 'url' = 'jdbc:mysql://YourHostname:3306/YourDatabaseName',
 'username' = 'YourUsername',
 'password' = 'YourPassword',
 'table-name' = 'YourTableName'
);
INSERT INTO mysql_load_node SELECT id, name, age, weight FROM mysql_extract_node;