Elasticsearch：使用 Apache FlinkElasticsearch 打造实时事件处理及搜索

Posted 2021-10-31 中国社区官方博客

从实时持续生成的数据中获取可操作的见解是当今许多企业的共同要求。 实时数据处理的一个广泛用例是仪表板。 支持此类用例的典型架构基于数据流处理器、具有低延迟读/写访问的数据存储和可视化框架。

在这篇博文中，我们演示了如何使用 Apache Flink 和 Elasticsearch 为流数据分析构建实时事件处理及搜索。 下图描述了我们的系统架构。在实际的应用中，我们可以使用 Kibana 共同打造实时仪表板解决方案。

在我们的架构中，Apache Flink 执行流分析作业，这些作业摄取数据流，应用转换来分析、转换和建模动态数据，并将其结果写入 Elasticsearch 索引。 Kibana 连接到索引并查询它以获取要可视化的数据。 我们架构的所有组件都是 Apache License 2.0 下的开源系统。 在今天的展示中，我将重点讲述如何从数据的摄入到 Flink 并做相应的一些处理，并最终写入到 Elasticsearch 中去。

为什么要使用 Apache Flink 进行流处理？

在深入探讨实现演示应用程序的细节之前，我们先讨论一些使 Apache Flink 成为出色流处理器的特性。Apache Flink 带有一组具有竞争力的流处理功能，其中一些在开源领域是独一无二的。最重要的是：

• 支持事件时间和乱序流：实际上，事件流很少按照它们产生的顺序到达，尤其是来自分布式系统和设备的流。直到现在，由应用程序员来纠正这种“时间漂移”，或者干脆忽略它并接受不准确的结果，因为流系统（至少在开源世界中）不支持事件时间（即处理事件当它们发生在现实世界时）。 Flink 是第一个支持乱序流并且能够根据时间戳一致处理事件的开源引擎。
• Scala 和 Java 中富有表现力且易于使用的 API：Flink 的 DataStream API 将许多在批处理 API 中众所周知的操作符（例如 map、reduce 和 join）移植到流媒体世界。此外，它还提供特定于流的操作，例如窗口（window）、拆分（split）和连接（connect）。对用户定义函数的一流支持简化了自定义应用程序行为的实现。 DataStream API 在 Scala 和 Java 中可用。
• 支持会话和未对齐的窗口：大多数流媒体系统都有一些窗口的概念，即基于某些时间函数的一组事件。不幸的是，在许多系统中，这些窗口是硬编码的，并与系统的内部检查点机制相关联。 Flink 是第一个将窗口与容错完全解耦的开源流引擎，允许更丰富的窗口形式，例如会话。
• 一致性、容错性和高可用性：Flink 保证在出现故障时状态更新的一致性（通常称为“exactly-once processing”），以及选定源和接收器之间的一致数据移动（例如，Kafka 和 HDFS 之间的一致数据移动）。Flink 还支持 worker 和 master 故障转移，消除任何单点故障。
• 低延迟和高吞吐量：我们已经将 Flink 的时钟频率设置为每核心每秒 150 万个事件，并且还观察到包括网络数据改组在内的作业的延迟在 25 毫秒范围内。使用调整旋钮，Flink 用户可以导航延迟-吞吐量权衡，使系统既适合高吞吐量数据摄取和转换，也适合超低延迟（毫秒范围）应用程序。
• 连接器和集成点：Flink 与各种开源系统集成，用于数据输入和输出（例如 HDFS、Kafka、Elasticsearch、HBase 等）、部署（例如 YARN）以及充当执行引擎对于其他框架（例如，Cascading、Google Cloud Dataflow）。 Flink 项目本身捆绑了一个 Hadoop MapReduce 兼容层、一个 Storm 兼容层，以及用于机器学习和图形处理的库。
• 开发人员生产力和操作简单性：Flink 可在各种环境中运行。 IDE 中的本地执行显着简化了 Flink 应用程序的开发和调试。在分布式设置中，Flink 以大规模横向扩展运行。 YARN 模式允许用户在几秒钟内启动 Flink 集群。 Flink 通过定义良好的 REST 接口来监控作业和整个系统的指标。内置的 Web 仪表板显示这些指标，并使 Flink 的监控非常方便。

