Flink Sink to ClickHouse with Table API with Connector

Build flink-connector-clickhouse

Get the code:

git clone git@github.com:itinycheng/flink-connector-clickhouse.git

Update the maven plugin version:

<maven-shade-plugin.version>3.6.0</maven-shade-plugin.version>

mvn clean install

Add connector lib to project

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.pkslow</groupId>
        <artifactId>bigdata</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <groupId>com.pkslow.bigdata</groupId>
    <artifactId>flink-clickhouse-table-api-with-connector</artifactId>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.16.3</flink.version>
    </properties>

    <dependencies>
<!--        Flink Base-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>

<!--        Flink Table API-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-loader</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>${flink.version}</version>
        </dependency>

<!--        Filesystem and csv-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>${flink.version}</version>
        </dependency>

<!--        ExecutorFactory-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
        </dependency>

<!--        JDBC-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- ClickHouse JDBC Driver -->
        <dependency>
            <groupId>com.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
            <version>0.7.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-clickhouse</artifactId>
            <version>1.16.0-SNAPSHOT</version>
        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.5.11</version>
        </dependency>


    </dependencies>

</project>

Java code:

package com.pkslow.bigdata;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class FlinkCsvToClickHouseWithConnectorMain {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        String sourceDDL = "CREATE TABLE csv_source (\n" +
                "    id INT,\n" +
                "    name STRING,\n" +
                "    age INT\n" +
                ") WITH (\n" +
                "    'connector' = 'filesystem',\n" +
                "    'path' = 'file:///Users/larry/IdeaProjects/pkslow-samples/bigdata/flink-csv-to-parquet/src/main/resources/input.csv',\n" +  // Update this path
                "    'format' = 'csv',\n" +
                "    'csv.ignore-parse-errors' = 'true'\n" +
                ")";

        tableEnv.executeSql(sourceDDL);

        String printTable = "CREATE TABLE print_table (\n" +
                "  id INT,\n" +
                "  name STRING,\n" +
                "  age INT\n" +
                ") WITH (\n" +
                "  'connector' = 'print'\n" +
                ");";

        tableEnv.executeSql(printTable);

        String insertSQL = "INSERT INTO print_table SELECT * FROM csv_source";
        tableEnv.executeSql(insertSQL);

        tableEnv.executeSql(
                "CREATE TABLE clickhouse_sink (\n" +
                        "    id BIGINT,\n" +
                        "    name STRING,\n" +
                        "    age INT\n" +
                        ") WITH (\n" +
                        "    'connector' = 'clickhouse',\n" +
                        "    'url' = 'clickhouse://localhost:8123',\n" +
                        "    'database-name' = 'flink',\n" +
                        "    'table-name' = 'persons',\n" +
//                        "    'driver' = 'com.clickhouse.jdbc.ClickHouseDriver',\n" +
                        "    'username' = 'larry',\n" +
                        "    'password' = 'larrydpk'\n" +
                        ")"
        );

        tableEnv.executeSql(
                "INSERT INTO clickhouse_sink\n" +
                        "SELECT id, name, age FROM csv_source"
        );

    }
}

5 Code

Code on GitHub: https://github.com/LarryDpk/pkslow-samples