flink 1.12 kafka 写elasticsearch

如果您想使用Flink 1.12将Kafka数据写入Elasticsearch中，可以按照以下步骤操作：

1. 首先，您需要在项目中添加Flink的Kafka和Elasticsearch依赖，例如：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>1.12.0</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
    <version>1.12.0</version>
</dependency>

2. 创建一个Flink Streaming Job，并使用Kafka作为数据源，例如：

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test");

DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), props));

3. 将数据转换为Elasticsearch的数据格式，并将其写入Elasticsearch中，例如：

List<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("localhost", 9200, "http"));

stream.map(new MapFunction<String, Map<String, Object>>() {
    @Override
    public Map<String, Object> map(String value) throws Exception {
        // 将数据转换为Elasticsearch的数据格式
        Map<String, Object> data = new HashMap<>();
        data.put("message", value);
        data.put("@timestamp", new Date());

        return data;
    }
}).addSink(new ElasticsearchSink.Builder<>(httpHosts, new ElasticsearchSinkFunction<Map<String, Object>>() {
    @Override
    public void process(Map<String, Object> element, RuntimeContext ctx, RequestIndexer indexer) {
        // 将数据写入Elasticsearch中
        IndexRequest request = Requests.indexRequest()
                .index("my-index")
                .source(element);
        indexer.add(request);
    }
}).build());

上述代码中，我们将Kafka中的数据转换为Elasticsearch的数据格式，然后使用ElasticsearchSinkFunction将数据写入Elasticsearch中。

希望这些能够帮到您！