使用FlinkSQL将数据写入到ClickHouse

缺乏、安全感 2022-11-13 00:57 1011阅读 0赞

# 使用FlinkSQL将地区主题数据写入到ClickHouse #

## 一、需求分析与思路 ##

1.  定义Table流环境
2.  把数据源定义为动态表
3.  通过SQL查询出结果表
4.  把结果表转换为数据流
5.  把数据流写入目标数据库

如果是Flink官方支持的数据库，也可以直接把目标数据表定义为动态表，用insert into 写入。由于ClickHouse目前官方没有支持的jdbc连接器（目前支持Mysql、 PostgreSQL、Derby）。

阿里云有实现好的connector, 我们使用这个connector.参考地址: [https://help.aliyun.com/document\_detail/185696.html?spm=a2c4g.11186623.6.574.d9c541ea3J78mc][https_help.aliyun.com_document_detail_185696.html_spm_a2c4g.11186623.6.574.d9c541ea3J78mc]

## 二、功能实现 ##

数据来源于topic: dwm\_order\_wide

### 2.1 导入ClickHouse连接器 ###

<dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>flink-connector-clickhouse</artifactId>
        <version>1.11.0</version>
    </dependency>

**注意:**

1.  由于该连接器目前在远程的maven仓库中找不到, 我们需要下载该连接器, 然后安装到本地仓库使用
2.  下载地址: https://clickhouse-release-open-access.oss-cn-shanghai.aliyuncs.com/doc-data/flink-connector-clickhouse-1.11.0.jar?spm=a2c4g.11186623.2.6.d9c541ea3J78mc&file=flink-connector-clickhouse-1.11.0.jar
3.  假设jar包被下载在如下目录: C:\\Users\\lzc\\Desktop\\connector
4.  安装jar到本地仓库(需要保证mvn命令已经配置到了path中):

mvn install:install-file -Dfile=C:\Users\lzc\Desktop\connector\flink-connector-clickhouse-1.11.0.jar
     -DgroupId=com.aliyun -DartifactId=flink-connector-clickhouse -Dversion=1.11.0 -Dpackaging=jar

### 2.2 在ClickHouse中创建表 ###

use gmall2021;
    create table province_stats_2021 (
       stt DateTime,
       edt DateTime,
       province_id  UInt64,
       province_name String,
       area_code String ,
       iso_code String,
       iso_3166_2 String , 
       order_amount Decimal64(2),
       order_count UInt64, 
       ts UInt64
    )engine =ReplacingMergeTree( ts)
            partition by  toYYYYMMDD(stt)
            order by   (stt,edt,province_id );

### 2.3 具体实现代码 ###

package com.gmall.realtime.app.dws;
    
    import com.gmall.realtime.app.BaseSqlAPP;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    public class DWSProvinceStateApp extends BaseSqlAPP { 
        public static void main(String[] args) { 
            new DWSProvinceStateApp().init(30003,2,"DWSProvinceStateApp");
        }
    
        @Override
        public void run(StreamTableEnvironment tenv) { 
    
            //1.建表与source关联 t1 dwm_order_wide
           tenv.executeSql("create table order_wide(" +
                    " province_id bigint, " +
                    " province_name string, " +
                    " province_area_code string, " +
                    " province_iso_code string, " +
                    " province_3166_2_code string, " +
                    " order_id bigint, " +
                    " split_total_amount decimal(12,2), " +
                    " create_time string, " +
                    " event_time as to_timestamp(create_time), " +
                    " watermark for event_time as event_time - interval '5' second " +
                    ")with(" +
                    " 'connector' = 'kafka'," +
                    " 'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                    " 'properties.group.id' = 'DWSProvinceStateApp', " +
                    " 'topic' = 'dwm_order_wide', " +
                    " 'scan.startup.mode' = 'latest-offset', " +
                    " 'format' = 'json' " +
                    ")");
    
    
          //2. 建表与sink表关联 t2
            tenv.executeSql("create table province_stats_2021(" +
                    " stt string, " +
                    " edt string, " +
                    " province_id bigint, " +
                    " province_name string, " +
                    " area_code string, " +
                    " iso_code string, " +
                    " iso_3166_2 string, " +
                    " order_amount decimal(20,2), " +
                    " order_count bigint, " +
                    " ts bigint, " +
                    " primary key(stt,edt,province_id) not enforced" +
                    ")with(" +
                    " 'connector' = 'clickhouse', " +
                    " 'url' = 'clickhouse://hadoop162:8123', " +
                    " 'database-name' = 'gmall2021', " +
                    " 'table-name' = 'province_stats_2021', " +
                    " 'sink.batch-size' = '100', " +
                    " 'sink.flush-interval' = '1000', " +
                    " 'sink.max-retries' = '3' " +
                    ")");
    
    
    
            //3.查询t1，把结果写入到t2
            final Table table = tenv.sqlQuery("select " +
                    " date_format(tumble_start(event_time, interval '10' second),'yyyy-MM-dd HH:mm:ss'), " +
                    " date_format(tumble_end(event_time, interval '10' second),'yyyy-MM-dd HH:mm:ss'), " +
                    " province_id, " +
                    " province_name, " +
                    " province_area_code, " +
                    " province_iso_code, " +
                    " province_3166_2_code, " +
                    " sum(split_total_amount) order_amount, " +
                    " count(distinct order_id) order_count, " +
                    " unix_timestamp()*1000 ts" +
                    " from order_wide " +
                    " group by " +
                    " tumble(event_time, interval '10' second)," +
                    " province_id, " +
                    " province_name, " +
                    " province_area_code, " +
                    " province_iso_code, " +
                    " province_3166_2_code");
    
            table.executeInsert("province_stats_2021");
    
    
        }
    }

[https_help.aliyun.com_document_detail_185696.html_spm_a2c4g.11186623.6.574.d9c541ea3J78mc]: https://help.aliyun.com/document_detail/185696.html?spm=a2c4g.11186623.6.574.d9c541ea3J78mc