flink写数据到clickhouse的代码

### 回答1：

答： Flink 写入 Clickhouse 的代码可以采用 JDBC Sink 方式，具体实现可以参考 Flink 官方文档中的示例：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/clickhouse.html。

### 回答2：
Flink是一个开源的流数据处理框架，而ClickHouse是一个高性能的列式数据库。使用Flink将数据写入到ClickHouse可以实现实时的数据处理和分析。

首先，需要在Flink项目中添加ClickHouse的依赖项。可以在pom.xml文件中添加以下代码：

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

然后，在Flink应用程序中使用ClickHouseSinkFunction将数据写入ClickHouse。以下是一个简单的示例代码：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.clickhouse.ClickHouseSink;
import org.apache.flink.streaming.connectors.clickhouse.ClickHouseSinkFunction;
import org.apache.flink.streaming.connectors.clickhouse.deps.GuavaDurationConverter;
import org.apache.flink.streaming.connectors.clickhouse.table.ClickHouseSinkBufferFlusher;
import org.apache.flink.streaming.connectors.clickhouse.table.ClickHouseSinkConfiguration;
import org.apache.flink.streaming.connectors.clickhouse.table.internal.ClickHouseStream;
import org.apache.flink.streaming.connectors.clickhouse.table.internal.options.ClickHouseOptions;
import org.apache.flink.types.Row;

import java.sql.Types;

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

        DataStream<Tuple2<String, Integer>> stream = env.socketTextStream("localhost", 9999)
                .map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String s) throws Exception {
                        String[] parts = s.split(",");
                        return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));
                    }
                });

        String[] fieldNames = {"name", "age"};
        int[] fieldTypes = {Types.VARCHAR, Types.INTEGER};

        ClickHouseOptions options = new ClickHouseOptions("jdbc:clickhouse://localhost:8123/default", "", "");
        ClickHouseSinkFunction<Tuple2<String, Integer>> sinkFunction = new ClickHouseSinkFunction<>(fieldNames, fieldTypes, options);

        ClickHouseSink<Tuple2<String, Integer>> sink = ClickHouseSink.buildSink(sinkFunction, new ClickHouseSinkConfiguration(), new ClickHouseSinkBufferFlusher(options));

        stream.addSink(sink);

        env.execute("Flink ClickHouse Example");
    }
}

上述代码使用`socketTextStream`从Socket读取输入数据，并使用`map`函数将数据转换为Tuple2对象。然后，我们定义ClickHouse的字段名和字段类型，并创建ClickHouseOptions对象，指定ClickHouse的连接URL、用户名和密码。

接下来，我们创建ClickHouseSinkFunction对象，并将其传递给ClickHouseSink。最后，将DataStream添加到sink中，通过调用`env.execute`来启动Flink作业。

通过添加适当的ClickHouse配置，并修改DataStream的源，您可以根据实际需求来改变代码。以上代码只是一个简单的示例，供参考。