目錄

 

一、導入clickhouse jdbc 依賴

二、編寫 Flink 寫入ClickHouse代碼

三、創建ClickHouse 表

四、運行向localhost，7777端口發送數據，并啟動Flink應用程序

五、查詢ClickHouse 數據結果，驗證數據是否寫入成功

---

一、導入clickhouse jdbc 依賴

        <!-- 寫入數據到clickhouse -->
        <dependency>
            <groupId>ru.yandex.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
            <version>0.1.54</version>
        </dependency>

二、編寫 Flink 寫入ClickHouse代碼

Java Bean實體類

package com.lei.domain;

public class J_User {
    public int id;
    public String name;
    public int age;

    public J_User(int id, String name, int age) {
        this.id = id;
        this.name = name;
        this.age = age;
    }

    public static J_User of(int id, String name, int age) {
        return new J_User(id, name, age);
    }
}

編寫ClickHouseUtil 工具類

package com.lei.util;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;


public class ClickHouseUtil {
    private static Connection connection;

    public static Connection getConn(String host, int port, String database) throws SQLException, ClassNotFoundException {
        Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
        String  address = "jdbc:clickhouse://" + host + ":" + port + "/" + database;
        connection = DriverManager.getConnection(address);
        return connection;
    }

    public static Connection getConn(String host, int port) throws SQLException, ClassNotFoundException {
        return getConn(host,port,"default");
    }
    public static Connection getConn() throws SQLException, ClassNotFoundException {
        return getConn("node-01",8123);
    }
    public void close() throws SQLException {
        connection.close();
    }
}

編寫 業務寫入ClickHouse類

package com.lei.util;

import com.lei.domain.J_User;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;


public class J_MyClickHouseUtil extends RichSinkFunction<J_User> {
    Connection connection = null;

    String sql;

    public J_MyClickHouseUtil(String sql) {
        this.sql = sql;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = ClickHouseUtil.getConn("node-01", 8123, "default");
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (connection != null) {
            connection.close();
        }
    }

    @Override
    public void invoke(J_User user, Context context) throws Exception {
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        preparedStatement.setLong(1, user.id);
        preparedStatement.setString(2, user.name);
        preparedStatement.setLong(3, user.age);
        preparedStatement.addBatch();

        long startTime = System.currentTimeMillis();
        int[] ints = preparedStatement.executeBatch();
        connection.commit();
        long endTime = System.currentTimeMillis();
        System.out.println("批量插入完畢用時：" + (endTime - startTime) + " -- 插入數據 = " + ints.length);
    }
}

編寫Flink 業務類，即執行業務邏輯

package com.lei.sinktest;

import com.lei.domain.J_User;
import com.lei.util.J_MyClickHouseUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
    進入clickhouse-client
    use default;
    drop table if exists user_table;

    CREATE TABLE default.user_table(id UInt16, name String, age UInt16 ) ENGINE = TinyLog();
 */
public class J05_ClickHouseSinkTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);

        // source
        DataStream<String> inputStream = env.socketTextStream("localhost", 7777);

        // Transform 操作
        SingleOutputStreamOperator<J_User> dataStream = inputStream.map(new MapFunction<String, J_User>() {
            @Override
            public J_User map(String data) throws Exception {
                String[] split = data.split(",");
                return J_User.of(Integer.parseInt(split[0]),
                        split[1],
                        Integer.parseInt(split[2]));
            }
        });

        // sink
        String sql = "INSERT INTO default.user_table (id, name, age) VALUES (?,?,?)";
        J_MyClickHouseUtil jdbcSink = new J_MyClickHouseUtil(sql);
        dataStream.addSink(jdbcSink);
        dataStream.print();

        env.execute("clickhouse sink test");
    }
}

---

三、創建ClickHouse 表

-- 進入clickhouse-client
use default;
drop table if exists user_table;

CREATE TABLE default.user_table(id UInt16, name String, age UInt16 ) ENGINE = TinyLog();

四、運行向localhost，7777端口發送數據，并啟動Flink應用程序

五、查詢ClickHouse 數據結果，驗證數據是否寫入成功

---

文章最后，給大家推薦一些受歡迎的技術博客鏈接：

1. JAVA相關的深度技術博客鏈接
2. Flink 相關技術博客鏈接
3. Spark 核心技術鏈接
4. 設計模式 —— 深度技術博客鏈接
5. 機器學習 —— 深度技術博客鏈接
6. Hadoop相關技術博客鏈接
7. 超全干貨--Flink思維導圖，花了3周左右編寫、校對
8. 深入JAVA 的JVM核心原理解決線上各種故障【附案例】
9. 請談談你對volatile的理解？--最近小李子與面試官的一場“硬核較量”
10. 聊聊RPC通信，經常被問到的一道面試題。源碼+筆記，包懂
11. 深入聊聊Java 垃圾回收機制【附原理圖及調優方法】

---

歡迎掃描下方的二維碼或 搜索 公眾號“大數據高級架構師”，我們會有更多、且及時的資料推送給您，歡迎多多交流！