Flink DataStream API （十五）Flink 输出到 Elasticsearch_elasticsearchsinkfunction

ElasticSearch 是一个分布式的开源搜索和分析引擎，适用于所有类型的数据。ElasticSearch有着简洁的 REST 风格的 API，以良好的分布式特性、速度和可扩展性而闻名，在大数据领域应用非常广泛。
Flink 为 ElasticSearch 专门提供了官方的 Sink 连接器，Flink 1.13 支持当前最新版本的ElasticSearch。

写入数据的 ElasticSearch 的测试步骤如下。
（1）添加 Elasticsearch 连接器依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
1
2
3
4
5

（2）启动 Elasticsearch

安装 Elasticsearch ：Linux地址安装

Elasticsearch 索引相关 ：索引相关

Elasticsearch 文档相关：文档相关

不太了解Elasticsearch的小伙伴可以点击 Elasticsearch 教学 由浅入深。

（3）编写输出到 Elasticsearch 的示例代码

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

        // 读取数据源，并行度为 1
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Alice", "./prod?id=100", 3000L)
        );

        //创建hosts
        ArrayList<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("hadoop102", 9200, "http"));

        // 创建一个 ElasticsearchSinkFunction
        ElasticsearchSinkFunction<Event> elasticsearchSinkFunction = new ElasticsearchSinkFunction<Event>() {
            @Override
            public void process(Event element, RuntimeContext ctx, RequestIndexer indexer) {
                HashMap<String, String> data = new HashMap<>();
                data.put(element.user, element.url); // 加入 user 和 url
                IndexRequest request = Requests.indexRequest()
                        .index("clicks") //索引
                        //.type("type") // Es 6 必须定义 type
                        .source(data); //数据源
                indexer.add(request); //添加进去
            }
        };

        stream.addSink(new ElasticsearchSink.Builder<Event>(httpHosts, elasticsearchSinkFunction).build());

        env.execute();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

源代码如下：Flink 输出到 Elasticsearch

与 RedisSink 类 似 ， 连 接 器 也 为 我 们 实 现 了 写 入 到 Elasticsearch 的SinkFunction——ElasticsearchSink。区别在于，这个类的构造方法是私有（private）的，我们需要使用 ElasticsearchSink 的 Builder 内部静态类，调用它的 build()方法才能创建出真正的SinkFunction。 而 Builder 的构造方法中又有两个参数：

• httpHosts：连接到的 Elasticsearch 集群主机列表
• elasticsearchSinkFunction：这并不是我们所说的 SinkFunction，而是用来说明具体处理逻辑、准备数据向 Elasticsearch 发送请求的函数
  具体的操作需要重写中 elasticsearchSinkFunction 中的 process 方法，我们可以将要发送的数据放在一个 HashMap 中，包装成 IndexRequest 向外部发送 HTTP 请求。

（4）运行代码，访问 Elasticsearch 查看是否收到数据，查询结果如下所示。