2018 Elastic Meetup南京交流會，由趙偉帶來以“ElasticLog with ES in CloudEdge”為題的演講。本文首先介紹了CloudEdge與ElasticLog是什麼，其次介紹了産品的構架圖以及ES的作用，最後介紹了ES在實踐過程中需要設計Index、配置設定Shard、快速将Spark裡資料寫入ES中和資料去重。本文提出的産品已經解決了一些瓶頸問題，但還有一些問題仍在路上。 阿裡雲Elasticsearch 1核2G首月免費試用，開始雲上實踐吧 直播視訊回顧 以下内容根據現場分享整理而成。
CloudEdge與ElasticLog
CloudEdge
CloudEdge是一款自注冊、自配置、自充值的網關産品，它可以分為兩部分，一部分是硬體，另一部分是盒子。CloudEdge通常部署在客戶網絡的進出口，所有的流量都會經過這個盒子進行安全掃描，并做相應的流量控制。
CloudEdge産品主要面臨着三方面的問題，第一方面是如何快速處理、快速存儲大量資料量的問題，以供使用者去查詢。第二方面是要滿足使用者想要查某個時間段的資料不需查整段資料的需求，研究人員應當怎樣做。第三方面是由于産品面對的是日本客戶，日本客戶比較謹慎，他們追求資料一條不能丢，一條不能多的精确度，研究人員應當怎樣做。
CloudEdge對流量進行掃描并進行控制後，會産生大量資料，例如網絡通路相關的資料等。Project ElasticLog是負責将不安全的資料進行攔截，接着将所有的資料按照一定的格式寫到軟體中，讓使用者能夠看到自己所需要關注的資訊。最終通過盒子産生資料，Project ElasticLog負責将不安全的資料進行攔截後，做出跟業務相關、資料相關的scalable 系統。
架構圖
目前此架構已經基本實作，但在部署上依舊存在一些問題。在上圖的架構中，藍色部分是資料的走向，直到資料落地，綠色部分負責資料落地後使用者查詢資料。在上圖中,首先，Client負責将資料聚合後上傳到Log receiververs上，其中Log receiververs是部署在AWS上的。其次，Log receiververs将資料上傳到Kinesis stream上，其中Kinesis stream是負責托管服務的。再次，Kinesis stream将資料上傳到經典的生産消費者模式區域。最後，資料傳輸到下遊，下遊中一部分負責把資料源源不斷的寫進ES裡去，另一部分負責把資料做相應的解析後寫進S3裡去，其中S3是負責存儲服務的。
經過以上處理後，理論上解決了問題，但實際上還需要進一步的延伸。第一，對ES裡面的資料進行聚合，即資料進去是以分鐘機關計算的，出來時需要聚合成小時機關；第二，将經常查詢的資料放到ES裡面，剩下的資料放到S3裡面；第三，對放到S3内部的資料進行查詢，目前對放到S3内部的資料進行查詢的方法是用AWS提供的Athena服務，并用托管的方式去搜尋S3上的資料，最後提供一個ECS供使用者去查詢資料。
ES的作用以及現狀
ES 在 ElasticLog中主要負責三件事情，第一，主要用來放一個月的熱資料，即每分鐘有130萬條資料進入，必須提供的是秒級查詢；第二，ES應建立在AWS上，需要三個專用主節點與十個資料節點，其中主節點指M4.Load，資料節點指C4.2xBug；第三，它最多需要考慮28個名額參數。
在ES服務中，WAS是首選的服務，因為WAS能夠提供相對可靠的ES服務，并且是全托管的服務。它可以從一些叢集的維護中解放出來，更多的集中在怎樣用ES把業務實作。當ES服務達到一定臨界值時，讓WAS去管理ES服務。另外，在ES服務中，WAS具有區域意識、易于使用、VPC+安全組、藍/綠部署以及螢幕+自動快照等特點，例如，區域意識指可以在不同的機房将資料進行備份。
ES實踐
ES的實踐包括四個部分，第一部分是怎樣去設計一個Index，使它能夠很大程度上影響到class的走向；第二部分是Shard的配置設定，即在名額裡面放多少個Shard是合适的；第三部分是快速的将Spark裡的資料寫到ES裡去；第四部分是資料的去重。四個部分的詳細介紹如下：
設計Index
設計Index有兩種形式，第一種形式是隻設計一個Index，但整個程式建立起來後許多地方都不能進行修改了，隻适用于小型資料；第二種形式是按時間去分割，即分割成每周一個Index或每月一個Index，能夠有效地解決資料量的問題。
上圖是ES實踐中的一個設計案例，在這個案例的設計中需做一些考量的性能。第一，ES裡面不能放兩個doc，因為存放兩個同名的doc需要字段類型一緻，且多個doc類型已經不被支援，是以一個Index僅需存放一個doc類型即可；第二，如果客戶隻在乎查詢結果，不考慮資料是什麼，那麼可以将source關掉，這樣會有效地節約存儲空間；第三，all的資料在做Index時能夠做到将所有的資料聚合起來，如果隻查找 某一段的資料則可把all關掉，這樣不僅能夠使空間的使用率得到有效地提升，還能使性能得到優化；第四，可以将dynamic改為strict。
在ES實踐中，它是将分鐘次元資料聚合成小時次元資料的，這就意味着将資料寫進ES時是分鐘Index，過一段時間後自動轉換成小時次元Index。那麼查詢時它是怎樣将分鐘Index切換成小時Index的呢？原因是Index提供了一個aliases功能，這樣資料一開始指向分鐘Index,當分鐘Index服務周期到了後，自動切換到小時Index，其中整個切換在資料的聚合裡面去完成即可。
Shard的配置設定
那麼在有了Index的基礎上，怎樣去合理的配置設定Shard的個數呢？另外，Shard是落在不同的叢集上，又該怎樣去選擇合适的叢集大小呢？具體介紹如下：
在設計好Index之後,進行讀取資料量，并估計每周以至每分鐘有多少資料量送進來。估計資料量的大小之後基本就能夠确定Index有多大以及需要多少個Index，接着确定磁盤空間以及需要多少機器裝置。進而算出Shard的個數，其中Shard的個數是資料節點個數的k倍，且建議每個Shard存儲不得超過30GB。最後進行相應的測試、調整以及回報，經過幾輪測試後，基本就可以确定業務大概需要什麼樣的叢集，這個叢集需要多大了。
快速地将Spark裡資料寫到ES内
首先，通過一個Spark Streaming程式把資料源源不斷地寫進ES裡去，接着，Spark調用ES-Hadoop庫把資料寫進ES當中。那麼怎樣快速地将Spark裡的資料寫到ES裡去呢？除了調整一些ES參數以外，還需要将Spark進行并行計算和ES進行并行寫入，使得并行task執行的個數等于叢集總共擁有的vCores個數，最終兩者對接實作資源的最大利用和快速地寫入。
資料去重
在上文講的架構中，由于Spark在做并行運算時會把任務分到每個task裡面去做，如果一個task寫入ES中失敗了，那麼Spark在task裡面就會重新寫入，進而導緻資料出現重複。要想将重複進到ES的資料去重，可以從使用自定義唯一ID、使用聚合來發現複制和删除檔案以及做明确的查詢三個方面入手。