從實時持續生成的資料中獲取可操作的見解是當今許多企業的共同要求， 實時資料處理的一個廣泛用例是儀表板， 支持此類用例的典型架構基于資料流處理器、具有低延遲讀/寫訪問的資料存盤和可視化框架，

在這篇博文中，我們演示了如何使用 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 的結合，它比將產生許多豐富的應用場景，