1 # clone the project
2git clone https://github.com/itinycheng/flink-connector-clickhouse.git
3
4 # enter the project directory
5cd flink-connector-clickhouse/
6
7 # display remote branches
8git branch -r
9
10 # checkout the branch you need
11git checkout $branch_name
12
13 # install or deploy the project to our own repository
14mvn clean install -DskipTests
15mvn clean deploy -DskipTests

1<dependency>
2    <groupId>org.apache.flink</groupId>
3    <artifactId>flink-connector-clickhouse</artifactId>
4    <version>1.16.0-SNAPSHOT</version>
5</dependency>

1-- register a clickhouse table `t_user` in flink sql.
2CREATE TABLE t_user (
3    `user_id` BIGINT,
4    `user_type` INTEGER,
5    `language` STRING,
6    `country` STRING,
7    `gender` STRING,
8    `score` DOUBLE,
9    `list` ARRAY<STRING>,
10    `map` Map<STRING, BIGINT>,
11    PRIMARY KEY (`user_id`) NOT ENFORCED
12) WITH (
13    'connector' = 'clickhouse',
14    'url' = 'clickhouse://{ip}:{port}',
15    'database-name' = 'tutorial',
16    'table-name' = 'users',
17    'sink.batch-size' = '500',
18    'sink.flush-interval' = '1000',
19    'sink.max-retries' = '3'
20);
21
22-- read data from clickhouse 
23SELECT user_id, user_type from t_user;
24
25-- write data into the clickhouse table from the table `T`
26INSERT INTO t_user
27SELECT cast(`user_id` as BIGINT), `user_type`, `lang`, `country`, `gender`, `score`, ARRAY['CODER', 'SPORTSMAN'], CAST(MAP['BABA', cast(10 as BIGINT), 'NIO', cast(8 as BIGINT)] AS MAP<STRING, BIGINT>) FROM T;

1val tEnv = TableEnvironment.create(setting)
2
3val props = new util.HashMap[String, String]()
4props.put(ClickHouseConfig.DATABASE_NAME, "default")
5props.put(ClickHouseConfig.URL, "clickhouse://127.0.0.1:8123")
6props.put(ClickHouseConfig.USERNAME, "username")
7props.put(ClickHouseConfig.PASSWORD, "password")
8props.put(ClickHouseConfig.SINK_FLUSH_INTERVAL, "30s")
9val cHcatalog = new ClickHouseCatalog("clickhouse", props)
10tEnv.registerCatalog("clickhouse", cHcatalog)
11tEnv.useCatalog("clickhouse")
12
13tEnv.executeSql("insert into `clickhouse`.`default`.`t_table` select...");

1TableEnvironment tEnv = TableEnvironment.create(setting);
2
3Map<String, String> props = new HashMap<>();
4props.put(ClickHouseConfig.DATABASE_NAME, "default")
5props.put(ClickHouseConfig.URL, "clickhouse://127.0.0.1:8123")
6props.put(ClickHouseConfig.USERNAME, "username")
7props.put(ClickHouseConfig.PASSWORD, "password")
8props.put(ClickHouseConfig.SINK_FLUSH_INTERVAL, "30s");
9Catalog cHcatalog = new ClickHouseCatalog("clickhouse", props);
10tEnv.registerCatalog("clickhouse", cHcatalog);
11tEnv.useCatalog("clickhouse");
12
13tEnv.executeSql("insert into `clickhouse`.`default`.`t_table` select...");

1> CREATE CATALOG clickhouse WITH (
2    'type' = 'clickhouse',
3    'url' = 'clickhouse://127.0.0.1:8123',
4    'username' = 'username',
5    'password' = 'password',
6    'database-name' = 'default',
7    'use-local' = 'false',
8    ...
9);
10
11> USE CATALOG clickhouse;
12> SELECT user_id, user_type FROM `default`.`t_user` limit 10;
13> INSERT INTO `default`.`t_user` SELECT ...;