flink 写入clickhouse

可以使用 Flink 的 JDBC Sink 将数据写入 ClickHouse 数据库。具体步骤如下：

1. 在 pom.xml 中添加 ClickHouse JDBC 驱动的依赖：

<dependency>
    <groupId>ru.yandex.clickhouse</groupId>
    <artifactId>clickhouse-jdbc</artifactId>
    <version>0.3.0</version>
</dependency>

2. 在 Flink 程序中创建 ClickHouse JDBC Sink：

import java.sql.PreparedStatement;
import java.sql.SQLException;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;

public class ClickHouseSink extends RichSinkFunction<String> {

    private static final long serialVersionUID = 1L;
    private static final Logger LOG = LoggerFactory.getLogger(ClickHouseSink.class);

    private ClickHouseConnection connection;
    private PreparedStatement statement;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        // 初始化 ClickHouse 连接
        ClickHouseDataSource dataSource = new ClickHouseDataSource("jdbc:clickhouse://<clickhouse-host>:<clickhouse-port>/<clickhouse-database>");
        connection = dataSource.getConnection();
        statement = connection.prepareStatement("INSERT INTO <clickhouse-table> (col1, col2, ...) VALUES (?, ?, ...)");
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        String[] fields = value.split(",");

        // 设置 PreparedStatement 的参数
        statement.setString(1, fields[0]);
        statement.setInt(2, Integer.parseInt(fields[1]));
        ...

        // 执行插入操作
        statement.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        super.close();

        // 关闭 ClickHouse 连接
        if (statement != null) {
            statement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }

}

3. 在 Flink 程序中使用 ClickHouse JDBC Sink 输出数据：

DataStream<String> dataStream = ... // 获取数据流
dataStream.addSink(new ClickHouseSink());

其中 `<clickhouse-host>`、`<clickhouse-port>`、`<clickhouse-database>` 和 `<clickhouse-table>` 分别表示 ClickHouse 数据库的主机名、端口号、数据库名称和数据表名称。在执行插入操作时，需要根据实际情况设置 PreparedStatement 的参数。