这些特性的结合使 Apache Flink 成为许多流处理应用程序的独特选择。

Flink stream processing API

 在接下来的步骤中，我们将按照上面的顺序来完成对事件的处理。

安装

对于没有接触 Flink 及 Elastic Stack 的开发者来说，你需要安装如下的部分：

Elasticsearch

你可以参考我之前的文章 “如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch” 在你自己喜欢的系统上安装 Elasticsearch。

Kibana

你可以参考我之前的文章 “ Kibana：如何在 Linux，MacOS 及 Windows上安装 Elastic 栈中的 Kibana” 在自己喜欢的系统上安装 Kibana。

Flink

对于这个部分的按照，你可以参考如下的链接：

• macOS: How to Install Apache Flink On Mac OS
• Windows: How to Install Apache Flink On Local Windows
• Ubuntu: How to Install Apache Flink On Ubuntu

在这些系统上的安装是非常直接的。针对我的安装，我选择 macOS。我使用如下的方式来运行 Flink:

$ start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host liuxg.
Starting taskexecutor daemon on host liuxg.

如上所示，它显示我们的 Flink 已经成功运行起来了。在启动后，我们甚至可以在浏览器中打开地址 http://localhost:8081来查看 Flink 的运行状态。我们甚至在这里可以提交我们的任务。

 如果你能看到上面的画面，说明我们的 Flink 的安装是成功的。

创建演示例子

接下来，我们将使用 Java 来构建一个展示的例子。它使用 API 来访问 Flink。如上所示，我们将使用  Flink 的 enviornment，source，transform 及 sink APIs 来构建我们的应用。为了方便大家学习，我已经把我的项目上传到 github 了。你需要使用如下的命令来进行下载：

git clone https://github.com/liu-xiao-guo/ElasticsearchFlink

你可以使用你自己喜欢的 IDE 来创建一个新的项目来开始。

source

在我们的练习中，我们将使用 nc 这个工具来发送数据。你需要在自己的平台上安装 nc。我们使用如下命令来启动 nc:

nc -l 8888

如上所示，它打开端口 8888，并侦听（-l）向这个端口发送的连接。我们可以在一个 terminal 中运行上面的命令。在下面的实验中，我们可以在这个 terminal 中打入字符串，并回车。这样它就可以把数据发送到一个已经建立的连接中。

ElasticsearchFlink.java

这是整个代码的最重要的部分。其实也是蛮简单的。我把代码贴下来：

ElasticsearchFlink.java

