flink 和es 实时数据

Apache Flink 是一个分布式流处理引擎，可以在流数据和批数据上进行处理，具有高效、高可用、高容错等特点。而 Elasticsearch（ES）是一个用于实时搜索和分析的分布式搜索引擎，可以高效地存储、搜索和分析大量数据。

Flink 和 ES 可以集成使用，实现实时数据处理和分析。常见的应用场景包括：

1. 数据实时同步：将 Flink 处理的流数据实时同步到 ES 中，以便进行快速搜索和分析。

2. 实时数据分析：使用 Flink 处理流数据，并将处理结果实时写入 ES 中，以便进行实时分析和可视化。

3. 实时报警：使用 Flink 处理流数据，根据特定的规则和条件实时检测数据，并将检测结果写入 ES 中，以便进行实时报警和处理。

要实现 Flink 和 ES 的集成，可以使用 Flink 的 Elasticsearch Connector。该 Connector 可以将 Flink 处理的数据实时写入 ES 中，同时支持数据批量提交、数据过滤、数据转换等功能，可以灵活地满足不同场景下的需求。

示例代码如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

DataStream<String> stream = env.socketTextStream("localhost", 9999);

// 将数据写入 Elasticsearch
List<HttpHost> httpHosts = Arrays.asList(new HttpHost("localhost", 9200, "http"));
ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(httpHosts, new ElasticsearchSinkFunction<String>() {
    public IndexRequest createIndexRequest(String element) {
        Map<String, String> json = new HashMap<>();
        json.put("data", element);
        return Requests.indexRequest()
            .index("my-index")
            .type("my-type")
            .source(json);
    }

    public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
        indexer.add(createIndexRequest(element));
    }
});

stream.addSink(esSinkBuilder.build());

env.execute("Flink and Elasticsearch Example");

该示例代码通过 Flink 处理从 socket 中读取的数据，并使用 ElasticsearchSink 将数据写入 Elasticsearch 中，其中 `httpHosts` 参数指定了 ES 地址和端口，`createIndexRequest` 方法用于构造写入 ES 的数据，`process` 方法用于将数据写入 ES。