从Flink SQL导入

更新时间: 2023-10-24 17:54:11

本文介绍如何使用Flink SQL方式将Flink中的数据写入云数据库ClickHouse

使用限制

只有Flink计算引擎VVR 3.0.2及以上版本支持使用Flink SQL写入云数据库ClickHouse

前提条件

  • 已在云数据库ClickHouse中创建表。更多信息,请参见创建表

  • 已在云数据库ClickHouse中设置白名单。更多信息,请参见设置白名单

  • 已开通Flink全托管。更多信息,请参见开通Flink全托管

操作步骤

  1. 登录Flink全托管控制台,新建作业。

    1. 登录实时计算控制台

    2. Flink全托管页签,单击目标工作空间操作列下的控制台

    3. SQL开发页面,单击新建

    4. 单击空白的流作业草稿

      Flink全托管为您提供了丰富的SQL代码模板和数据同步模板,每种模板都为您提供了使用场景、代码示例和使用指导。您也可以直接单击对应的模板进行使用,快速实现您的业务逻辑,详情请参见代码模板数据同步模板

    5. 单击下一步

    6. 新建文件对话框,填写作业配置信息。

      作业参数

      说明

      文件名称

      作业的名称。

      说明

      作业名称在当前项目中必须保持唯一。

      存储位置

      指定该作业的存储位置。

      您还可以在现有文件夹右侧,单击新建文件夹图标,新建子文件夹。

      引擎版本

      当前作业使用的Flink的引擎版本。引擎版本号含义、版本对应关系和生命周期重要时间点详情请参见引擎版本介绍

    7. 单击创建

  2. 编写并执行代码。

    1. 创建源表、结果表,并将源表数据插入到结果表中。

      --创建源表sls_test_single_local。
      CREATE TEMPORARY TABLE sls_test_single_local (
        id INT,
        name VARCHAR,
        age BIGINT,
        rate FLOAT
      ) WITH (
        'connector' = 'datagen',
        'rows-per-second' = '50'
      );
      --创建结果表clickhouse_output。
      CREATE TEMPORARY TABLE clickhouse_output (
        id INT,
        name VARCHAR,
        age BIGINT,
        rate FLOAT
      ) WITH (
        'connector' = 'clickhouse',
        'url' = 'jdbc:clickhouse://demo.aliyuncs.com:8123',
        'userName' = 'test',
        'password' = '280226Ck',
        'tableName' = 'sls_test_single_local'
      );
      --将源表数据插入到结果表。
      INSERT INTO clickhouse_output
      SELECT 
        id,
        name,
        age,
        rate
      FROM sls_test_single_local;
      说明

      创建结果表的语法说明,请参见ClickHouse结果表

    2. 单击保存

    3. 单击验证

    4. 单击上线

    5. 在弹出的上线确认窗口中单击确认,将作业发布至生产环境。

    6. 上线成功后,在弹出的上线成功,请前往运维查看详情中单击运维

  3. 作业运维页面,启动作业。

    1. 单击右上方启动

    2. 在弹出的作业启动配置页面,单击确认启动

  4. 云数据库ClickHouse中查询表。

    1. 登录云数据库ClickHouse控制台

    2. 在页面左上角,选择目标集群所在的地域。

    3. 集群列表页面,选择目标集群对应类型的实例列表,单击目标集群ID。

    4. 集群信息页面,单击右上方导航栏的登录数据库

    5. 登录实例页面,输入数据库账号和密码,单击登录

    6. 输入查询语句并单击执行(F8)。本文使用下面的示例语句。

      select * from db01.sls_test_single_local;

      查询结果如下。flink数据导入ClickHouse表

上一篇: 使用JDBC Connector导入 下一篇: 从大数据导入/同步
阿里云首页 云数据库 ClickHouse 相关技术圈