import com.liuxg.User;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ElasticsearchFlink {
    public static void main(String[] args) {
        // Create Flink environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Define a source
        try {
            DataStreamSource<String> source = env.socketTextStream("localhost", 8888);

            DataStream<String> filterSource = source.filter(new FilterFunction<String>() {
                @Override
                public boolean filter(String s) throws Exception {
                    return !s.contains("hello");
                }
            });

            DataStream<User> transSource = filterSource.map(value -> {
                String[] fields = value.split(",");
                return new User(fields[ 0 ], fields[ 1 ]);
            });

            // Use ESBuilder  to construct an output
            List<HttpHost> hosts = new ArrayList<>();
            hosts.add(new HttpHost("localhost", 9200, "http"));
            ElasticsearchSink.Builder<User> builder = new ElasticsearchSink.Builder<User>(hosts,
                    new ElasticsearchSinkFunction<User>() {
                        @Override
                        public void process(User u, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                            Map<String, String> jsonMap = new HashMap<>();
                            jsonMap.put("id", u.id);
                            jsonMap.put("name", u.name);
                            IndexRequest indexRequest = Requests.indexRequest();
                            indexRequest.index("flink-test");
                            // indexRequest.id("1000");
                            indexRequest.source(jsonMap);
                            requestIndexer.add(indexRequest);
                        }
                    });
            
            // Define a sink
            builder.setBulkFlushMaxActions(1);
            transSource.addSink(builder.build());

            // Execute the transform
            env.execute("flink-es");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

如上所示，我们在开始的部分得到 enviroment：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

我们接下来，使用如下的方法来建立一个和 localhost:8888 端口的连接：

   DataStreamSource<String> source = env.socketTextStream("localhost", 8888);

如果我们的 nc 已经成功运行，那么上面的句子将正常返回。

接下来，我们使用一个 Flink 的 filter 功能。它对数据做一个简单的 transform。如果字符串中含有 "hello"，这个数据将被忽略。最终它不会被写入到 Elasticsearch 中：

            DataStream<String> filterSource = source.filter(new FilterFunction<String>() {
                @Override
                public boolean filter(String s) throws Exception {
                    return !s.contains("hello");
                }
            });

再接下来，我们使用一个 Map 的 transform 功能。比如，当我们的输入数据为 1,liuxg 时，我们希望提前到的数据是 id:1, 及 name:liuxg。

            DataStream<User> transSource = filterSource.map(value -> {
                String[] fields = value.split(",");
                return new User(fields[ 0 ], fields[ 1 ]);
            });

这个 transfrom 也非常简单。

在 Flink API 的最后部分是 sink。我们可以通过如下的方式来写入数据到 Elasticsearch 中：

            // Use ESBuilder  to construct an output
            List<HttpHost> hosts = new ArrayList<>();
            hosts.add(new HttpHost("localhost", 9200, "http"));
            ElasticsearchSink.Builder<User> builder = new ElasticsearchSink.Builder<User>(hosts,
                    new ElasticsearchSinkFunction<User>() {
                        @Override
                        public void process(User u, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                            Map<String, String> jsonMap = new HashMap<>();
                            jsonMap.put("id", u.id);
                            jsonMap.put("name", u.name);
                            IndexRequest indexRequest = Requests.indexRequest();
                            indexRequest.index("flink-test");
                            // indexRequest.id("1000");
                            indexRequest.source(jsonMap);
                            requestIndexer.add(indexRequest);
                        }
                    });
            
            // Define a sink
            builder.setBulkFlushMaxActions(1);
            transSource.addSink(builder.build());

            // Execute the transform
            env.execute("flink-es");

在这里需要注意的是我们在 hosts 的构建中：

   hosts.add(new HttpHost("localhost", 9200, "http"));

我们需要根据自己的 Elasticsearch 地址及端口号做相应的修改。在上面特别需要指出的是如下的这句：

builder.setBulkFlushMaxActions(1);

因为 Flink 有批处理及实时处理，在上面我们设置这个参数值为1，表明每当收到任何的信息，就会立即进行处理，而不需要等到收集到一定的事件后再做处理。

我们接下运行应用。在运行之前我们确保 nc 已经成功运行，否则应用将会退出。我们接下在 nc 运行所在的界面中打入如下的一行字并回车：

1,liuxg

 我们在 Kibana 中进行查看：

GET _cat/indices/flink-test

它将显示有一个叫做 flink-test 的索引已经被成功地创建了：

 我们再接着使用如下的命令来进行搜索：

GET flink-test/_search

我们看到有一个文档已经被创建了。

我们再接下来打入如下的一行字：

2,hello

显然在这个输入中，它含有 hello 字符串。在我们的设计中，如果含有 hello，那么在 filter 的设计中将返回 false，也就是说这个数据将不被写入到 Elasticsearch 中。我们可以在 Kibana 中使用上面的同样的命令来进行查看。

结论

在这篇博文中，我们演示了如何使用 Apache Flink 和 Elasticsearch 构建实时事件处理及搜索的应用程序。 通过支持事件时间处理，Apache Flink 能够产生有意义且一致的结果，即使对于历史数据或在事件无序到达的环境中也是如此。 与其他开源流处理解决方案相比，具有灵活窗口语义的富有表现力的 DataStream API 可显着减少自定义应用程序逻辑。 在本次的展示中，我们使用了 Flink 的极少一部分对数据 transform 的功能。Flink 具有许多的数据分析功能。通过 Flink 和 Elastic Stack 的结合，它比将产生许多丰富的应用场景。