Day 25 - Elasticsearch Ingest节点数据管道处理器

bindiego 发表了文章 • 1 个评论 • 6510 次浏览 • 2018-12-24 18:18 • 来自相关话题

Day 23 - 基于 HanLP 的 ES 中文分词插件

rochy 发表了文章 • 0 个评论 • 7620 次浏览 • 2018-12-23 11:51 • 来自相关话题

一、分词插件

1、分词器概念

在 ES 中，分词器的作用是从文本中提取出若干词元（token）来支持索引的存储和搜索，分词器（Analyzer）由一个分解器（Tokenizer）、零个或多个词元过滤器（TokenFilter）组成。

分解器用于将字符串分解成一系列词元，词元过滤器的作用是对分词器提取出来的词元做进一步处理，比如转成小写，增加同义词等。处理后的结果称为索引词（Term），引擎会建立 Term 和原文档的倒排索引（Inverted Index），这样就能根据 Term 很快到找到源文档了。

![文本分词并索引的过程](http://img.luooqi.com/FsIejZLnwcQiiTYNlLl-a8sLUepg)

2、选择分词器

目前 ES 分词插件的选择性还是很多的，分词插件的核心就是提供各种分词器（Analyzer）、分解器（Tokenizer）、词元过滤器（TokenFilter）；根据依赖的核心分词包（分词算法）的不同显现出不同的差异性，除了分词算法之外，是否支持用户自定义词典，是否支持词典热更新等其他附加功能也是选择分词插件时需要参考的。

下面列出选择分词插件需要考虑的因素（仅供参考）：

分词准确性：大家都希望分词结果能够尽可能准确，与分词准确性直接相关的就是用户词典了，此外才是分词算法；
分词算法：个人认为无需纠结于分词算法，大多数分词包提供的分词算法都比较类似，选择时不需要过于纠结；
分词速度：这个与分词算法直接相关，基于词典的分词算法一般比基于模型的分词算法要快；基于词典如果考虑词频、命名实体识别、词性标注则会慢一些；
启动速度：当词典较大时，初始化词典会比较慢，某些分词器会对词典进行缓存，第二次启动会非常速度；
内存占用：与分词算法、词典大小、模型大小均有关系，设计精巧的算法对内存占用较小；
易用性：分词器是否开箱即用，是否可以直接使用在线链接或者压缩包进行安装，是否需要复杂的配置；
扩展性：是否支持用户自定义词典、是否支持自定义分词算法、是否支持热更新等；
是否开源：开源的分词器在遇到问题的时候可以自己进行深度调试，甚至可以进行二次开发；
社区活跃度：这个看一下 github 的 star 数或者依赖的分词包的 star 数和 issue 数目即可判定；
更新频率：是否能够与最新版的 ES 同步更新。

二、HanLP 简介

HanLP 是一系列模型与算法组成的 NLP 工具包，具备功能完善、性能高效、架构清晰、语料时新、可自定义的特点,详情可参考 github 介绍：[https://github.com/hankcs/HanLP](https://github.com/hankcs/HanLP)。

选择 HanLP 作为核心的分词包开发 ES 分词插件，主要考虑以下因素：
HanLP 是 Java 分词包中最为流行的；
HanLP 提供了多种分词器，既可以基于词典也可以基于模型（在一亿字的大型综合语料库上训练的分词模型）；
HanLP 坚持使用明文词典，这样可以借助社区的力量对词典不断进行完善；
完善的开发文档和代码样例，较为活跃的用户群体；
个人参与了部分功能的开发，对代码结构较为熟悉。

三、开发分词插件

1、代码结构
conf：插件的配置文件、HanLP 的配置文件、Java 安全策略文件；
scr.main.java.assemby：插件打包（maven-assembly-plugin）配置文件；
org.elasticsearch.plugin.hanlp.analysis：分词插件核心构建器；
org.elasticsearch.plugin.hanlp.conf：管理插件配置、分词器配置以及 HanLP 配置；
org.elasticsearch.plugin.hanlp.lucene：HanLP 中文分词 Lucene 插件，对 Lucune 分词进行实现；
scr.main.resources：插件属性文件所在目录

![插件代码结构](http://img.luooqi.com/FnyoQBBEjOGfmv7sEQHQToqqba0r)

2、TokenStream

Analyzer 类是一个抽象类，是所有分词器的基类，它通过 TokenStream 类将文本转换为词汇单元流；TokenStream 有两种实现 Tokenizer（输入为 Reader）和 TokenFilter（输入为另一个 TokenStream）。

![文本分词流程](http://img.luooqi.com/FmvVcPk8KM-Ujd-lqh6qmToXL18C)

TokenStream 基本使用流程：
1. 实例化 TokenStream，向 AttributeSource 添加/获取属性（词汇单元文本、位置增量、偏移量、词汇类型等）；
2. 调用 reset() 方法，将流（stream）重置到原始（clean）状态；
3. 循环调用 incrementToken() 方法，并处理 Attribute 属性信息，直到它返回 false 表示流处理结束；
4. 调用 end() 方法，确保流结束（end-of-stream）的操作可以被执行；
5. 调用 close() 方法释放资源。
 
 // 实例化 TokenStream TokenStream tokenStream = new IKAnalyzer().tokenStream("keywords",new StringReader("思想者")); // 向 AttributeSource 添加/获取属性 CharTermAttribute attribute = tokenStream.addAttribute(CharTermAttribute.class); // 将流（stream）重置到原始（clean）状态 tokenStream.reset(); // 判断是否还有下一个 Token while(tokenStream.incrementToken()) { System.out.println(attribute); } tokenStream.end(); tokenStream.close(); 
 
 综上，开发 Tokenizer 或者 TokenFilter 时，需要重点关注 reset、incrementToken、end、close 四个方法的实现。
 
 3、开发中的小技巧
 
 获取插件目录或文件目录
 
  //获取插件根目录 private static Path getPluginPath() { return env.pluginsFile().resolve("analysis-hanlp"); } //获取插件目录下的文件 private static Path getDefDicConfigPath() { return env.pluginsFile().resolve("analysis-hanlp/hanlp.properties").toAbsolutePath(); } 
 
 插件属性文件
 
 如果希望插件属性文件（plugin-descriptor.properties）能够自动根据 pom.xml 中的属性进行赋值，则需要将文件防止到 resources 文件夹下。
 
 插件版本兼容性
 
 从实际测试来看：
ES5.X 及其以上的代码是完全复用的，也就是说代码逻辑不需要调整；
ES5.X 到 ES6.2.X 的插件是可以通用的，其特征是打包的时候需要将插件的文件全部打包到 elasticsearch 文件夹下；
ES6.3.X 以上的插件是可以通用的，打包的时候插件的文件全部打包到根目录即可。

也就是说，如果你升级了新版本 ES，对于插件升级，大多数情况只需要修改下 plugin-descriptor.properties 文件中 ES 的版本号即可。

4、安全策略文件

在插件开发中经常会使用到文件读取、属性读取、网络链接等功能，如果不提前注册安全策略，在调用这些功能的时候会报以下错误java.security.AccessControlException: access denied。

官方给出的解决方案就是新建一个 plugin-security.policy 文件，然后在文件中声明需要的权限信息，最后在打包的时候将文件放置到插件的根目录，这样在使用 zip 包进行安装的时候，ES 会提示用户插件所需的权限信息，需要用户确认后插件才能正常安装。
```
@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
@ WARNING: plugin requires additional permissions @
@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
java.io.FilePermission <> read,write,delete
java.lang.RuntimePermission createClassLoader
java.lang.RuntimePermission getClassLoader
java.lang.RuntimePermission setContextClassLoader
java.net.SocketPermission * connect,resolve
java.util.PropertyPermission * read,write
See http://docs.oracle.com/javase/ ... .html
for descriptions of what these permissions allow and the associated risks.

Continue with installation? [y/N]y
-> Installed analysis-hanlp
```

5、安全策略的坑

最开始认为只需要添加了 policy 文件，且打包到正确的位置即可解决插件的权限问题，因为在插件安装的时候 ES 已经提示了所需权限，但是代码在实际执行的时候依旧报 AccessControlException 的错误。

参考了多个 HanLP 的 ES 分词插件，都没有获得较好的方法，后来考虑到 IK 分词器远程加载词典时，需要网络连接权限，就去看了下其远程词典加载的代码，最终找到了正确的使用方法。
 // 需要特殊权限的代码 AccessController.doPrivileged((PrivilegedAction<Segment>) () -> { Segment segment; if (config.getAlgorithm().equals("extend")) { segment = new ViterbiSegment(); } else { segment = HanLP.newSegment(config.getAlgorithm()); } // 在此处显示调用一下分词，使得加载词典、缓存词典的操作可以正确执行 System.out.println( segment.seg("HanLP中文分词工具包！")); return segment; }); 

四、插件特色

简单介绍一下插件的特点：
内置多种分词模式，适合不同场景；
内置词典，无需额外配置即可使用；
支持外置词典，用户可自定义分词算法，基于词典或是模型；
支持分词器级别的自定义词典，便于用于多租户场景；
支持远程词典热更新（待开发）；
拼音过滤器、繁简体过滤器（待开发）；
基于词语或单字的 ngram 切分分词（待开发）。

Github 地址：[https://github.com/AnyListen/e ... hanlp](https://github.com/AnyListen/e ... -hanlp)

Day22 - 熟练使用ES离做好搜索还差多远？

nodexy 发表了文章 • 2 个评论 • 9213 次浏览 • 2018-12-21 17:27 • 来自相关话题

作者：杨振涛搜索引擎架构师@vivo
首次发布：Elasticsearch中文社区
发布日期：2018-12-22

搜索引擎作为互联网发展历史中一个非常典型的产品/业务形态，时至今日并没有太大的突破性变化；主流形态可以划分为大搜、垂搜、企业级搜索和站内/app内搜索等。除了Google, Yahoo, Bing, Ask 等以及国内百度、搜狗、360、神马等是人们熟识的大搜之外，非业内人士还真不知道其他还有哪些公司以及有哪些搜索产品或业务场景。实际上，在信息爆炸的时代，几乎每家有点儿规模的公司都或多或少要涉及到搜索引擎，最起码你需要接触SEO/SEM。本文将从非大搜企业的搜索需求出发，并基于开源技术栈来介绍和探讨搜索引擎在实践中的几个核心任务及其主要解决思路。同时为了避免重复，本文以外链形式引用了大量网络已有的国内外公开资料，方便大家参考，需要注意的是部分内容可能会随着时间推移而过期或链接失效。

提到开源搜索引擎，在Java技术栈里以Lucene, Solr/SolrCloud及Elasticsearch为代表的几个项目可能最为流行。本文的写作初衷是解答”熟练使用ES等开源搜索引擎解决方案以后，要如何才能做好搜索产品/业务？“ 希望对你有所帮助，如果你有关于此话题的更多实践经验或不同见解，欢迎留言评论交流。

1. 做好搜索引擎意味着什么？

有一位同行的文章总结了好的搜索引擎的衡量维度：

相关性
体验
性能

其中相关性是非常重要的一个维度，这里我将通过引用一篇文章来介绍什么是”相关性工程“以及”相关性工程师“ http://www.flax.co.uk/blog/201 ... ound/ 。

相关性工程中的“相关性”，主要是指代用户的Query与索引库中的Doc之间的相关性，所以可以分别从索引数据和Query两个方面来考虑。

相关性工程考虑的第一个特征就是基于已有索引数据的文本相关度计算，通常有TF-IDF、BM25、BM25F等。Elasticsearch早期的版本默认都是TF-IDF，目前已更改为BM25。对于中文数据，分词方法和策略也会直接影响到文本相关度的计算；其次匹配方式也非常重要；最后就是基于此的相关性算分了。

相关性工程还可以考虑更多的特征，尤其是从索引数据之外来挖掘出的特征，比如索引文档的权威性、时效性、专业性、质量与口碑评分、热度与流行度等。结合NLP技术，相关性工程还可以考虑语义距离等特征，丰富召回结果。当然这些特征的处理与机器学习中的特征工程基本一致，比如涉及归一化问题、权重问题、稀疏性问题、非典型分布等等。

相关性工程考虑的另一个重要特征是用户点击反馈数据，即对于用户所看到的搜索结果列表，点击行为被看作是对当前搜索结果的一种认可，用户点了哪个位置的doc对于继续优化相关性至关重要。这两天有个著名案例就是Google的劈柴在听证会上解释为什么在Google搜索Idiot出现的都是特朗普的照片。

体验涉及的方面较多，最重要的就是产品功能和交互方面的体验了，比如一个典型的搜索产品，C端可能具备以下功能：

搜索前：搜索框，搜索入口，热搜榜/飙升榜/大家都在搜，搜索发现，默认搜索词，历史搜索记录，猜你想搜，分类搜索，语音输入搜索/图片搜索；广告位
搜索中：搜索联想直达，搜索联想词，输入纠错，关键词匹配高亮
搜索后：搜索结果列表，列表页推荐/广告，特形展示，列表穿插，搜了还搜，搜索详情页，详情页搜索推荐，无结果及少结果填充，筛选条件/筛选器，自主排序，列表样式切换（宫格 | 列表）

除了产品功能，还需要考虑搜索引擎的可运营性，比如搜索运营管理系统，至少要具备基本功能的各种黑白灰名单，包括人工干预，优化分词的自定义词典、同义词典、停用词典，以及对查询词的强制改写或者升降权；对索引内容的管控，比如对检索字段的；对召回和排序的相关参数的优化和调整等等；此外，还有配套的SEO或ASO系统，以及各种数据指标相关的看板系统。

而搜索结果中的特形展示，也是目前比较主流的产品形式，不管是自然结果还是搜索广告，都可以提供更快捷的体验，甚至一度成为知识图谱在搜索产品中应用的代表性功能。另外搜索联想中的直达服务，也是目前比较流行的，可以进一步缩短用户的操作路径，直达目标内容或服务。

更多关于产品体验和设计类问题可以参考 http://www.woshipm.com/tag/搜索功能

性能方面，搜索引擎的每一次查询理论上都是实时运算，大部分搜索引擎系统都是实时或准实时的，这就要求在用户感知上要有基本的响应时间（RT）保障，比如在国内公网环境下，200ms是比较优秀的体验，300ms-500ms是正常的体验，500ms+就需要尽快去优化。除去其中的网络I/O等开销，对于后端搜索服务的RT，一般是T99在100ms以内，T90在50ms以内，具体标准取决与当事业务和产品。除了RT，可用性也是非常重要的，一般要求99.9%以上；另外，索引数据的生效时间也很重要，比如新加入的索引，或者已有索引的更新和删除，秒级生效是比较好的体验。需要明确的一点是，这里的性能指标我们针对的是To C用户，如果是企业级搜索甚至是基于ES的一个即时查询分析系统，可能复杂查询的秒级响应也是很正常的。

延伸阅读：

阿里云-开放搜索-最佳实践-功能篇-相关性实践 https://help.aliyun.com/document_detail/29186.html
Defining relevance engineering 什么是相关性工程 http://www.flax.co.uk/blog/201 ... ound/

2. 搜索引擎是典型的机器学习问题

云计算、大数据、AI 先后成为IT与互联网行业的热点，三者经常被称为CBA或ABC技术，而这些都与一家公司密切相关，那就是 Google ! 众所周知 Google 是一家著名的全球搜索引擎公司，但其产品远不止搜索引擎。从Google的三驾马车 GFS, MapReduce, BigTable开始，后来有了 Yahoo牵头的开源实现Hadoop（Hadoop最早来自于Nutch，是的，没错，就是那个开发了 Lucene 的 Doug Cutting所开源的Nutch，他被称作Hadoop之父），到后来的云计算与大数据技术蓬勃，到今天的AI热潮，各种深度学习各种NN， Google Brain，开源的Tensorflow，对Google来说这一切都是搜索引擎业务驱动的水到渠成的发展轨迹。可以说搜索引擎是天生的机器学习问题，有着诸多的机器学习/深度学习应用场景。这里顺便DISS下一些眼高手低的迷糊党，互联网圈儿曾有人遇到求职者表示想做AI，却不做搜索不做推荐不做广告！（本人内心：你咋不上天呢！请记住AI is a buzzword. ）当今的互联网或移动互联网，搜索、推荐与广告是三大典型的所谓AI应用落地方向，其他的也有但并未发展成熟，至少还没有成熟的变现模式；而这三者或多或少有些交集，搜索几乎是最基础的一个，比如推荐也需要建立索引，需要检索TOP N，搜索广告也需要做召回和排序。

如果我们把搜索引擎的核心模型简化下，其实主要是在解决三大类问题：

- 数据：内容侧的爬虫，预处理，内容分析和理解，索引建立和存储；用户侧的Query理解，改写，意图识别等
- 算法/模型：把相关性和排序等业务问题抽象为回归或分类/聚类问题，特征工程，离线训练模型，在线预测
- 策略：为满足业务需求而制定并持续优化的一系列规则、模型或模型组合、参数优化等活动，并通过工程化实现体现到线上系统，以及配套的试验和评估系统

2.1 Ranking 排序

常见的排序策略有：

单维度排序：顾名思义按照单个维度来排序，没有任何复杂性可言，在召回结果集不太大的情况下实时排序即可。
优先级排序：相对单维度排序而已一般是先按维度A排，当A的排序依据一样或相等时再按维度B排序，以此类推。
加权排序：针对多个维度或特征，赋予不同权重，并按求和之后的得分来排序；实践中通常会采用分层加权排序（第一层加权排序之后，得到不少于2个得分，继续加权后排序），或者分组加权排序（第一层分组来加权排序后，对所得到的得分可能按业务需求进行非求和类的运算比如乘法，再按最终得分排序）的策略。加权排序的难点在于，如何设置并持续优化这些权重，通常会建模为典型的机器学习问题来拟合。
机器学习排序：即所谓LTR，根据用户点击或人工标注数据集建立学习目标，然后通过特征工程来挖掘与目标有关系的一系列特征，并建立学习模型，通过训练集获得模型参数，以该组参数为基准做预测，上线后再基于用户点击数据持续优化该模型的参数。LTR是一个通用方法的称谓，不是某一个具体算法的名称，具体算法名称参见下文。

排序问题可以简单抽象成为预测用户点击列表中对象的概率问题。

2.2 ES生态内的 LTR

关于LTR的理论和方法学，已经有很多论文和资料了 ( 参考 wikipedia , LTR简介-微软亚研院， LTR Pairwise to Listwise ,  大规模LTR-Google，LTR书籍 )，感兴趣的可以阅读，这里主要提供几个JAVA和ES生态的工程实现参考。

es-ltr插件  http://es-learn-to-rank.labs.o19s.com/

Set of command line tools for Learning To Rank https://github.com/SeaseLtd/ltr-tools

es的ranking evaluation api https://www.elastic.co/guide/e ... .html

Java LTR类库： RankLib  https://sourceforge.net/p/lemur/wiki/RankLib/

支持算法如下：

MART (Multiple Additive Regression Trees, a.k.a. Gradient boosted regression tree)
RankNet
RankBoost
AdaRank
Coordinate Ascent
LambdaMART
ListNet
Random Forests

Day24 - Predator捕捉病毒样本

swordsmanli 发表了文章 • 0 个评论 • 3311 次浏览 • 2018-12-21 16:18 • 来自相关话题

对，你一定看过一个电影，情节是这样的，他们拿着长矛去狩猎异形怪物，它们比人类强健，它们的脸部的器官布置得出奇丑陋。它们的身上总是带了一堆很先进的狩猎武器，它们喜欢在杀死猎物后将尸体剥皮，还会将猎物头骨加工成工艺品，当成战利品收藏。对，这部电影系列就叫Predator。好了，言归正传，我们今天讲的故事其实非常简单，讲述的是elasticsearch引擎在安全领域的简单应用，如何通过elasticsearch来搜索一个病毒，我们开发了一个小小的工具来帮我做跨集群查询，以及SQL-DSL转换接口，我们把这个小工具叫做predator。

背景

我司主要是做病毒相关工作的，近年来，数据爆炸，病毒软件也成几何级数倍数增长，大数据病毒出现自然需要对应的大数据工具来处理它们，简单来讲，就是我们可以把病毒样本的一些属性剥离到elasticsearch中，就和日志来描述一个用户的行为一样，本质来说，它们都是数据，然后，我们研究病毒的一些特征属性，通过简单的搜索，就可以快速分析出一堆可能的病毒样本，再然后，通过一系列的测试，过滤，我们就可以真正的找到我们想要的病毒样本，并且通过这些规则持续的追踪它们，是不是很简单?

问题

事情是那么简单，但是在使用elasticsearch作为特征库的过程中，我们也有这样的问题：
1，多种维度特征
由于存在多种维度特征的病毒，不通模块剥离出不通病毒属性，所以存在多张表来存属性，那么在query的时候就需要跨表，甚至跨集群查询。
2，DSL的复杂度
由于内部研究员们对elasticStack并不熟悉，加上DSL语言相对复杂，我们需要使用更加接近hunman特性的SQL来转换DSL语言。

数据处理架构

我们有一个类似的数据处理架构

Predator和它的Spear

因此，我们开发了一个小工具，其实，这个小工具非常简单，只是简单的解决了上述2个问题：
使用Elasticsearch-SQL插件来包装一个restful的DSL转换SQL接口，当然，目前ES6已经完全支持SQL接口了，哈哈，早点出来我们就不用做那么工作了:) :):)。
简单的写个跨集群的查下聚合器就可以实现跨表查下，其实，这个功能只是简单的查下封装，只是针对特殊的业务场景，没啥参考价值。
至于Spear，它其实就是个predator service的客户端，哈哈，像不像铁血战士拿着长矛开着非常去狩猎的样子:) 。

这是一个规则：

这是规则的查询结果：

长矛的sample code:
```python

cross cluster search by dsls

import json
from spear import Spear
sp = Spear()
dsl_1 = {}
dsl_2 = {}
query_dict = {
json.dumps(dsl_1): {
"cluster": "es_cluster_1",
"type":"xxx"
},
json.dumps(dsl_2): {
"cluster": "es_cluster_2",
"type": "yyy"
}
}
sp.cross_count_by_dsl(query_dict, is_show_help=False)
```
当然长矛也支持SQL接口

总结

其实，这个只是一个user case的工程实践，可以看到的是，伟大的ElasticStack在各行各业，各种大数据领域，如果抛开领域的概念，一切都是数据，那么理论上来说我们可以使用elasticsearch处理任何类型的数据，当然目前业界典型的应用场景还是搜索，日志，甚至于APM，总之，紧跟社区可以学到很多东西啦。

Day 20 - Elastic性能实战指南

laoyang360 发表了文章 • 10 个评论 • 4871 次浏览 • 2018-12-20 09:14 • 来自相关话题

让Elasticsearch飞起来!——性能优化实践干货

0、题记

Elasticsearch性能优化的最终目的：用户体验爽。
关于爽的定义——著名产品人梁宁曾经说过“人在满足时候的状态叫做愉悦，人不被满足就会难受，就会开始寻求。如果这个人在寻求中，能立刻得到即时满足，这种感觉就是爽！”。
Elasticsearch的爽点就是：快、准、全!
关于Elasticsearch性能优化，阿里、腾讯、京东、携程、滴滴、58等都有过很多深入的实践总结，都是非常好的参考。本文换一个思路，基于Elasticsearch的爽点，进行性能优化相关探讨。

1、集群规划优化实践

1.1 基于目标数据量规划集群

在业务初期，经常被问到的问题，要几个节点的集群，内存、CPU要多大，要不要SSD？
最主要的考虑点是：你的目标存储数据量是多大？可以针对目标数据量反推节点多少。

1.2 要留出容量Buffer

注意：Elasticsearch有三个警戒水位线，磁盘使用率达到85%、90%、95%。
不同警戒水位线会有不同的应急处理策略。
这点，磁盘容量选型中要规划在内。控制在85%之下是合理的。
当然，也可以通过配置做调整。

1.3 ES集群各节点尽量不要和其他业务功能复用一台机器。

除非内存非常大。
举例：普通服务器，安装了ES+Mysql+redis，业务数据量大了之后，势必会出现内存不足等问题。

1.4 磁盘尽量选择SSD

Elasticsearch官方文档肯定推荐SSD，考虑到成本的原因。需要结合业务场景，
如果业务对写入、检索速率有较高的速率要求，建议使用SSD磁盘。
阿里的业务场景，SSD磁盘比机械硬盘的速率提升了5倍。
但要因业务场景而异。

1.5 内存配置要合理

官方建议：堆内存的大小是官方建议是：Min（32GB，机器内存大小/2）。
Medcl和wood大叔都有明确说过，不必要设置32/31GB那么大，建议：热数据设置：26GB，冷数据：31GB。
总体内存大小没有具体要求，但肯定是内容越大，检索性能越好。
经验值供参考：每天200GB+增量数据的业务场景，服务器至少要64GB内存。
除了JVM之外的预留内存要充足，否则也会经常OOM。

1.6 CPU核数不要太小

CPU核数是和ESThread pool关联的。和写入、检索性能都有关联。
建议：16核+。

1.7 超大量级的业务场景，可以考虑跨集群检索

除非业务量级非常大，例如：滴滴、携程的PB+的业务场景，否则基本不太需要跨集群检索。

1.8 集群节点个数无需奇数

ES内部维护集群通信，不是基于zookeeper的分发部署机制，所以，无需奇数。
但是discovery.zen.minimum_master_nodes的值要设置为：候选主节点的个数/2+1，才能有效避免脑裂。

1.9 节点类型优化分配

集群节点数：<=3，建议：所有节点的master：true， data：true。既是主节点也是路由节点。
集群节点数：>3, 根据业务场景需要，建议：逐步独立出Master节点和协调/路由节点。

1.10 建议冷热数据分离

热数据存储SSD和普通历史数据存储机械磁盘，物理上提高检索效率。

2、索引优化实践

Mysql等关系型数据库要分库、分表。Elasticserach的话也要做好充分的考虑。

2.1 设置多少个索引？

建议根据业务场景进行存储。
不同通道类型的数据要分索引存储。举例：知乎采集信息存储到知乎索引；APP采集信息存储到APP索引。

2.2 设置多少分片？

建议根据数据量衡量。
经验值：建议每个分片大小不要超过30GB。

2.3 分片数设置？

建议根据集群节点的个数规模，分片个数建议>=集群节点的个数。
5节点的集群，5个分片就比较合理。
注意：除非reindex操作，分片数是不可以修改的。

2.4副本数设置？

除非你对系统的健壮性有异常高的要求，比如：银行系统。可以考虑2个副本以上。
否则，1个副本足够。
注意：副本数是可以通过配置随时修改的。

2.5不要再在一个索引下创建多个type

即便你是5.X版本，考虑到未来版本升级等后续的可扩展性。
建议：一个索引对应一个type。6.x默认对应_doc，5.x你就直接对应type统一为doc。

2.6 按照日期规划索引

随着业务量的增加，单一索引和数据量激增给的矛盾凸显。
按照日期规划索引是必然选择。
好处1：可以实现历史数据秒删。很对历史索引delete即可。注意：一个索引的话需要借助delete_by_query+force_merge操作，慢且删除不彻底。
好处2：便于冷热数据分开管理，检索最近几天的数据，直接物理上指定对应日期的索引，速度快的一逼！
操作参考：模板使用+rollover API使用。

2.7 务必使用别名

ES不像mysql方面的更改索引名称。使用别名就是一个相对灵活的选择。

3、数据模型优化实践

3.1 不要使用默认的Mapping

默认Mapping的字段类型是系统自动识别的。其中：string类型默认分成：text和keyword两种类型。如果你的业务中不需要分词、检索，仅需要精确匹配，仅设置为keyword即可。
根据业务需要选择合适的类型，有利于节省空间和提升精度，如：浮点型的选择。

3.2 Mapping各字段的选型流程

3.3 选择合理的分词器

常见的开源中文分词器包括：ik分词器、ansj分词器、hanlp分词器、结巴分词器、海量分词器、“ElasticSearch最全分词器比较及使用方法” 搜索可查看对比效果。
如果选择ik，建议使用ik_max_word。因为：粗粒度的分词结果基本包含细粒度ik_smart的结果。

3.4 date、long、还是keyword

根据业务需要，如果需要基于时间轴做分析，必须date类型；
如果仅需要秒级返回，建议使用keyword。

4、数据写入优化实践

4.1 要不要秒级响应？

Elasticsearch近实时的本质是：最快1s写入的数据可以被查询到。
如果refresh_interval设置为1s，势必会产生大量的segment，检索性能会受到影响。
所以，非实时的场景可以调大，设置为30s，甚至-1。

4.2 减少副本，提升写入性能。

写入前，副本数设置为0，
写入后，副本数设置为原来值。

4.3 能批量就不单条写入

批量接口为bulk，批量的大小要结合队列的大小，而队列大小和线程池大小、机器的cpu核数。

4.4 禁用swap

在Linux系统上，通过运行以下命令临时禁用交换：

sudo swapoff -a

5、检索聚合优化实战

5.1 禁用 wildcard模糊匹配

数据量级达到TB+甚至更高之后，wildcard在多字段组合的情况下很容易出现卡死，甚至导致集群节点崩溃宕机的情况。
后果不堪设想。
替代方案：
方案一：针对精确度要求高的方案:两套分词器结合，standard和ik结合，使用match_phrase检索。
方案二：针对精确度要求不高的替代方案：建议ik分词，通过match_phrase和slop结合查询。

5.2极小的概率使用match匹配

中文match匹配显然结果是不准确的。很大的业务场景会使用短语匹配“match_phrase"。
match_phrase结合合理的分词词典、词库，会使得搜索结果精确度更高，避免噪音数据。

5.3 结合业务场景，大量使用filter过滤器

对于不需要使用计算相关度评分的场景，无疑filter缓存机制会使得检索更快。
举例：过滤某邮编号码。

5.3控制返回字段和结果

和mysql查询一样，业务开发中，select * 操作几乎是不必须的。
同理，ES中，_source 返回全部字段也是非必须的。
要通过_source 控制字段的返回，只返回业务相关的字段。
网页正文content，网页快照html_content类似字段的批量返回，可能就是业务上的设计缺陷。
显然，摘要字段应该提前写入，而不是查询content后再截取处理。

5.4 分页深度查询和遍历

分页查询使用：from+size;
遍历使用：scroll；
并行遍历使用：scroll+slice。
斟酌集合业务选型使用。

5.5 聚合Size的合理设置

聚合结果是不精确的。除非你设置size为2的32次幂-1，否则聚合的结果是取每个分片的Top size元素后综合排序后的值。
实际业务场景要求精确反馈结果的要注意。
尽量不要获取全量聚合结果——从业务层面取TopN聚合结果值是非常合理的。因为的确排序靠后的结果值意义不大。

5.6 聚合分页合理实现

聚合结果展示的时，势必面临聚合后分页的问题，而ES官方基于性能原因不支持聚合后分页。
如果需要聚合后分页，需要自开发实现。包含但不限于：
方案一：每次取聚合结果，拿到内存中分页返回。
方案二：scroll结合scroll after集合redis实现。

6、业务优化

让Elasticsearch做它擅长的事情，很显然，它更擅长基于倒排索引进行搜索。
业务层面，用户想最快速度看到自己想要的结果，中间的“字段处理、格式化、标准化”等一堆操作，用户是不关注的。
为了让Elasticsearch更高效的检索，建议：
1）要做足“前戏”
字段抽取、倾向性分析、分类/聚类、相关性判定放在写入ES之前的ETL阶段进行；
2）“睡服”产品经理
产品经理基于各种奇葩业务场景可能会提各种无理需求。
作为技术人员，要“通知以情晓之以理”，给产品经理讲解明白搜索引擎的原理、Elasticsearch的原理，哪些能做，哪些真的“臣妾做不到”。

7、小结

实际业务开发中，公司一般要求又想马儿不吃草，又想马儿飞快跑。
对于Elasticsearch开发也是，硬件资源不足（cpu、内存、磁盘都爆满）几乎没有办法提升性能的。
除了检索聚合，让Elasticsearch做N多相关、不相干的工作，然后得出结论“Elastic也就那样慢，没有想像的快”。
你脑海中是否也有类似的场景浮现呢？
提供相对NB的硬件资源、做好前期的各种准备工作、让Elasticsearch轻装上阵，相信你的Elasticsearch也会飞起来！

来日我们再相会......

推荐阅读：
1、阿里：https://elasticsearch.cn/article/6171
2、滴滴：http://t.cn/EUNLkNU
3、腾讯：http://t.cn/E4y9ylL
4、携程：https://elasticsearch.cn/article/6205
5、社区：https://elasticsearch.cn/article/6202
6、社区：https://elasticsearch.cn/article/708
7、社区：https://elasticsearch.cn/article/6202

Elasticsearch基础、进阶、实战第一公众号

Day 19 - 通过点击反馈优化es搜索结果排序

laigood 发表了文章 • 1 个评论 • 6799 次浏览 • 2018-12-19 00:15 • 来自相关话题

相信不少人都把es当做一个主要的搜索引擎来使用，但是对于搜索结果之后的点击反馈，es没有很好的方案。比如说用户搜索了某些关键词，点击了某些结果，而这些结果并不是排在最前面的，但确实是用户最想要的。那有没有什么方法可以使它们排在前面呢？一种简单的做法就是就是离线统计文档的点击率，然后在排序时根据这个点击率进行加权，但这样笼统的算法不一定适合所有情况。现在就来简单介绍下learning to rank，翻译过来就是学习排序，可以根据点击日志里面的记录，来反向影响搜索结果的排序。刚好这个库也有es的插件，下面以这个插件的官方demo来解释下如何使用。
demo的下载地址如下，都是python脚本，环境需求：python3+，es
https://github.com/o19s/elasti ... /demo
1.准备数据
python prepare.py
下载RankLib.jar (用来训练模型) 和tmdb.json (测试数据集，tmdb的电影数据)
2.导测试数据入es
python index_ml_tmdb.py
3.训练模型
python train.py
训练脚本很简单，但是脚本里面有丰富的实现，下面介绍下主要方法。
load_features(FEATURE_SET_NAME)
这个是读取特征信息，demo定义了两个特征，分别在1.json

{

    "query": {

        "match": {

            "title": "{{keywords}}"

        }

    }

}

和2.json

{

    "query": {

        "match": {

            "overview": "{{keywords}}"

        }

    }

}

1就是查title，2就是查overview，生成训练数据时就是需要根据特征的查询语法，去es里面匹配相关得分作为特征分数。

movieJudgments = judgments_by_qid(judgments_from_file(filename=JUDGMENTS_FILE))

读取生成训练数据的原始数据，官方称其为决策列表（Judgment list），第一列是数值为0-4的权重，数值越大，相关性越高。回到我们最初的需求就是越多人点击的文档，那么这个权重就越大。第二列是queryid，同次查询结果中的queryid一样，第三列是文档id，这里就是电影id，第四列是文档标题，这里就是电影名。
4   qid:1 #    7555   Rambo
3  qid:1 #    1370   Rambo III
3  qid:1 #    1369   Rambo: First Blood Part II
3  qid:1 #    1368   First Blood
0  qid:1 #    136278 Blood
4  qid:2 #    1366   Rocky
3  qid:2 #    1246   Rocky Balboa
3  qid:2 #    60375  Rocky VI
3  qid:2 #    1371   Rocky III
3  qid:2 #    1375   Rocky V

log_features(es, judgments_dict=movieJudgments, search_index=INDEX_NAME)

build_features_judgments_file(movieJudgments, filename=JUDGMENTS_FILE_FEATURES)

之后就是生成特征集，就是把上面的每条训练数据根据特征查询语句扔进es里面进行查询，把得分放到1和2特征后面，如:下面数据第一条中的，1:12.318446就表示1特征的分数，2:10.573845表示2特征的分数，然后把特征集写到文件。
生成完的特征集如下：
4   qid:1  1:12.318446    2:10.573845 # 7555 rambo
3  qid:1  1:10.357836    2:11.950331 # 1370 rambo
3  qid:1  1:7.0104666    2:11.220029 # 1369 rambo
3  qid:1  1:0.0  2:11.220029 # 1368 rambo
0  qid:1  1:0.0  2:0.0 # 136278 rambo
4  qid:2  1:10.686367    2:8.814796 # 1366  rocky
3  qid:2  1:8.985519 2:9.984467 # 1246  rocky
3  qid:2  1:8.985519 2:8.067647 # 60375 rocky
3  qid:2  1:8.985519 2:5.6604943 # 1371 rocky
3  qid:2  1:8.985519 2:7.3007236 # 1375 rocky
特征集出来后就是训练了，demo提供10总不同的算法，训练好之后把结果传到es提供服务

for modelType in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]:

    # 0, MART

    # 1, RankNet

    # 2, RankBoost

    # 3, AdaRank

    # 4, coord Ascent

    # 6, LambdaMART

    # 7, ListNET

    # 8, Random Forests

    # 9, Linear Regression

    Logger.logger.info("*** Training %s " % modelType)

    train_model(judgments_with_features_file=JUDGMENTS_FILE_FEATURES, model_output='model.txt',

                which_model=modelType)

    save_model(script_name="test_%s" % modelType, feature_set=FEATURE_SET_NAME,       model_fname='model.txt')

4.最后搜索数据
python search.py Rambo
搜索时主要用到了es里面的rescore特性，就是对前面topn条记录根据模型进行再排序，查询dsl如下：

{

  "query": {

      "multi_match": {

          "query": "Rambo",

          "fields": ["title", "overview"]

       }

   },

  "rescore": {

      "query": {

        "rescore_query": {

            "sltr": {

                "params": {

                    "keywords": "Rambo"

                },

                "model": "test_1",

            }

         }

      }

   }

}

得到结果
Rambo
Rambo III
Rambo: First Blood Part II
First Blood
In the Line of Duty: The F.B.I. Murders
Son of Rambow
Spud
当然这个是最简单的一个例子，深入研究可以参考官方文档，很详细：https://elasticsearch-learning ... test/

Day 16 - Elasticsearch性能调优

白衬衣发表了文章 • 2 个评论 • 15100 次浏览 • 2018-12-16 20:10 • 来自相关话题

因为总是看到很多同学在说elasticsearch性能不够好，集群不够稳定，询问关于elasticsearch的调优，但是每次都是一个个点的单独讲，很多时候都是case by case的解答，今天简单梳理下日常的elasticsearch使用调优，以下仅为自己日常经验之谈，如有疏漏，还请大家帮忙指正。

一、配置文件调优

elasticsearch.yml

内存锁定

bootstrap.memory_lock：true 允许 JVM 锁住内存，禁止操作系统交换出去。

zen.discovery

Elasticsearch 默认被配置为使用单播发现，以防止节点无意中加入集群。组播发现应该永远不被使用在生产环境了，否则你得到的结果就是一个节点意外的加入到了你的生产环境，仅仅是因为他们收到了一个错误的组播信号。
ES是一个P2P类型的分布式系统，使用gossip协议，集群的任意请求都可以发送到集群的任一节点，然后es内部会找到需要转发的节点，并且与之进行通信。
在es1.x的版本，es默认是开启组播，启动es之后，可以快速将局域网内集群名称，默认端口的相同实例加入到一个大的集群，后续再es2.x之后，都调整成了单播，避免安全问题和网络风暴；
单播discovery.zen.ping.unicast.hosts，建议写入集群内所有的节点及端口，如果新实例加入集群，新实例只需要写入当前集群的实例，即可自动加入到当前集群，之后再处理原实例的配置即可，新实例加入集群，不需要重启原有实例；
节点zen相关配置：
discovery.zen.ping_timeout：判断master选举过程中，发现其他node存活的超时设置，主要影响选举的耗时，参数仅在加入或者选举 master 主节点的时候才起作用
discovery.zen.join_timeout：节点确定加入到集群中，向主节点发送加入请求的超时时间，默认为3s
discovery.zen.minimum_master_nodes：参与master选举的最小节点数，当集群能够被选为master的节点数量小于最小数量时，集群将无法正常选举。

故障检测（ fault detection ）

两种情况下回进行故障检测，第一种是由master向集群的所有其他节点发起ping，验证节点是否处于活动状态；第二种是：集群每个节点向master发起ping，判断master是否存活，是否需要发起选举。
故障检测需要配置以下设置使用
形如：
discovery.zen.fd.ping_interval 节点被ping的频率，默认为1s。
discovery.zen.fd.ping_timeout 等待ping响应的时间，默认为 30s，运行的集群中，master 检测所有节点，以及节点检测 master 是否正常。
discovery.zen.fd.ping_retries ping失败/超时多少导致节点被视为失败，默认为3。

https://www.elastic.co/guide/e ... .html

队列数量

不建议盲目加大es的队列数量，如果是偶发的因为数据突增，导致队列阻塞，加大队列size可以使用内存来缓存数据，如果是持续性的数据阻塞在队列，加大队列size除了加大内存占用，并不能有效提高数据写入速率，反而可能加大es宕机时候，在内存中可能丢失的上数据量。
哪些情况下，加大队列size呢？GET /_cat/thread_pool，观察api中返回的queue和rejected，如果确实存在队列拒绝或者是持续的queue，可以酌情调整队列size。

https://www.elastic.co/guide/e ... .html

内存使用

设置indices的内存熔断相关参数，根据实际情况进行调整，防止写入或查询压力过高导致OOM，
indices.breaker.total.limit: 50%，集群级别的断路器，默认为jvm堆的70%；
indices.breaker.request.limit: 10%，单个request的断路器限制，默认为jvm堆的60%；
indices.breaker.fielddata.limit: 10%，fielddata breaker限制，默认为jvm堆的60%。

https://www.elastic.co/guide/e ... .html

根据实际情况调整查询占用cache，避免查询cache占用过多的jvm内存，参数为静态的，需要在每个数据节点配置。
indices.queries.cache.size: 5%，控制过滤器缓存的内存大小，默认为10%。接受百分比值，5%或者精确值，例如512mb。

https://www.elastic.co/guide/e ... .html

创建shard

如果集群规模较大，可以阻止新建shard时扫描集群内全部shard的元数据，提升shard分配速度。
cluster.routing.allocation.disk.include_relocations: false，默认为true。

https://www.elastic.co/guide/e ... .html

二、系统层面调优

jdk版本

当前根据官方建议，选择匹配的jdk版本；

jdk内存配置

首先，-Xms和-Xmx设置为相同的值，避免在运行过程中再进行内存分配，同时，如果系统内存小于64G，建议设置略小于机器内存的一半，剩余留给系统使用。
同时，jvm heap建议不要超过32G（不同jdk版本具体的值会略有不同），否则jvm会因为内存指针压缩导致内存浪费，详见：
https://www.elastic.co/guide/c ... .html

交换分区

关闭交换分区，防止内存发生交换导致性能下降（部分情况下，宁死勿慢）
swapoff -a

文件句柄

Lucene 使用了大量的文件。同时，Elasticsearch 在节点和 HTTP 客户端之间进行通信也使用了大量的套接字，所有这一切都需要足够的文件描述符，默认情况下，linux默认运行单个进程打开1024个文件句柄，这显然是不够的，故需要加大文件句柄数
ulimit -n 65536

https://www.elastic.co/guide/e ... .html

mmap

Elasticsearch 对各种文件混合使用了 NioFs（注：非阻塞文件系统）和 MMapFs （注：内存映射文件系统）。请确保你配置的最大映射数量，以便有足够的虚拟内存可用于 mmapped 文件。这可以暂时设置：
sysctl -w vm.max_map_count=262144
或者你可以在 /etc/sysctl.conf 通过修改 vm.max_map_count 永久设置它。

https://www.elastic.co/guide/c ... .html

磁盘

如果你正在使用 SSDs，确保你的系统 I/O 调度程序是配置正确的。当你向硬盘写数据，I/O 调度程序决定何时把数据实际发送到硬盘。大多数默认 nix 发行版下的调度程序都叫做 cfq（完全公平队列）。但它是为旋转介质优化的：机械硬盘的固有特性意味着它写入数据到基于物理布局的硬盘会更高效。
这对 SSD 来说是低效的，尽管这里没有涉及到机械硬盘。但是，deadline 或者 noop 应该被使用。deadline 调度程序基于写入等待时间进行优化， noop 只是一个简单的 FIFO 队列。
echo noop > /sys/block/sd/queue/scheduler

磁盘挂载

mount -o noatime,data=writeback,barrier=0,nobh /dev/sd* /esdata*
其中，noatime，禁止记录访问时间戳；data=writeback，不记录journal；barrier=0，因为关闭了journal，所以同步关闭barrier；
nobh，关闭buffer_head，防止内核影响数据IO

磁盘其他注意事项

使用 RAID 0。条带化 RAID 会提高磁盘I/O，代价显然就是当一块硬盘故障时整个就故障了，不要使用镜像或者奇偶校验 RAID 因为副本已经提供了这个功能。
另外，使用多块硬盘，并允许 Elasticsearch 通过多个 path.data 目录配置把数据条带化分配到它们上面。
不要使用远程挂载的存储，比如 NFS 或者 SMB/CIFS。这个引入的延迟对性能来说完全是背道而驰的。

三、elasticsearch使用方式调优

当elasticsearch本身的配置没有明显的问题之后，发现es使用还是非常慢，这个时候，就需要我们去定位es本身的问题了，首先祭出定位问题的第一个命令：

hot_threads

GET /_nodes/hot_threads&interval=30s

抓取30s的节点上占用资源的热线程，并通过排查占用资源最多的TOP线程来判断对应的资源消耗是否正常，一般情况下，bulk，search类的线程占用资源都可能是业务造成的，但是如果是merge线程占用了大量的资源，就应该考虑是不是创建index或者刷磁盘间隔太小，批量写入size太小造成的。

https://www.elastic.co/guide/e ... .html

pending_tasks

GET /_cluster/pending_tasks

有一些任务只能由主节点去处理，比如创建一个新的索引或者在集群中移动分片，由于一个集群中只能有一个主节点，所以只有这一master节点可以处理集群级别的元数据变动。在99.9999%的时间里，这不会有什么问题，元数据变动的队列基本上保持为零。在一些罕见的集群里，元数据变动的次数比主节点能处理的还快，这会导致等待中的操作会累积成队列。这个时候可以通过pending_tasks api分析当前什么操作阻塞了es的队列，比如，集群异常时，会有大量的shard在recovery，如果集群在大量创建新字段，会出现大量的put_mappings的操作，所以正常情况下，需要禁用动态mapping。

https://www.elastic.co/guide/e ... .html

字段存储

当前es主要有doc_values，fielddata，storefield三种类型，大部分情况下，并不需要三种类型都存储，可根据实际场景进行调整：
当前用得最多的就是doc_values，列存储，对于不需要进行分词的字段，都可以开启doc_values来进行存储（且只保留keyword字段），节约内存，当然，开启doc_values会对查询性能有一定的影响，但是，这个性能损耗是比较小的，而且是值得的；

fielddata构建和管理 100% 在内存中，常驻于 JVM 内存堆，所以可用于快速查询，但是这也意味着它本质上是不可扩展的，有很多边缘情况下要提防，如果对于字段没有分析需求，可以关闭fielddata；

storefield主要用于_source字段，默认情况下，数据在写入es的时候，es会将doc数据存储为_source字段，查询时可以通过_source字段快速获取doc的原始结构，如果没有update，reindex等需求，可以将_source字段disable；

_all，ES在6.x以前的版本，默认将写入的字段拼接成一个大的字符串，并对该字段进行分词，用于支持整个doc的全文检索，在知道doc字段名称的情况下，建议关闭掉该字段，节约存储空间，也避免不带字段key的全文检索；

norms：搜索时进行评分，日志场景一般不需要评分，建议关闭；

tranlog

Elasticsearch 2.0之后为了保证不丢数据，每次 index、bulk、delete、update 完成的时候，一定触发刷新 translog 到磁盘上，才给请求返回 200 OK。这个改变在提高数据安全性的同时当然也降低了一点性能。
如果你不在意这点可能性，还是希望性能优先，可以在 index template 里设置如下参数：
 { "index.translog.durability": "async" } 

index.translog.sync_interval：对于一些大容量的偶尔丢失几秒数据问题也并不严重的集群，使用异步的 fsync 还是比较有益的。比如，写入的数据被缓存到内存中，再每5秒执行一次 fsync ，默认为5s。小于的值100ms是不允许的。
index.translog.flush_threshold_size：translog存储尚未安全保存在Lucene中的所有操作。虽然这些操作可用于读取，但如果要关闭并且必须恢复，则需要重新编制索引。此设置控制这些操作的最大总大小，以防止恢复时间过长。达到设置的最大size后，将发生刷新，生成新的Lucene提交点，默认为512mb。

refresh_interval

执行刷新操作的频率，这会使索引的最近更改对搜索可见，默认为1s，可以设置-1为禁用刷新，对于写入速率要求较高的场景，可以适当的加大对应的时长，减小磁盘io和segment的生成；

禁止动态mapping

动态mapping的坏处：
1.造成集群元数据一直变更，导致集群不稳定；
2.可能造成数据类型与实际类型不一致；
3.对于一些异常字段或者是扫描类的字段，也会频繁的修改mapping，导致业务不可控。
动态mapping配置的可选值及含义如下：
true：支持动态扩展，新增数据有新的字段属性时，自动添加对于的mapping，数据写入成功
false：不支持动态扩展，新增数据有新的字段属性时，直接忽略，数据写入成功
strict：不支持动态扩展，新增数据有新的字段时，报错，数据写入失败

批量写入

批量请求显然会大大提升写入速率，且这个速率是可以量化的，官方建议每次批量的数据物理字节数5-15MB是一个比较不错的起点，注意这里说的是物理字节数大小。文档计数对批量大小来说不是一个好指标。比如说，如果你每次批量索引 1000 个文档，记住下面的事实：
1000 个 1 KB 大小的文档加起来是 1 MB 大。
1000 个 100 KB 大小的文档加起来是 100 MB 大。
这可是完完全全不一样的批量大小了。批量请求需要在协调节点上加载进内存，所以批量请求的物理大小比文档计数重要得多。
从 5–15 MB 开始测试批量请求大小，缓慢增加这个数字，直到你看不到性能提升为止。然后开始增加你的批量写入的并发度（多线程等等办法）。
用iostat 、 top 和 ps 等工具监控你的节点，观察资源什么时候达到瓶颈。如果你开始收到 EsRejectedExecutionException ，你的集群没办法再继续了：至少有一种资源到瓶颈了。或者减少并发数，或者提供更多的受限资源（比如从机械磁盘换成 SSD），或者添加更多节点。

索引和shard

es的索引，shard都会有对应的元数据，且因为es的元数据都是保存在master节点，且元数据的更新是要hang住集群向所有节点同步的，当es的新建字段或者新建索引的时候，都会要获取集群元数据，并对元数据进行变更及同步，此时会影响集群的响应，所以需要关注集群的index和shard数量，建议如下：
1.使用shrink和rollover api，相对生成合适的数据shard数；
2.根据数据量级及对应的性能需求，选择创建index的名称，形如：按月生成索引：test-YYYYMM，按天生成索引：test-YYYYMMDD；
3.控制单个shard的size，正常情况下，日志场景，建议单个shard不大于50GB，线上业务场景，建议单个shard不超过20GB；

segment merge

段合并的计算量庞大，而且还要吃掉大量磁盘 I/O。合并在后台定期操作，因为他们可能要很长时间才能完成，尤其是比较大的段。这个通常来说都没问题，因为大规模段合并的概率是很小的。
如果发现merge占用了大量的资源，可以设置：
index.merge.scheduler.max_thread_count: 1
特别是机械磁盘在并发 I/O 支持方面比较差，所以我们需要降低每个索引并发访问磁盘的线程数。这个设置允许 max_thread_count + 2 个线程同时进行磁盘操作，也就是设置为 1 允许三个线程。
对于 SSD，你可以忽略这个设置，默认是 Math.min(3, Runtime.getRuntime().availableProcessors() / 2) ，对 SSD 来说运行的很好。
业务低峰期通过force_merge强制合并segment，降低segment的数量，减小内存消耗；
关闭冷索引，业务需要的时候再进行开启，如果一直不使用的索引，可以定期删除，或者备份到hadoop集群；

自动生成_id

当写入端使用特定的id将数据写入es时，es会去检查对应的index下是否存在相同的id，这个操作会随着文档数量的增加而消耗越来越大，所以如果业务上没有强需求，建议使用es自动生成的id，加快写入速率。

routing

对于数据量较大的业务查询场景，es侧一般会创建多个shard，并将shard分配到集群中的多个实例来分摊压力，正常情况下，一个查询会遍历查询所有的shard，然后将查询到的结果进行merge之后，再返回给查询端。此时，写入的时候设置routing，可以避免每次查询都遍历全量shard，而是查询的时候也指定对应的routingkey，这种情况下，es会只去查询对应的shard，可以大幅度降低合并数据和调度全量shard的开销。

使用alias

生产提供服务的索引，切记使用别名提供服务，而不是直接暴露索引名称，避免后续因为业务变更或者索引数据需要reindex等情况造成业务中断。

避免宽表

在索引中定义太多字段是一种可能导致映射爆炸的情况，这可能导致内存不足错误和难以恢复的情况，这个问题可能比预期更常见，index.mapping.total_fields.limit ，默认值是1000

避免稀疏索引

因为索引稀疏之后，对应的相邻文档id的delta值会很大，lucene基于文档id做delta编码压缩导致压缩率降低，从而导致索引文件增大，同时，es的keyword，数组类型采用doc_values结构，每个文档都会占用一定的空间，即使字段是空值，所以稀疏索引会造成磁盘size增大，导致查询和写入效率降低。

Day 13 - Elasticsearch-Hadoop打通Elasticsearch和Hadoop

Jasonbian 发表了文章 • 3 个评论 • 9731 次浏览 • 2018-12-13 09:34 • 来自相关话题

ES-Hadoop打通Elasticsearch和Hadoop

介绍

Elasticsearch作为强大的搜索引擎，Hadoop HDFS是分布式文件系统。

ES-Hadoop是一个深度集成Hadoop和ElasticSearch的项目，也是ES官方来维护的一个子项目。Elasticsearch可以将自身的Document导入到HDFS中用作备份；同时也可以将存储在HDFS上的结构化文件导入为ES中的Document，通过实现Hadoop和ES之间的输入输出，可以在Hadoop里面对ES集群的数据进行读取和写入，充分发挥Map-Reduce并行处理的优势，为Hadoop数据带来实时搜索的可能。

ES-Hadoop插件支持Map-Reduce、Cascading、Hive、Pig、Spark、Storm、yarn等组件。

ES-Hadoop整个数据流转图如下：

环境配置

Elasticsearch 5.0.2
Centos 7
elasticsearch-hadoop 5.0.2
repository-hdfs-5.0.2

Elasticsearch备份数据到HDFS

介绍

Elasticsearch副本提供了数据高可靠性，在部分节点丢失的情况下不中断服务；但是副本并不提供对灾难性故障的保护，同时在运维人员误操作情况下也不能保障数据的可恢复性。对于这种情况，我们需要对Elasticsearch集群数据的真正备份。

通过快照的方式，将Elasticsearch集群中的数据备份到HDFS上，这样数据既存在于Elasticsearch集群中，有存在于HDFS上。当ES集群出现不可恢复的故障时，可以将数据从HDFS上快速恢复。

操作步骤
下载插件 <https://artifacts.elastic.co/d ... gt%3B 保存在/usr/local下
安装插件

 cd /usr/local/es/elasticsearch-5.0.2/bin ./elasticsearch-plugin install file:///usr/local/repository-hdfs-5.0.2.zip 
安装成功后需要重启Elasticsearch

备份与恢复
构建一个仓库

 PUT <a href="http://192.168.10.74:9200/_snapshot/backup" rel="nofollow" target="_blank">http://192.168.10.74:9200/_snapshot/backup</a> { "type": "hdfs", "settings": { "uri": "hdfs://192.168.10.170:9000", "path": "/es", "conf_location": "/usr/local/hadoop/etc/hadoop/hdfs-site.xml" } } 
备份快照

 PUT <a href="http://192.168.10.74:9200/_snapshot/backup/snapshot_users?wait_for_completion=true" rel="nofollow" target="_blank">http://192.168.10.74:9200/_sna ... Dtrue</a> { "indices": "users", //备份users的index，注意不设置这个属性，默认是备份所有index "ignore_unavailable": true, "include_global_state": false } 
恢复快照

 POST <a href="http://192.168.10.74:9200/_snapshot/backup/snapshot_users/_restore" rel="nofollow" target="_blank">http://192.168.10.74:9200/_sna ... store</a> { "indices": "users", //指定索引恢复，不指定就是所有 "ignore_unavailable": true, //忽略恢复时异常索引 "include_global_state": false //是否存储全局转态信息,fasle代表有一个或几个失败，不会导致整个任务失败 } 

整合Spark与Elasticsearch

整体思路
数据首先存储在HDFS上，可以通过Spark SQL直接导入到ES中
Spark SQL可以直接通过建立Dataframe或者临时表连接ES，达到搜索优化、减少数据量和数据筛选的目的，此时数据只在ES内存中而不再Spark SQL中
筛选后的数据重新导入到Spark SQL中进行查询

引入依赖

```java

org.elasticsearch
elasticsearch-hadoop
5.0.2

```

### 具体流程
数据在HDFS上，数据存储在HDFS的每个DataNode的block上
数据加载到Spark SQL
- 数据从HDFS加载到Spark SQL中，以RDD形式存储
 
 java JavaRDD<String> textFile = spark.read().textFile("hdfs://192.168.10.170:9000/csv/user.csv") 
- 添加数据结构信息转换为新的RDD
 
 java JavaRDD<UserItem> dataSplits = textFile.map(line -> { String records = line.toString().trim(); String record = records.substring(0,records.length() - 1).trim(); String[] parts = record.split("\\|"); UserItem u = new UserItem(); u.setName(parts[0]); u.setAge(parts[1]); u.setHeight(parts[2]); return u; }); 
- 根据新的RDD创建DataFrame
 
 java DataSet<Row> ds = spark.createDataFrame(dataSplits, UserItem.class); 
- 由Dataset创建索引，并写入ES
 
  JavaEsSparkSQL.saveToEs(ds, "es_spark/users"); 
数据在ES中建立索引
Spark SQL通过索引对ES中的数据进行查询

java SparkSession spark = SparkSession.builder().appName("es-spark").master("local").config("es.index.auto.create", true).getOrCreate(); Map<String, String> options = new HashMap<>(); options.put("pushdown", "true"); options.put("es.nodes","192.168.10.74:9200"); Dataset<Row> df = spark.read().options(options).format("org.elasticsearch.spark.sql").load("es_spark/users"); df.createOrReplaceTempView("users"); Dataset<Row> userSet = spark.sql("SELECT name FORM users WHERE age >=10 AND age <= 20"); userSet.show(); 

结束

ES-Hadoop无缝打通了ES和Hadoop两个非常优秀的框架，从而让ES的强大检索性能帮助我们快速分析海量数据。

Day 12 - Elasticsearch日志场景最佳实践

ginger 发表了文章 • 0 个评论 • 7488 次浏览 • 2018-12-12 16:35 • 来自相关话题

1. 背景

Elasticsearch可广泛应用于日志分析、全文检索、结构化数据分析等多种场景，大幅度降低维护多套专用系统的成本，在开源社区非常受欢迎。然而Elasticsearch为满足多种不同的使用场景，底层组合使用了多种数据结构，部分数据结构对具体的用户使用场景可能是冗余的，从而导致默认情况下无法达到性能和成本最优化。
幸运的是，Elasticsearch提供非常灵活的模板配置能力，用户可以按需进行优化。多数情况下，用户结合使用场景进行优化后，Elasticsearch的性能都会有数倍的提升，成本也对应有倍数级别的下降。本文主要介绍不同日志使用场景下的调优经验。

2. 日志处理基本流程

日志处理的基本流程包含：日志采集 -> 数据清洗 -> 存储 -> 可视化分析。Elastic Stack提供完整的日志解决方案，帮助用户完成对日志处理全链路的管理，推荐大家使用。每个流程的处理如下：

日志采集：从业务所在的机器上，较实时的采集日志传递给下游。常用开源组件如Beats、Logstash、Fluentd等。
数据清洗：利用正则解析等机制，完成日志从文本数据到结构化数据的转换。用户可使用Logstash 或 Elasticsearch Ingest模块等完成数据清洗。
存储：使用Elasticsearch对数据进行持久存储，并提供全文搜索和分析能力。
可视化分析：通过图形界面，完成对日志的搜索分析，常用的开源组件如Kibana、Grafana。
![](http://km.oa.com/files/photos/ ... 97.png)

使用Elastic Stack处理日志的详细过程，用户可参考官方文章[Getting started with the Elastic Stack](https://www.elastic.co/guide/e ... k.html)，这里不展开介绍。

3. 日志场景调优

对于Elasticsearch的通用调优，之前分享的文章[Elasticsearch调优实践](https://cloud.tencent.com/developer/article/1156231)，详细介绍了Elasticsearch在性能、稳定性方面的调优经验。而对于日志场景，不同的场景使用方式差别较大，这里主要介绍常见使用方式下，性能和成本的优化思路。

3.1 基础场景

对于多数简单日志使用场景，用户一般只要求存储原始日志，并提供按关键字搜索日志记录的能力。对于此类场景，用户可跳过数据清洗阶段，并参考如下方式进行优化：
建议打开最优压缩，一般可降低40%存储。
设置原始日志字段（message）为text，去除keyword类型子字段，提供全文搜索能力，降低存储。
关闭_all索引，前面已通过message提供全文搜索能力。
对于其他字符串字段，统一设置为keyword类型，避免默认情况下字符串字段同时存储text、keyword两种类型的数据。
使用开源组件（如Beats）上报数据时会包含较多辅助信息，用户可通过修改组件配置文件进行裁剪。

这样去除message的keyword子字段、_all等冗余信息后，再加上最优压缩，可以保证数据相对精简。下面给出这类场景的常用模板，供用户参考：
 { "order": 5, "template": "my_log_*", "settings": { "translog.durability": "async", "translog.sync_interval": "5s", "index.refresh_interval": "30s", "index.codec": "best_compression" # 最优压缩 }, "mappings": { "_default_": { "_all": { # 关闭_all索引 "enabled": false }, "dynamic_templates": [ { "log": { # 原始日志字段，分词建立索引 "match": "message", "mapping": { "type": "text" } } }, { "strings": { # 其他字符串字段，统一设置为keyword类型 "match_mapping_type": "string", "mapping": { "type": "keyword" } } } ] } } } 

3.2 精准搜索场景

对于部分用户，普通的全文检索并不能满足需求，希望精准搜索日志中的某部分，例如每条日志中包含程序运行时多个阶段的耗时数据，对具体一个阶段的耗时进行搜索就比较麻烦。对于此类场景，用户可基于基础场景，进行如下调整：
清洗过程中，可仅解析出需要精准搜索的部分作为独立字段，用于精准搜索。
对于精准搜索字段，如果无排序/聚合需求，可以关闭doc_values；对于字符串，一般使用keyword，可按需考虑使用text。

下面给出这类场景的常用模板，供用户参考：
 { "order": 5, "template": "my_log_*", "settings": { "translog.durability": "async", "translog.sync_interval": "5s", "index.refresh_interval": "30s", "index.codec": "best_compression" # 最优压缩 }, "mappings": { "_default_": { "_all": { # 关闭_all索引 "enabled": false }, "dynamic_templates": [ { "log": { # 原始日志字段，分词建立索引 "match": "message", "mapping": { "type": "text" } } }, { "precise_fieldx": { # 精准搜索字段 "match": "fieldx", "mapping": { "type": "keyword", "doc_values": false } } }, { "strings": { # 其他字符串字段，统一设置为keyword类型 "match_mapping_type": "string", "mapping": { "type": "keyword" } } } ] } } } 

3.3 统计分析场景

对于某些场景，日志包含的主要是程序运行时输出的统计信息，用户通常会完全解析日志进行精确查询、统计分析，而是否保存原始日志关系不大。对于此类场景，用户可进行如下调整：
清洗过程中，解析出所有需要的数据作为独立字段；原始日志非必要时，建议去除。
如果有强需求保留原始日志，可以设置该字段enabled属性为false，只存储不索引。
多数字段保持默认即可，会自动建立索引、打开doc_values，可用于查询、排序、聚合。
对部分无排序/聚合需求、开销高的字段，可以关闭doc_values。

下面给出这类场景的常用模板，供用户参考：

 { "order": 5, "template": "my_log_*", "settings": { "translog.durability": "async", "translog.sync_interval": "5s", "index.refresh_interval": "30s", "index.codec": "best_compression" # 最优压缩 }, "mappings": { "_default_": { "_all": { # 关闭_all索引 "enabled": false }, "dynamic_templates": [ { "log": { # 原始日志字段，关闭索引 "match": "message", "mapping": { "enabled": false } } }, { "index_only_fieldx": { # 仅索引的字段，无排序/聚合需求 "match": "fieldx", "mapping": { "type": "keyword", "doc_values": false } } }, { "strings": { # 其他字符串字段，统一设置为keyword类型 "match_mapping_type": "string", "mapping": { "type": "keyword" } } } ] } } } 

ES 5.1及之后的版本，支持关键字查询时[自动选择目标字段](https://www.elastic.co/guide/e ... _field)，用户没有必要再使用原始日志字段提供不指定字段进行查询的能力。

4. 小结

日志的使用方式比较灵活，本文结合常见的客户使用方式，从整体上对性能、成本进行优化。用户也可结合自身业务场景，参考文章[Elasticsearch调优实践](https://cloud.tencent.com/developer/article/1156231)进行更细致的优化。

Day 10 - Elasticsearch 分片恢复并发数过大引发的bug分析

howardhuang 发表了文章 • 4 个评论 • 11399 次浏览 • 2018-12-10 11:43 • 来自相关话题

大家好，今天为大家分享一次 ES 的填坑经验。主要是关于集群恢复过程中，分片恢复并发数调整过大导致集群 hang 死的问题。

场景描述

废话不多说，先来描述场景。某日，腾讯云线上某 ES 集群，15个节点，2700+ 索引，15000+ 分片，数十 TB 数据。由于机器故障，某个节点被重启，此时集群有大量的 unassigned 分片，集群处于 yellow 状态。为了加快集群恢复的速度，手动调整分片恢复并发数，原本想将默认值为2的 node_concurrent_recoveries 调整为10，结果手一抖多加了一个0，设定了如下参数：

 curl -X PUT "localhost:9200/_cluster/settings" -H 'Content-Type: application/json' -d' { "persistent": { "cluster.routing.allocation.node_concurrent_recoveries": 100, "indices.recovery.max_bytes_per_sec": "40mb" } } ' 
设定之后，观察集群 unassigned 分片，一开始下降的速度很快。大约几分钟后，数量维持在一个固定值不变了，然后，然后就没有然后了，集群所有节点 generic 线程池卡死，虽然已存在的索引读写没问题，但是新建索引以及所有涉及 generic 线程池的操作全部卡住。立马修改分片恢复并发数到10，通过管控平台一把重启了全部节点，约15分钟后集群恢复正常。接下来会先介绍一些基本的概念，然后再重现这个问题并做详细分析。

基本概念

ES 线程池（thread pool）

ES 中每个节点有多种线程池，各有用途。重要的有：

generic ：通用线程池，后台的 node discovery，上述的分片恢复（node recovery）等等一些通用后台的操作都会用到该线程池。该线程池线程数量默认为配置的处理器数量（processors）* 4，最小128，最大512。
index ：index/delete 等索引操作会用到该线程池，包括自动创建索引等。默认线程数量为配置的处理器数量，默认队列大小：200.
search ：查询请求处理线程池。默认线程数量：int((# of available_processors * 3) / 2) + 1，默认队列大小：1000.
get ：get 请求处理线程池。默认线程数量为配置的处理器数量，默认队列大小：1000.
write ：单个文档的 index/delete/update 以及 bulk 请求处理线程。默认线程数量为配置的处理器数量，默认队列大小：200，在写多的日志场景我们一般会将队列调大。
还有其它线程池，例如备份回档（snapshot）、analyze、refresh 等，这里就不一一介绍了。详细可参考官方文档：https://www.elastic.co/guide/e ... .html

集群恢复之分片恢复

我们知道 ES 集群状态分为三种，green、yellow、red。green 状态表示所有分片包括主副本均正常被分配；yellow 状态表示所有主分片已分配，但是有部分副本分片未分配；red 表示有部分主分片未分配。
一般当集群中某个节点因故障失联或者重启之后，如果集群索引有副本的场景，集群将进入分片恢复阶段（recovery）。此时一般是 master 节点发起更新集群元数据任务，分片的分配策略由 master 决定，具体分配策略可以参考腾讯云+社区的这篇文章了解细节：https://cloud.tencent.com/deve ... 34743 。各节点收到集群元数据更新请求，检查分片状态并触发分片恢复流程，根据分片数据所在的位置，有多种恢复的方式，主要有以下几种：
EXISTING_STORE ：数据在节点本地存在，从本地节点恢复。
PEER ：本地数据不可用或不存在，从远端节点（源分片，一般是主分片）恢复。
SNAPSHOT ：数据从备份仓库恢复。
LOCAL_SHARDS ：分片合并（shrink）场景，从本地别的分片恢复。

PEER 场景分片恢复并发数主要由如下参数控制：
cluster.routing.allocation.node_concurrent_incoming_recoveries：节点上最大接受的分片恢复并发数。一般指分片从其它节点恢复至本节点。
cluster.routing.allocation.node_concurrent_outgoing_recoveries ：节点上最大发送的分片恢复并发数。一般指分片从本节点恢复至其它节点。
cluster.routing.allocation.node_concurrent_recoveries ：该参数同时设置上述接受发送分片恢复并发数为相同的值。
详细参数可参考官方文档：https://www.elastic.co/guide/e ... .html

 集群卡住的主要原因就是从远端节点恢复（PEER Recovery）的并发数过多，导致 generic 线程池被用完。涉及目标节点（target）和源节点（source）的恢复交互流程，后面分析问题时我们再来详细讨论。

问题复现与剖析

 为了便于描述，我用 ES 6.4.3版本重新搭建了一个三节点的集群。单节点 1 core，2GB memory。新建了300个 index，单个 index 5个分片一个副本，共 3000 个 shard。每个 index 插入大约100条数据。
先设定分片恢复并发数，为了夸张一点，我直接调整到200，如下所示：
 curl -X PUT "localhost:9200/_cluster/settings" -H 'Content-Type: application/json' -d' { "persistent": { "cluster.routing.allocation.node_concurrent_recoveries": 200 // 设定分片恢复并发数 } } ' 

接下来停掉某节点，模拟机器挂掉场景。几分钟后，观察集群分片恢复数量，卡在固定数值不再变化：

通过 allocation explain 查看分片分配状态，未分配的原因是受到最大恢复并发数的限制：

观察线程池的数量，generic 线程池打满128.

此时查询或写入已有索引不受影响，但是新建索引这种涉及到 generic 线程池的操作都会卡住。
通过堆栈分析，128 个 generic 线程全部卡在 PEER recovery 阶段。

 现象有了，我们来分析一下这种场景，远程分片恢复（PEER Recovery）流程为什么会导致集群卡住。

 当集群中有分片的状态发生变更时，master 节点会发起集群元数据更新（cluster state update）请求给所有节点。其它节点收到该请求后，感知到分片状态的变更，启动分片恢复流程。部分分片需要从其它节点恢复，代码层面，涉及分片分配的目标节点（target）和源节点（source）的交互流程如下：

 6.x 版本之后引入了 seqNo，恢复会涉及到 seqNo+translog，这也是6.x提升恢复速度的一大改进。我们重点关注流程中第 2、4、5、7、10、12 步骤中的远程调用，他们的作用分别是：
第2步：分片分配的目标节点向源节点（一般是主分片）发起分片恢复请求，携带起始 seqNo 和 syncId。
第4步：发送数据文件信息，告知目标节点待接收的文件清单。
第5步：发送 diff 数据文件给目标节点。
第7步：源节点发送 prepare translog 请求给目标节点，等目标节点打开 shard level 引擎，准备接受 translog。
第10步：源节点发送指定范围的 translog 快照给目标节点。
第12步：结束恢复流程。

       我们可以看到除第5步发送数据文件外，多次远程交互 submitRequest 都会调用 txGet，这个调用底层用的是基于 AQS 改造过的 sync 对象，是一个同步调用。如果一端 generic 线程池被这些请求打满，发出的请求等待对端返回，而发出的这些请求由于对端 generic 线程池同样的原因被打满，只能 pending 在队列中，这样两边的线程池都满了而且相互等待对端队列中的线程返回，就出现了分布式死锁现象。

问题处理

       为了避免改动太大带来不确定的 side effect，针对腾讯云 ES 集群我们目前先在 rest 层拒掉了并发数超过一定值的参数设定请求并提醒用户。与此同时，我们向官方提交了 issue：https://github.com/elastic/ela ... 36195 进行跟踪。

总结

       本文旨在描述集群恢复过程出现的集群卡死场景，避免更多的 ES 用户踩坑，没有对整体分片恢复做详细的分析，大家想了解详细的分片恢复流程可以参考腾讯云+社区 Elasticsearch 专栏相关的文章：https://cloud.tencent.com/developer/column/2428

完结，谢谢！

Day 9 - 动手实现一个自定义beat

Xinglong 发表了文章 • 0 个评论 • 4004 次浏览 • 2018-12-09 21:09 • 来自相关话题

参考

https://elasticsearch.cn/article/113

https://www.elastic.co/blog/build-your-own-beat

介绍

公司内部有统一的log收集系统，并且实现了定制的filebeat进行log收集。为了实现实时报警和监控，自定义的beat并没有直接把输出发给elasticsearch后端，而是中间会经过storm或者durid进行实时分析，然后落入es或者hdfs。同时由于是统一log收集，所以目前还没有针对具体的不同应用进行log的内容的切分，导致所有的log都是以一行为单位落入后端存储。于是需要针对不同的业务部门定制不同的beat。
本文初步尝试定制一个可以在beat端解析hdfs audit log的beat，限于篇幅，只实现了基本的文件解析功能。下面会从环境配置，代码实现，运行测试三个方面进行讲解。

环境配置

go version go1.9.4 linux/amd64

python version: 2.7.9

不得不吐槽下python的安装，各种坑。因为系统默认的python版本是2.7.5，而cookiecutter建议使用2.7.9

下面的工具会提供python本身需要依赖的一些native包

 yum install openssl -y yum install openssl-devel -y yum install zlib-devel -y yum install zlib -y 

安装python

 wget <a href="https://www.python.org/ftp/python/2.7.9/Python-2.7.9.tgz" rel="nofollow" target="_blank">https://www.python.org/ftp/pyt ... 9.tgz</a> tar -zxvf Python-2.7.9.tgz cd ~/python/Python-2.7.9 ./configure --prefix=/usr/local/python-2.7.9 make make install rm -f /bin/python ln -s /usr/local/python-2.7.9/bin/python /bin/python 

安装工具包 distribute, setuptools, pip

 cd ~/python/setuptools-19.6 && python setup.py install cd ~/python/pip-1.5.4 && python setup.py install cd ~/python/distribute-0.7.3 && python setup.py install 

安装cookiecutter

 pip install --user cookiecutter 

安装cookiecutter所依赖的工具

 pip install backports.functools-lru-cache pip install six pip install virtualenv 

*** virtualenv 安装好了之后，所在目录是在python的目录里面 (/usr/local/python-2.7.9/bin/virtualenv)，需要配置好PATH，这个工具稍后会被beat的Makefile用到

代码实现

需要实现的功能比较简单，目标是打开hdfs-audit.log文件，然后逐行读取，同时解析出必要的信息，组装成event，然后发送出去，如果对接的es的话，需要同时支持自动在es端创建正确的mapping

使用官方提供的beat模板创建自己的beat

需要设置好环境变量$GOPATH，本例子中GOPATH=/root/go

 $ go get github.com/elastic/beats $ cd $GOPATH/src/github.com/elastic/beats $ git checkout 5.1 [root@minikube-2830379 suxingfate]# cookiecutter /root/go/src/github.com/elastic/beats/generate/beat project_name [Examplebeat]: hdfsauditbeat github_name [your-github-name]: suxingfate beat [hdfsauditbeat]: beat_path [github.com/suxingfate]: full_name [Firstname Lastname]: xinglong make setup 

到这里，模板就生成了，然后就是定制需要的东西。
1 _meta/beat.yml # 配置模板文件，定义我们的beat会接受哪些配置项
2 config/config.go #使用go的struct定义整个config对象，包含所有的配置项
3 beater/hdfsauditbeat.go # 核心逻辑代码
4 _meta/fields.yml #这里是跟es对接的时候给es定义的mapping

1 _meta/beat.yml

这里增加了path，为后面配置hdfs-audit.log文件的位置留好坑
```
[root@minikube-2830379 hdfsauditbeat]# cat _meta/beat.yml
################### Hdfsauditbeat Configuration Example #########################

############################# Hdfsauditbeat ######################################

hdfsauditbeat:

Defines how often an event is sent to the output

period: 1s
path: "."
```

2 config/config.go

这里把path定义到struct里面，后面核心代码就可以从config对象获得path了

 [root@minikube-2830379 hdfsauditbeat]# cat config/config.go // Config is put into a different package to prevent cyclic imports in case // it is needed in several locations package config import "time" type Config struct { Period time.Duration `config:"period"` Path string `config:"path"` } var DefaultConfig = Config{ Period: 1 * time.Second, Path: ".", } 

3 beater/hdfsauditbeat.go

这里需要改动的地方是：
3.1 定义了一个catAudit函数来解析目标文件的每一行，同时生成自定义的event，然后发送出去
3.2 Run函数调用自定义的catAudit函数，从而把我们的功能嵌入

 [root@minikube-2830379 hdfsauditbeat]# cat beater/hdfsauditbeat.go package beater import ( "fmt" "time" "os" "io" "bufio" "strings" "github.com/elastic/beats/libbeat/beat" "github.com/elastic/beats/libbeat/common" "github.com/elastic/beats/libbeat/logp" "github.com/elastic/beats/libbeat/publisher" "github.com/suxingfate/hdfsauditbeat/config" ) type Hdfsauditbeat struct { done chan struct{} config config.Config client publisher.Client } // Creates beater func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) { config := config.DefaultConfig if err := cfg.Unpack(&config); err != nil { return nil, fmt.Errorf("Error reading config file: %v", err) } bt := &Hdfsauditbeat{ done: make(chan struct{}), config: config, } return bt, nil } func (bt *Hdfsauditbeat) Run(b *beat.Beat) error { logp.Info("hdfsauditbeat is running! Hit CTRL-C to stop it.") bt.client = b.Publisher.Connect() ticker := time.NewTicker(bt.config.Period) counter := 1 for { select { case <-bt.done: return nil case <-ticker.C: } bt.catAudit(bt.config.Path) logp.Info("Event sent") counter++ } } func (bt *Hdfsauditbeat) Stop() { bt.client.Close() close(bt.done) } func (bt *Hdfsauditbeat) catAudit(auditFile string) { file, err := os.OpenFile(auditFile, os.O_RDWR, 0666) if err != nil { //fmt.Println("Open file error!", err) return } defer file.Close() buf := bufio.NewReader(file) for { line, err := buf.ReadString('\n') line = strings.TrimSpace(line) if line == "" { return } timeEnd := strings.Index(line, ",") timeString := line[0 :timeEnd] tm, _ := time.Parse("2006-01-02 03:04:05", timeString) ugiStart := strings.Index(line, "ugi=") + 4 ugiEnd := strings.Index(line, " (auth") ugi := line[ugiStart :ugiEnd] cmdStart := strings.Index(line, "cmd=") + 4 line = line[cmdStart:len(line)] cmdEnd := strings.Index(line, " ") cmd := line[0 : cmdEnd] srcStart := strings.Index(line, "src=") + 4 line = line[srcStart:len(line)] srcEnd := strings.Index(line, " ") src := line[0:srcEnd] dstStart := strings.Index(line, "dst=") + 4 line = line[dstStart:len(line)] dstEnd := strings.Index(line, " ") dst := line[0:dstEnd] event := common.MapStr{ "@timestamp": common.Time(time.Unix(tm.Unix(), 0)), "ugi": ugi, "cmd": cmd, "src": src, "dst": dst, } bt.client.PublishEvent(event) if err != nil { if err == io.EOF { //fmt.Println("File read ok!") break } else { //fmt.Println("Read file error!", err) return } } } } 

4 _meat/fields.yml

```
[root@minikube-2830379 hdfsauditbeat]# less _meta/fields.yml
key: hdfsauditbeat
title: hdfsauditbeat
description:
fields:
- name: counter
 type: long
 required: true
 description: >
 PLEASE UPDATE DOCUMENTATION
 
 new fiels added hdfsaudit
- name: entrytime
 type: date
- name: ugi
 type: keyword
- name: cmd
 type: keyword
- name: src
 type: keyword
- name: dst
 type: keyword
```

测试

首先编译好项目
 make update make 
然后会发现生成了一个hdfsauditbeat文件，这个就是二进制的可执行文件。下面进行测试，这里偷了个懒，没有发给es，而是吐到console进行观察。
修改了一下配置文件，需要指定正确的需要消费的audit log文件的路径，另外就是修改了output为console

```
 [root@minikube-2830379 hdfsauditbeat]# cat hdfsauditbeat.yml
 ################### Hdfsauditbeat Configuration Example #########################
 
 ############################# Hdfsauditbeat ######################################
 
 hdfsauditbeat:
 
 Defines how often an event is sent to the output
 
 period: 1s
 path: "/root/go/hdfs-audit.log"
 
 ================================ General =====================================
 
 The name of the shipper that publishes the network data. It can be used to group
 
 all the transactions sent by a single shipper in the web interface.
 
 name:
 
 The tags of the shipper are included in their own field with each
 
 transaction published.
 
 tags: ["service-X", "web-tier"]
 
 Optional fields that you can specify to add additional information to the
 
 output.
 
 fields:
 
 env: staging
 
 ================================ Outputs =====================================
 
 Configure what outputs to use when sending the data collected by the beat.
 
 Multiple outputs may be used.
 
 -------------------------- Elasticsearch output ------------------------------
 
 output.elasticsearch:
 
 Array of hosts to connect to.
 
 hosts: ["localhost:9200"]
 
 Optional protocol and basic auth credentials.
 
 protocol: "https"
 
 username: "elastic"
 
 password: "changeme"
 
 ----------------------------- Logstash output --------------------------------
 
 output.logstash:
 
 The Logstash hosts
 
 hosts: ["localhost:5044"]
 
 Optional SSL. By default is off.
 
 List of root certificates for HTTPS server verifications
 
 ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
 Certificate for SSL client authentication
 
 ssl.certificate: "/etc/pki/client/cert.pem"
 
 Client Certificate Key
 
 ssl.key: "/etc/pki/client/cert.key"
 
 output.console:
 pretty: true
 
 ================================ Logging =====================================
 
 Sets log level. The default log level is info.
 
 Available log levels are: critical, error, warning, info, debug
 
 logging.level: debug
 
 At debug level, you can selectively enable logging only for some components.
 
 To enable all selectors use ["*"]. Examples of other selectors are "beat",
 
 "publish", "service".
 
 logging.selectors: ["*"]
 
  开始执行 
 [root@minikube-2830379 hdfsauditbeat]# ./hdfsauditbeat
 {
 "@timestamp": "2018-12-09T03:00:00.000Z",
 "beat": {
 "hostname": "minikube-2830379.lvs02.dev.abc.com",
 "name": "minikube-2830379.lvs02.dev.abc.com",
 "version": "5.1.3"
 },
 "cmd": "create",
 "dst": "null",
 "src": "/app-logs/app/logs/application_1540949675029_717305/lvsdpehdc25dn0444.stratus.lvs.abc.com_8042.tmp",
 "ugi": "appmon@APD.ABC.COM"
 }
 {
 "@timestamp": "2018-12-09T03:00:00.000Z",
 "beat": {
 "hostname": "minikube-2830379.lvs02.dev.abc.com",
 "name": "minikube-2830379.lvs02.dev.abc.com",
 "version": "5.1.3"
 },
 "cmd": "create",
 "dst": "null",
 "src": "/app-logs/appmon/logs/application_1540949675029_717305/lvsdpehdc25dn0444.stratus.lvs.abc.com_8042.tmp",
 "ugi": "appmon@APD.ABC.COM"
 }
```

结束

使用自定义beat给我们提供了很大的灵活性，虽然pipline或者logstash也可以做到，但是使用场景还是有很大差别的。如果是调用特殊的命令获得输出，或者是本文的场景都更适合定制化beat。

Day 8 - 如何使用Spark快速将数据写入Elasticsearch

Ricky Huo 发表了文章 • 0 个评论 • 11160 次浏览 • 2018-12-08 09:58 • 来自相关话题

如何使用Spark快速将数据写入Elasticsearch

说到数据写入Elasticsearch，最先想到的肯定是Logstash。Logstash因为其简单上手、可扩展、可伸缩等优点被广大用户接受。但是尺有所短，寸有所长，Logstash肯定也有它无法适用的应用场景，比如：

海量数据ETL
海量数据聚合
多源数据处理

为了满足这些场景，很多同学都会选择Spark，借助Spark算子进行数据处理，最后将处理结果写入Elasticsearch。

我们部门之前利用Spark对Nginx日志进行分析，统计我们的Web服务访问情况，将Nginx日志每分钟聚合一次最后将结果写入Elasticsearch，然后利用Kibana配置实时监控Dashboard。Elasticsearch和Kibana都很方便、实用，但是随着类似需求越来越多，如何快速通过Spark将数据写入Elasticsearch成为了我们的一大问题。

今天给大家推荐一款能够实现数据快速写入的黑科技——[Waterdrop](https://github.com/InterestingLab/waterdrop)，一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在Spark之上，简单易用，灵活配置，无需开发。

Kafka to Elasticsearch

和Logstash一样，Waterdrop同样支持多种类型的数据输入，这里我们以最常见的Kakfa作为输入源为例，讲解如何使用Waterdrop将数据快速写入Elasticsearch

Log Sample

原始日志格式如下:
 127.0.0.1 elasticsearch.cn 114.250.140.241 0.001s "127.0.0.1:80" [26/Oct/2018:21:54:32 +0800] "GET /article HTTP/1.1" 200 123 "-" - "Dalvik/2.1.0 (Linux; U; Android 7.1.1; OPPO R11 Build/NMF26X)" 

Elasticsearch Document

我们想要统计，一分钟每个域名的访问情况，聚合完的数据有以下字段:
 domain String hostname String status int datetime String count int 

Waterdrop with Elasticsearch

接下来会给大家详细介绍，我们如何通过Waterdrop读取Kafka中的数据，对数据进行解析以及聚合，最后将处理结果写入Elasticsearch中。

Waterdrop

[Waterdrop](https://github.com/InterestingLab/waterdrop)同样拥有着非常丰富的插件，支持从Kafka、HDFS、Hive中读取数据，进行各种各样的数据处理，并将结果写入Elasticsearch、Kudu或者Kafka中。

Prerequisites

首先我们需要安装Waterdrop，安装十分简单，无需配置系统环境变量
1. 准备Spark环境
2. 安装Waterdrop
3. 配置Waterdrop
 
 以下是简易步骤，具体安装可以参照[Quick Start](https://interestinglab.github. ... -start)
 
```yaml
cd /usr/local
wget https://archive.apache.org/dis ... 7.tgz
tar -xvf https://archive.apache.org/dis ... 7.tgz
wget https://github.com/Interesting ... 1.zip
unzip waterdrop-1.1.1.zip
cd waterdrop-1.1.1

vim config/waterdrop-env.sh

指定Spark安装路径

SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.2.0-bin-hadoop2.7}
```
 
 Waterdrop Pipeline
 
 与Logstash一样，我们仅需要编写一个Waterdrop Pipeline的配置文件即可完成数据的导入，相信了解Logstash的朋友可以很快入手Waterdrop配置。
 
 配置文件包括四个部分，分别是Spark、Input、filter和Output。
 
 Spark
 
 这一部分是Spark的相关配置，主要配置Spark执行时所需的资源大小。
  spark { spark.app.name = "Waterdrop" spark.executor.instances = 2 spark.executor.cores = 1 spark.executor.memory = "1g" } 
 
 Input
 
 这一部分定义数据源，如下是从Kafka中读取数据的配置案例，
 
  kafkaStream { topics = "waterdrop-es" consumer.bootstrap.servers = "localhost:9092" consumer.group.id = "waterdrop_es_group" consumer.rebalance.max.retries = 100 } 
 
 Filter
 
 在Filter部分，这里我们配置一系列的转化，包括正则解析将日志进行拆分、时间转换将HTTPDATE转化为Elasticsearch支持的日期格式、对Number类型的字段进行类型转换以及通过SQL进行数据聚合
```yaml
filter {

使用正则解析原始日志

最开始数据都在raw_message字段中

grok {
source_field = "raw_message"
pattern = '%{NOTSPACE:hostname}\s%{NOTSPACE:domain}\s%{IP:remote_addr}\s%{NUMBER:request_time}s\s\"%{DATA:upstream_ip}\"\s\[%{HTTPDATE:timestamp}\]\s\"%{NOTSPACE:method}\s%{DATA:url}\s%{NOTSPACE:http_ver}\"\s%{NUMBER:status}\s%{NUMBER:body_bytes_send}\s%{DATA:referer}\s%{NOTSPACE:cookie_info}\s\"%{DATA:user_agent}'
}

将"dd/MMM/yyyy:HH:mm:ss Z"格式的数据转换为

Elasticsearch中支持的格式

date {
source_field = "timestamp"
target_field = "datetime"
source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
target_time_format = "yyyy-MM-dd'T'HH:mm:ss.SSS+08:00"
}

利用SQL对数据进行聚合

sql {
table_name = "access_log"
sql = "select domain, hostname, int(status), datetime, count(*) from access_log group by domain, hostname, status, datetime"
}
}
```
 
 Output
 
 最后我们将处理好的结构化数据写入Elasticsearch。
 
 yaml output { elasticsearch { hosts = ["localhost:9200"] index = "waterdrop-${now}" es.batch.size.entries = 100000 index_time_format = "yyyy.MM.dd" } } 
 
 Running Waterdrop
 
 我们将上述四部分配置组合成为我们的配置文件config/batch.conf。
 
 vim config/batch.conf
 
```
spark {
spark.app.name = "Waterdrop"
spark.executor.instances = 2
spark.executor.cores = 1
spark.executor.memory = "1g"
}
input {
kafkaStream {
topics = "waterdrop-es"
consumer.bootstrap.servers = "localhost:9092"
consumer.group.id = "waterdrop_es_group"
consumer.rebalance.max.retries = 100
}
}
filter {

使用正则解析原始日志

最开始数据都在raw_message字段中

grok {
source_field = "raw_message"
pattern = '%{IP:hostname}\s%{NOTSPACE:domain}\s%{IP:remote_addr}\s%{NUMBER:request_time}s\s\"%{DATA:upstream_ip}\"\s\[%{HTTPDATE:timestamp}\]\s\"%{NOTSPACE:method}\s%{DATA:url}\s%{NOTSPACE:http_ver}\"\s%{NUMBER:status}\s%{NUMBER:body_bytes_send}\s%{DATA:referer}\s%{NOTSPACE:cookie_info}\s\"%{DATA:user_agent}'
}

将"dd/MMM/yyyy:HH:mm:ss Z"格式的数据转换为

Elasticsearch中支持的格式

date {
source_field = "timestamp"
target_field = "datetime"
source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
target_time_format = "yyyy-MM-dd'T'HH:mm:00.SSS+08:00"
}

利用SQL对数据进行聚合

sql {
table_name = "access_log"
sql = "select domain, hostname, status, datetime, count(*) from access_log group by domain, localhost, status, datetime"
}
}
output {
elasticsearch {
hosts = ["localhost:9200"]
index = "waterdrop-${now}"
es.batch.size.entries = 100000
index_timeformat = "yyyy.MM.dd"
}
}
```
 
 执行命令，指定配置文件，运行Waterdrop，即可将数据写入Elasticsearch。这里我们以本地模式为例。
 
 ./bin/start-waterdrop.sh --config config/batch.conf -e client -m 'local[2]'
 
 最后，写入Elasticsearch中的数据如下，再配上Kibana就可以实现Web服务的实时监控了^^.
 
  "_source": { "domain": "elasticsearch.cn", "hostname": "localhost", "status": "200", "datetime": "2018-11-26T21:54:00.000+08:00", "count": 26 } 
 
 Conclusion
 
 在这篇文章中，我们介绍了如何通过Waterdrop将Kafka中的数据写入Elasticsearch中。仅仅通过一个配置文件便可快速运行一个Spark Application，完成数据的处理、写入，无需编写任何代码，十分简单。
 
 当数据处理过程中有遇到Logstash无法支持的场景或者Logstah性能无法达到预期的情况下，都可以尝试使用Waterdrop解决问题。
 
 希望了解Waterdrop与Elasticsearch、Kafka、Hadoop结合使用的更多功能和案例，可以直接进入项目主页[https://github.com/InterestingLab/waterdrop](https://github.com/InterestingLab/waterdrop)
 
 我们近期会再发布一篇《如何用Spark和Elasticsearch做交互式数据分析》，敬请期待.
 
 Contract us
 
 欢迎联系我们交流Spark和Elasticsearch:
 
 Garyelephant：微信: garyelephant
 
 RickyHuo：微信: chodomatte1994

Day 7 - Elasticsearch中数据是如何存储的

weizijun 发表了文章 • 7 个评论 • 72734 次浏览 • 2018-12-07 13:55 • 来自相关话题

前言

很多使用Elasticsearch的同学会关心数据存储在ES中的存储容量，会有这样的疑问：xxTB的数据入到ES会使用多少存储空间。这个问题其实很难直接回答的，只有数据写入ES后，才能观察到实际的存储空间。比如同样是1TB的数据，写入ES的存储空间可能差距会非常大，可能小到只有300~400GB，也可能多到6-7TB，为什么会造成这么大的差距呢？究其原因，我们来探究下Elasticsearch中的数据是如何存储。文章中我以Elasticsearch 2.3版本为示例，对应的lucene版本是5.5，Elasticsearch现在已经来到了6.5版本，数字类型、列存等存储结构有些变化，但基本的概念变化不多，文章中的内容依然适用。

Elasticsearch索引结构

Elasticsearch对外提供的是index的概念，可以类比为DB，用户查询是在index上完成的，每个index由若干个shard组成，以此来达到分布式可扩展的能力。比如下图是一个由10个shard组成的index。

shard是Elasticsearch数据存储的最小单位，index的存储容量为所有shard的存储容量之和。Elasticsearch集群的存储容量则为所有index存储容量之和。

一个shard就对应了一个lucene的library。对于一个shard，Elasticsearch增加了translog的功能，类似于HBase WAL，是数据写入过程中的中间数据，其余的数据都在lucene库中管理的。

所以Elasticsearch索引使用的存储内容主要取决于lucene中的数据存储。

lucene数据存储

下面我们主要看下lucene的文件内容，在了解lucene文件内容前，大家先了解些lucene的基本概念。

lucene基本概念

segment : lucene内部的数据是由一个个segment组成的，写入lucene的数据并不直接落盘，而是先写在内存中，经过了refresh间隔，lucene才将该时间段写入的全部数据refresh成一个segment，segment多了之后会进行merge成更大的segment。lucene查询时会遍历每个segment完成。由于lucene* 写入的数据是在内存中完成，所以写入效率非常高。但是也存在丢失数据的风险，所以Elasticsearch基于此现象实现了translog，只有在segment数据落盘后，Elasticsearch才会删除对应的translog。
doc : doc表示lucene中的一条记录
field ：field表示记录中的字段概念，一个doc由若干个field组成。
term ：term是lucene中索引的最小单位，某个field对应的内容如果是全文检索类型，会将内容进行分词，分词的结果就是由term组成的。如果是不分词的字段，那么该字段的内容就是一个term。
倒排索引（inverted index）: lucene索引的通用叫法，即实现了term到doc list的映射。
正排数据：搜索引擎的通用叫法，即原始数据，可以理解为一个doc list。
docvalues :Elasticsearch中的列式存储的名称，Elasticsearch除了存储原始存储、倒排索引，还存储了一份docvalues，用作分析和排序。

lucene文件内容

lucene包的文件是由很多segment文件组成的，segments_xxx文件记录了lucene包下面的segment文件数量。每个segment会包含如下的文件。

| Name | Extension | Brief Description |
| ------ | ------ | ------ |
|Segment Info|.si|segment的元数据文件|
|Compound File|.cfs, .cfe|一个segment包含了如下表的各个文件，为减少打开文件的数量，在segment小的时候，segment的所有文件内容都保存在cfs文件中，cfe文件保存了lucene各文件在cfs文件的位置信息|
|Fields|.fnm|保存了fields的相关信息|
|Field Index|.fdx|正排存储文件的元数据信息|
|Field Data|.fdt|存储了正排存储数据，写入的原文存储在这|
|Term Dictionary|.tim|倒排索引的元数据信息|
|Term Index|.tip|倒排索引文件，存储了所有的倒排索引数据|
|Frequencies|.doc|保存了每个term的doc id列表和term在doc中的词频|
|Positions|.pos|Stores position information about where a term occurs in the index
全文索引的字段，会有该文件，保存了term在doc中的位置|
|Payloads|.pay|Stores additional per-position metadata information such as character offsets and user payloads
全文索引的字段，使用了一些像payloads的高级特性会有该文件，保存了term在doc中的一些高级特性|
|Norms|.nvd, .nvm|文件保存索引字段加权数据|
|Per-Document Values|.dvd, .dvm|lucene的docvalues文件，即数据的列式存储，用作聚合和排序|
|Term Vector Data|.tvx, .tvd, .tvf|Stores offset into the document data file
保存索引字段的矢量信息，用在对term进行高亮，计算文本相关性中使用|
|Live Documents|.liv|记录了segment中删除的doc|

测试数据示例

下面我们以真实的数据作为示例，看看lucene中各类型数据的容量占比。

写100w数据，有一个uuid字段，写入的是长度为36位的uuid，字符串总为3600w字节，约为35M。

数据使用一个shard，不带副本，使用默认的压缩算法，写入完成后merge成一个segment方便观察。

使用线上默认的配置，uuid存为不分词的字符串类型。创建如下索引：

 PUT test_field { "settings": { "index": { "number_of_shards": "1", "number_of_replicas": "0", "refresh_interval": "30s" } }, "mappings": { "type": { "_all": { "enabled": false }, "properties": { "uuid": { "type": "string", "index": "not_analyzed" } } } } } 

首先写入100w不同的uuid，使用磁盘容量细节如下：

 health status index pri rep docs.count docs.deleted store.size pri.store.size green open test_field 1 0 1000000 0 122.7mb 122.7mb -rw-r--r-- 1 weizijun staff 41M Aug 19 21:23 _8.fdt -rw-r--r-- 1 weizijun staff 17K Aug 19 21:23 _8.fdx -rw-r--r-- 1 weizijun staff 688B Aug 19 21:23 _8.fnm -rw-r--r-- 1 weizijun staff 494B Aug 19 21:23 _8.si -rw-r--r-- 1 weizijun staff 265K Aug 19 21:23 _8_Lucene50_0.doc -rw-r--r-- 1 weizijun staff 44M Aug 19 21:23 _8_Lucene50_0.tim -rw-r--r-- 1 weizijun staff 340K Aug 19 21:23 _8_Lucene50_0.tip -rw-r--r-- 1 weizijun staff 37M Aug 19 21:23 _8_Lucene54_0.dvd -rw-r--r-- 1 weizijun staff 254B Aug 19 21:23 _8_Lucene54_0.dvm -rw-r--r-- 1 weizijun staff 195B Aug 19 21:23 segments_2 -rw-r--r-- 1 weizijun staff 0B Aug 19 21:20 write.lock 
可以看到正排数据、倒排索引数据，列存数据容量占比几乎相同，正排数据和倒排数据还会存储Elasticsearch的唯一id字段，所以容量会比列存多一些。

35M的uuid存入Elasticsearch后，数据膨胀了3倍，达到了122.7mb。Elasticsearch竟然这么消耗资源，不要着急下结论，接下来看另一个测试结果。

我们写入100w一样的uuid，然后看看Elasticsearch使用的容量。

 health status index pri rep docs.count docs.deleted store.size pri.store.size green open test_field 1 0 1000000 0 13.2mb 13.2mb -rw-r--r-- 1 weizijun staff 5.5M Aug 19 21:29 _6.fdt -rw-r--r-- 1 weizijun staff 15K Aug 19 21:29 _6.fdx -rw-r--r-- 1 weizijun staff 688B Aug 19 21:29 _6.fnm -rw-r--r-- 1 weizijun staff 494B Aug 19 21:29 _6.si -rw-r--r-- 1 weizijun staff 309K Aug 19 21:29 _6_Lucene50_0.doc -rw-r--r-- 1 weizijun staff 7.0M Aug 19 21:29 _6_Lucene50_0.tim -rw-r--r-- 1 weizijun staff 195K Aug 19 21:29 _6_Lucene50_0.tip -rw-r--r-- 1 weizijun staff 244K Aug 19 21:29 _6_Lucene54_0.dvd -rw-r--r-- 1 weizijun staff 252B Aug 19 21:29 _6_Lucene54_0.dvm -rw-r--r-- 1 weizijun staff 195B Aug 19 21:29 segments_2 -rw-r--r-- 1 weizijun staff 0B Aug 19 21:26 write.lock 

这回35M的数据Elasticsearch容量只有13.2mb，其中还有主要的占比还是Elasticsearch的唯一id，100w的uuid几乎不占存储容积。

所以在Elasticsearch中建立索引的字段如果基数越大(count distinct)，越占用磁盘空间。

我们再看看存100w个不一样的整型会是如何。

 health status index pri rep docs.count docs.deleted store.size pri.store.size green open test_field 1 0 1000000 0 13.6mb 13.6mb -rw-r--r-- 1 weizijun staff 6.1M Aug 28 10:19 _42.fdt -rw-r--r-- 1 weizijun staff 22K Aug 28 10:19 _42.fdx -rw-r--r-- 1 weizijun staff 688B Aug 28 10:19 _42.fnm -rw-r--r-- 1 weizijun staff 503B Aug 28 10:19 _42.si -rw-r--r-- 1 weizijun staff 2.8M Aug 28 10:19 _42_Lucene50_0.doc -rw-r--r-- 1 weizijun staff 2.2M Aug 28 10:19 _42_Lucene50_0.tim -rw-r--r-- 1 weizijun staff 83K Aug 28 10:19 _42_Lucene50_0.tip -rw-r--r-- 1 weizijun staff 2.5M Aug 28 10:19 _42_Lucene54_0.dvd -rw-r--r-- 1 weizijun staff 228B Aug 28 10:19 _42_Lucene54_0.dvm -rw-r--r-- 1 weizijun staff 196B Aug 28 10:19 segments_2 -rw-r--r-- 1 weizijun staff 0B Aug 28 10:16 write.lock 

从结果可以看到，100w整型数据，Elasticsearch的存储开销为13.6mb。如果以int型计算100w数据的长度的话，为400w字节，大概是3.8mb数据。忽略Elasticsearch唯一id字段的影响，Elasticsearch实际存储容量跟整型数据长度差不多。

我们再看一下开启最佳压缩参数对存储空间的影响：

 health status index pri rep docs.count docs.deleted store.size pri.store.size green open test_field 1 0 1000000 0 107.2mb 107.2mb -rw-r--r-- 1 weizijun staff 25M Aug 20 12:30 _5.fdt -rw-r--r-- 1 weizijun staff 6.0K Aug 20 12:30 _5.fdx -rw-r--r-- 1 weizijun staff 688B Aug 20 12:31 _5.fnm -rw-r--r-- 1 weizijun staff 500B Aug 20 12:31 _5.si -rw-r--r-- 1 weizijun staff 265K Aug 20 12:31 _5_Lucene50_0.doc -rw-r--r-- 1 weizijun staff 44M Aug 20 12:31 _5_Lucene50_0.tim -rw-r--r-- 1 weizijun staff 322K Aug 20 12:31 _5_Lucene50_0.tip -rw-r--r-- 1 weizijun staff 37M Aug 20 12:31 _5_Lucene54_0.dvd -rw-r--r-- 1 weizijun staff 254B Aug 20 12:31 _5_Lucene54_0.dvm -rw-r--r-- 1 weizijun staff 224B Aug 20 12:31 segments_4 -rw-r--r-- 1 weizijun staff 0B Aug 20 12:00 write.lock 

结果中可以发现，只有正排数据会启动压缩，压缩能力确实强劲，不考虑唯一id字段，存储容量大概压缩到接近50%。

我们还做了一些实验，Elasticsearch默认是开启_all参数的，_all可以让用户传入的整体json数据作为全文检索的字段，可以更方便的检索，但在现实场景中已经使用的不多，相反会增加很多存储容量的开销，可以看下开启_all的磁盘空间使用情况：

 health status index pri rep docs.count docs.deleted store.size pri.store.size green open test_field 1 0 1000000 0 162.4mb 162.4mb -rw-r--r-- 1 weizijun staff 41M Aug 18 22:59 _20.fdt -rw-r--r-- 1 weizijun staff 18K Aug 18 22:59 _20.fdx -rw-r--r-- 1 weizijun staff 777B Aug 18 22:59 _20.fnm -rw-r--r-- 1 weizijun staff 59B Aug 18 22:59 _20.nvd -rw-r--r-- 1 weizijun staff 78B Aug 18 22:59 _20.nvm -rw-r--r-- 1 weizijun staff 539B Aug 18 22:59 _20.si -rw-r--r-- 1 weizijun staff 7.2M Aug 18 22:59 _20_Lucene50_0.doc -rw-r--r-- 1 weizijun staff 4.2M Aug 18 22:59 _20_Lucene50_0.pos -rw-r--r-- 1 weizijun staff 73M Aug 18 22:59 _20_Lucene50_0.tim -rw-r--r-- 1 weizijun staff 832K Aug 18 22:59 _20_Lucene50_0.tip -rw-r--r-- 1 weizijun staff 37M Aug 18 22:59 _20_Lucene54_0.dvd -rw-r--r-- 1 weizijun staff 254B Aug 18 22:59 _20_Lucene54_0.dvm -rw-r--r-- 1 weizijun staff 196B Aug 18 22:59 segments_2 -rw-r--r-- 1 weizijun staff 0B Aug 18 22:53 write.lock 

开启_all比不开启多了40mb的存储空间，多的数据都在倒排索引上，大约会增加30%多的存储开销。所以线上都直接禁用。

然后我还做了其他几个尝试，为了验证存储容量是否和数据量成正比，写入1000w数据的uuid，发现存储容量基本为100w数据的10倍。我还验证了数据长度是否和数据量成正比，发现把uuid增长2倍、4倍，存储容量也响应的增加了2倍和4倍。在此就不一一列出数据了。

lucene各文件具体内容和实现

lucene数据元信息文件

文件名为：segments_xxx

该文件为lucene数据文件的元信息文件，记录所有segment的元数据信息。

该文件主要记录了目前有多少segment，每个segment有一些基本信息，更新这些信息定位到每个segment的元信息文件。

lucene元信息文件还支持记录userData，Elasticsearch可以在此记录translog的一些相关信息。

文件示例

具体实现类

```
public final class SegmentInfos implements Cloneable, Iterable {
// generation是segment的版本的概念，从文件名中提取出来，实例中为：2t/101
private long generation; // generation of the "segments_N" for the next commit

private long lastGeneration; // generation of the "segments_N" file we last successfully read
// or wrote; this is normally the same as generation except if
// there was an IOException that had interrupted a commit

/ Id for this commit; only written starting with Lucene 5.0 */
private byte[] id;

/* Which Lucene version wrote this commit, or null if this commit is pre-5.3. /
private Version luceneVersion;

/ Counts how often the index has been changed. */
public long version;

/ Used to name new segments. */
// TODO: should this be a long ...?
public int counter;

/* Version of the oldest segment in the index, or null if there are no segments. /
private Version minSegmentLuceneVersion;

private List segments = new ArrayList<>();

/ Opaque Map<String, String> that user can specify during IndexWriter.commit */
public Map<String,String> userData = Collections.emptyMap();
}

/** Embeds a [read-only] SegmentInfo and adds per-commit
- fields.
 *
- @lucene.experimental */
 public class SegmentCommitInfo {
 
 /* The {@link SegmentInfo} that we wrap. /
 public final SegmentInfo info;
 
 // How many deleted docs in the segment:
 private int delCount;
 
 // Generation number of the live docs file (-1 if there
 // are no deletes yet):
 private long delGen;
 
 // Normally 1+delGen, unless an exception was hit on last
 // attempt to write:
 private long nextWriteDelGen;
 
 // Generation number of the FieldInfos (-1 if there are no updates)
 private long fieldInfosGen;
 
 // Normally 1+fieldInfosGen, unless an exception was hit on last attempt to
 // write
 private long nextWriteFieldInfosGen; //fieldInfosGen == -1 ? 1 : fieldInfosGen + 1;
 
 // Generation number of the DocValues (-1 if there are no updates)
 private long docValuesGen;
 
 // Normally 1+dvGen, unless an exception was hit on last attempt to
 // write
 private long nextWriteDocValuesGen; //docValuesGen == -1 ? 1 : docValuesGen + 1;
 
 // TODO should we add .files() to FieldInfosFormat, like we have on
 // LiveDocsFormat?
 // track the fieldInfos update files
 private final Set fieldInfosFiles = new HashSet<>();
 
 // Track the per-field DocValues update files
 private final Map<Integer,Set> dvUpdatesFiles = new HashMap<>();
 
 // Track the per-generation updates files
 @Deprecated
 private final Map<Long,Set> genUpdatesFiles = new HashMap<>();
 
 private volatile long sizeInBytes = -1;
 }
 
```

segment的元信息文件

文件后缀：.si

每个segment都有一个.si文件，记录了该segment的元信息。

segment元信息文件中记录了segment的文档数量，segment对应的文件列表等信息。

文件示例

具体实现类

```
 /**
- Information about a segment such as its name, directory, and files related
- to the segment.
 *
- @lucene.experimental
 */
 public final class SegmentInfo {
 
 // _bl
 public final String name;
 
 /* Where this segment resides. /
 public final Directory dir;
 
 /* Id that uniquely identifies this segment. /
 private final byte[] id;
 
 private Codec codec;
 
 // Tracks the Lucene version this segment was created with, since 3.1. Null
 // indicates an older than 3.0 index, and it's used to detect a too old index.
 // The format expected is "x.y" - "2.x" for pre-3.0 indexes (or null), and
 // specific versions afterwards ("3.0.0", "3.1.0" etc.).
 // see o.a.l.util.Version.
 private Version version;
 
 private int maxDoc; // number of docs in seg
 
 private boolean isCompoundFile;
 
 private Map<String,String> diagnostics;
 
 private Set setFiles;
 
 private final Map<String,String> attributes;
 }
```

fields信息文件

文件后缀：.fnm

该文件存储了fields的基本信息。

fields信息中包括field的数量，field的类型，以及IndexOpetions，包括是否存储、是否索引，是否分词，是否需要列存等等。

文件示例

具体实现类

```
 /**
- Access to the Field Info file that describes document fields and whether or
- not they are indexed. Each segment has a separate Field Info file. Objects
- of this class are thread-safe for multiple readers, but only one thread can
- be adding documents at a time, with no other reader or writer threads
- accessing this object.
 /
 public final class FieldInfo {
 / Field's name */
 public final String name;
 
 /* Internal field number /
 //field在内部的编号
 public final int number;
 
 //field docvalues的类型
 private DocValuesType docValuesType = DocValuesType.NONE;
 
 // True if any document indexed term vectors
 private boolean storeTermVector;
 
 private boolean omitNorms; // omit norms associated with indexed fields
 
 //index的配置项
 private IndexOptions indexOptions = IndexOptions.NONE;
 
 private boolean storePayloads; // whether this field stores payloads together with term positions
 
 private final Map<String,String> attributes;
 
 // docvalues的generation
 private long dvGen;
 }
```

数据存储文件

文件后缀：.fdx, .fdt

索引文件为.fdx，数据文件为.fdt，数据存储文件功能为根据自动的文档id，得到文档的内容，搜索引擎的术语习惯称之为正排数据，即doc_id -> content，es的_source数据就存在这

索引文件记录了快速定位文档数据的索引信息，数据文件记录了所有文档id的具体内容。

文件示例

具体实现类

```
 /**
- Random-access reader for {@link CompressingStoredFieldsIndexWriter}.
- @lucene.internal
 */
 public final class CompressingStoredFieldsIndexReader implements Cloneable, Accountable {
 private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(CompressingStoredFieldsIndexReader.class);
 
 final int maxDoc;
 
 //docid索引，快速定位某个docid的数组坐标
 final int[] docBases;
 
 //快速定位某个docid所在的文件offset的startPointer
 final long[] startPointers;
 
 //平均一个chunk的文档数
 final int[] avgChunkDocs;
 
 //平均一个chunk的size
 final long[] avgChunkSizes;
 
 final PackedInts.Reader[] docBasesDeltas; // delta from the avg
 
 final PackedInts.Reader[] startPointersDeltas; // delta from the avg
 }
 
 /**
- {@link StoredFieldsReader} impl for {@link CompressingStoredFieldsFormat}.
- @lucene.experimental
 */
 public final class CompressingStoredFieldsReader extends StoredFieldsReader {
 
 //从fdt正排索引文件中获得
 private final int version;
 
 // field的基本信息
 private final FieldInfos fieldInfos;
 
 //fdt正排索引文件reader
 private final CompressingStoredFieldsIndexReader indexReader;
 
 //从fdt正排索引文件中获得，用于指向fdx数据文件的末端，指向numChunks地址4
 private final long maxPointer;
 
 //fdx正排数据文件句柄
 private final IndexInput fieldsStream;
 
 //块大小
 private final int chunkSize;
 
 private final int packedIntsVersion;
 
 //压缩类型
 private final CompressionMode compressionMode;
 
 //解压缩处理对象
 private final Decompressor decompressor;
 
 //文档数量，从segment元数据中获得
 private final int numDocs;
 
 //是否正在merge，默认为false
 private final boolean merging;
 
 //初始化时new了一个BlockState，BlockState记录下当前正排文件读取的状态信息
 private final BlockState state;
 //chunk的数量
 private final long numChunks; // number of compressed blocks written
 
 //dirty chunk的数量
 private final long numDirtyChunks; // number of incomplete compressed blocks written
 
 //是否close，默认为false
 private boolean closed;
 }
```

倒排索引文件

索引后缀：.tip,.tim

倒排索引也包含索引文件和数据文件，.tip为索引文件，.tim为数据文件，索引文件包含了每个字段的索引元信息，数据文件有具体的索引内容。

5.5.0版本的倒排索引实现为FST tree，FST tree的最大优势就是内存空间占用非常低，具体可以参看下这篇文章：[http://www.cnblogs.com/bonelee/p/6226185.html ](http://www.cnblogs.com/bonelee/p/6226185.html )

[http://examples.mikemccandless ... %2Bit](http://examples.mikemccandless ... d%2Bit) 为FST图实例，可以根据输入的数据构造出FST图

 输入到 FST 中的数据为: String inputValues[] = {"mop","moth","pop","star","stop","top"}; long outputValues[] = {0,1,2,3,4,5}; 
生成的 FST 图为:

文件示例

具体实现类

```
 public final class BlockTreeTermsReader extends FieldsProducer {
 // Open input to the main terms dict file (_X.tib)
 final IndexInput termsIn;
 // Reads the terms dict entries, to gather state to
 // produce DocsEnum on demand
 final PostingsReaderBase postingsReader;
 private final TreeMap<String,FieldReader> fields = new TreeMap<>();
 
 / File offset where the directory starts in the terms file. */
 /索引数据文件tim的数据的尾部的元数据的地址
 private long dirOffset;
 /* File offset where the directory starts in the index file. /
 
 //索引文件tip的数据的尾部的元数据的地址
 private long indexDirOffset;
 
 //semgent的名称
 final String segment;
 
 //版本号
 final int version;
 
 //5.3.x index, we record up front if we may have written any auto-prefix terms，示例中记录的是false
 final boolean anyAutoPrefixTerms;
 }
 
 /
- BlockTree's implementation of {@link Terms}.
- @lucene.internal
 */
 public final class FieldReader extends Terms implements Accountable {
 
 //term的数量
 final long numTerms;
 
 //field信息
 final FieldInfo fieldInfo;
 
 final long sumTotalTermFreq;
 
 //总的文档频率
 final long sumDocFreq;
 
 //文档数量
 final int docCount;
 
 //字段在索引文件tip中的起始位置
 final long indexStartFP;
 
 final long rootBlockFP;
 
 final BytesRef rootCode;
 
 final BytesRef minTerm;
 
 final BytesRef maxTerm;
 
 //longs：metadata buffer, holding monotonic values
 final int longsSize;
 
 final BlockTreeTermsReader parent;
 
 final FST index;
 }
```

倒排链文件

文件后缀：.doc, .pos, .pay

.doc保存了每个term的doc id列表和term在doc中的词频

全文索引的字段，会有.pos文件，保存了term在doc中的位置

全文索引的字段，使用了一些像payloads的高级特性才会有.pay文件，保存了term在doc中的一些高级特性

文件示例

具体实现类

```
 /**
- Concrete class that reads docId(maybe frq,pos,offset,payloads) list
- with postings format.
 *
- @lucene.experimental
 */
 public final class Lucene50PostingsReader extends PostingsReaderBase {
 private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(Lucene50PostingsReader.class);
 private final IndexInput docIn;
 private final IndexInput posIn;
 private final IndexInput payIn;
 final ForUtil forUtil;
 private int version;
 
 //不分词的字段使用的是该对象，基于skiplist实现了倒排链
 final class BlockDocsEnum extends PostingsEnum {
 }
 
 //全文检索字段使用的是该对象
 final class BlockPostingsEnum extends PostingsEnum {
 }
 
 //包含高级特性的字段使用的是该对象
 final class EverythingEnum extends PostingsEnum {
 }
 }
```

列存文件（docvalues）

文件后缀：.dvm, .dvd

索引文件为.dvm，数据文件为.dvd。

lucene实现的docvalues有如下类型：
1、NONE 不开启docvalue时的状态
2、NUMERIC 单个数值类型的docvalue主要包括（int，long，float，double）
3、BINARY 二进制类型值对应不同的codes最大值可能超过32766字节，
4、SORTED 有序增量字节存储，仅仅存储不同部分的值和偏移量指针，值必须小于等于32766字节
5、SORTED_NUMERIC 存储数值类型的有序数组列表
6、SORTED_SET 可以存储多值域的docvalue值，但返回时，仅仅只能返回多值域的第一个docvalue
7、对应not_anaylized的string字段，使用的是SORTED_SET类型，number的类型是SORTED_NUMERIC类型

其中SORTED_SET 的 SORTED_SINGLE_VALUED类型包括了两类数据： binary + numeric， binary是按ord排序的term的列表，numeric是doc到ord的映射。

文件示例

具体实现类

 /** reader for {@link Lucene54DocValuesFormat} */ final class Lucene54DocValuesProducer extends DocValuesProducer implements Closeable { //number类型的field的列存列表 private final Map<String,NumericEntry> numerics = new HashMap<>(); //字符串类型的field的列存列表 private final Map<String,BinaryEntry> binaries = new HashMap<>(); //有序字符串类型的field的列存列表 private final Map<String,SortedSetEntry> sortedSets = new HashMap<>(); //有序number类型的field的列存列表 private final Map<String,SortedSetEntry> sortedNumerics = new HashMap<>(); //字符串类型的field的ords列表 private final Map<String,NumericEntry> ords = new HashMap<>(); //docId -> address -> ord 中field的ords列表 private final Map<String,NumericEntry> ordIndexes = new HashMap<>(); //field的数量 private final int numFields; //内存使用量 private final AtomicLong ramBytesUsed; //数据源的文件句柄 private final IndexInput data; //文档数 private final int maxDoc; // memory-resident structures private final Map<String,MonotonicBlockPackedReader> addressInstances = new HashMap<>(); private final Map<String,ReverseTermsIndex> reverseIndexInstances = new HashMap<>(); private final Map<String,DirectMonotonicReader.Meta> directAddressesMeta = new HashMap<>(); //是否正在merge private final boolean merging; } /** metadata entry for a numeric docvalues field */ static class NumericEntry { private NumericEntry() {} /** offset to the bitset representing docsWithField, or -1 if no documents have missing values */ long missingOffset; /** offset to the actual numeric values */ //field的在数据文件中的起始地址 public long offset; /** end offset to the actual numeric values */ //field的在数据文件中的结尾地址 public long endOffset; /** bits per value used to pack the numeric values */ public int bitsPerValue; //format类型 int format; /** count of values written */ public long count; /** monotonic meta */ public DirectMonotonicReader.Meta monotonicMeta; //最小的value long minValue; //Compressed by computing the GCD long gcd; //Compressed by giving IDs to unique values. long table[]; /** for sparse compression */ long numDocsWithValue; NumericEntry nonMissingValues; NumberType numberType; } /** metadata entry for a binary docvalues field */ static class BinaryEntry { private BinaryEntry() {} /** offset to the bitset representing docsWithField, or -1 if no documents have missing values */ long missingOffset; /** offset to the actual binary values */ //field的在数据文件中的起始地址 long offset; int format; /** count of values written */ public long count; //最短字符串的长度 int minLength; //最长字符串的长度 int maxLength; /** offset to the addressing data that maps a value to its slice of the byte[] */ public long addressesOffset, addressesEndOffset; /** meta data for addresses */ public DirectMonotonicReader.Meta addressesMeta; /** offset to the reverse index */ public long reverseIndexOffset; /** packed ints version used to encode addressing information */ public int packedIntsVersion; /** packed ints blocksize */ public int blockSize; } 

参考资料

[lucene source code](https://github.com/apache/luce ... /5.5.0)

[lucene document](https://lucene.apache.org/core/5_5_0/)

[lucene字典实现原理——FST](http://www.cnblogs.com/bonelee/p/6226185.html )

Day 6 - Logstash Pipeline-to-Pipeline 尝鲜

rockybean 发表了文章 • 3 个评论 • 10214 次浏览 • 2018-12-06 23:40 • 来自相关话题

Logstash 在 6.0 推出了 multiple pipeline 的解决方案，即在一个 logstash 实例中可以同时进行多个独立数据流程的处理工作，如下图所示。

![](https://ws1.sinaimg.cn/large/6 ... xr.jpg)

而在这之前用户只能通过在单机运行多个 logstash 实例或者在配置文件中增加大量 if-else 条件判断语句来解决。要使用 multiple pipeline 也很简单，只需要将不同的 pipeline 在 config/pipeline.yml中定义好即可，如下所示：

```yaml

pipeline.id: apache
pipeline.batch.size: 125
queue.type: persisted
path.config: "/path/to/config/apache.cfg"
pipeline.id: nginx
path.config: "/path/to/config/nginx.cfg"
`` 其中apache和nginx作为独立的 pipeline 执行，而且配置也可以独立设置，互不干扰。pipeline.yml的引入极大地简化了 logstash 的配置管理工作，使得新手也可以很快完成复杂的 ETL 配置。 在 6.3 版本中，Logstash 又增加了Pipeline-to-Pipeline的管道机制（beta），即管道和管道之间可以连接在一起组成一个完成的数据处理流。熟悉 linux 的管道命令|`的同学应该可以很快明白这种模式的好处。这无疑使得 Logstash 的配置会更加灵活，今天我们就来了解下这种灵活自由的配置方式。

1. 上手

废话少说，快速上手。修改 config/pipeline.yml文件如下：

```yaml
- pipeline.id: upstream
 config.string: input { stdin {} } output { pipeline { send_to => [test_output] } }
- pipeline.id: downstream
 config.string: input { pipeline { address => test_output } } output{ stdout{}}
 `` 然后运行 logstash，其中-r` 表示配置文件有改动时自动重新加载，方便我们调试。
 
 bin/logstash -r
 
 在终端随意输入字符(比如aaa)后回车，会看到屏幕输出了类似下面的内容，代表运行成功了。
 
 json { "@timestamp" => 2018-12-06T14:43:50.310Z, "@version" => "1", "message" => "aaa", "host" => "rockybean-MacBook-Pro.local" } 
 
 我们再回头看下这个配置，upstreamoutput 使用了名为 pipeline 的 plugin，然后 send_to的输出对象test_output是在 downstream的 input pipeline plugin 中定义的。通过这个唯一的address(虚拟地址)就能够把不同的 pipeline 连接在一起组成一个更长的pipeline来处理数据。类似下图所示：
 
 ![](https://ws1.sinaimg.cn/large/6 ... y9.jpg)
 
 当数据由 upstream传递给 downstream时会进行一个复制操作，这也意味着在这两个 pipeline 中的数据是完全独立的，互不影响。有一点要注意的是：数据的复制会增加额外的性能开销，比如会加大 JVM Heap 的使用。
 
 2. 使用场景
 
 使用方法是不是很简单，接下来我们来看下官方为我们开的几个脑洞。
 
 2.1 Distributor Pattern 分发者模式
 
 该模式执行效果类似下图所示：
 
 ![](https://ws1.sinaimg.cn/large/6 ... l1.jpg)
 
 在一个 pipeline 处理输入，然后根据不同的数据类型再分发到对应的 Pipeline 去处理。这种模式的好处在于统一输入端口，隔离不同类型的处理配置文件，减少由于配置文件混合在一起带来的维护成本。大家可以想一想如果不用这种Pipeline-to-Pipeline的方式，我们如果轻松做到一个端口处理多个来源的数据呢？
 
 这种模式的参考配置如下所示：
 
```yaml

config/pipelines.yml
pipeline.id: beats-server
config.string: |
input { beats { port => 5044 } }
output {
if [type] == apache {
pipeline { send_to => weblogs }
} else if [type] == system {
pipeline { send_to => syslog }
} else {
pipeline { send_to => fallback }
}
}
pipeline.id: weblog-processing
config.string: |
input { pipeline { address => weblogs } }
filter {

Weblog filter statements here...

}
output {
elasticsearch { hosts => [es_cluster_a_host] }
}
pipeline.id: syslog-processing
config.string: |
input { pipeline { address => syslog } }
filter {

Syslog filter statements here...

}
output {
elasticsearch { hosts => [es_cluster_b_host] }
}
pipeline.id: fallback-processing
config.string: |
input { pipeline { address => fallback } }
output { elasticsearch { hosts => [es_cluster_b_host] } }
```

2.2 Output Isolator Pattern 输出隔离模式

虽然 Logstash 的一个 pipeline 可以配置多个 output，但是这多个 output 会相依为命，一旦某一个 output 出问题，会导致另一个 output 也无法接收新数据。而通过这种模式可以完美解决这个问题。其运行方式如下图所示：

![](https://ws1.sinaimg.cn/large/6 ... 8p.jpg)

通过输出到两个独立的 pipeline，解除相互之间的影响，比如 http service 出问题的时候，es 依然可以正常接收数据，而且两个 pipeline 可以配置独立的队列来保障数据的完备性，其配置如下所示：

```yaml

config/pipelines.yml
pipeline.id: intake
queue.type: persisted
config.string: |
input { beats { port => 5044 } }
output { pipeline { send_to => [es, http] } }
pipeline.id: buffered-es
queue.type: persisted
config.string: |
input { pipeline { address => es } }
output { elasticsearch { } }
pipeline.id: buffered-http
queue.type: persisted
config.string: |
input { pipeline { address => http } }
output { http { } }
```

2.3 Forked Path Pattern 克隆路径模式

这个模式类似 Output Isolator Pattern，只是在不同的 output pipeline 中可以配置不同的 filter 来完成各自输出的数据处理需求，这里就不展开讲了，可以参考如下的配置，其中不同 output pipeline 的 filter 是不同的，比如 partner 这个 pipeline 去掉了一些敏感数据：

```

config/pipelines.yml
pipeline.id: intake
queue.type: persisted
config.string: |
input { beats { port => 5044 } }
output { pipeline { send_to => ["internal-es", "partner-s3"] } }
pipeline.id: buffered-es
queue.type: persisted
config.string: |
input { pipeline { address => "internal-es" } }

Index the full event

output { elasticsearch { } }
pipeline.id: partner
queue.type: persisted
config.string: |
input { pipeline { address => "partner-s3" } }
filter {

Remove the sensitive data

mutate { remove_field => 'sensitive-data' }
}
output { s3 { } } # Output to partner's bucket
```

2.4 Collector Pattern 收集者模式

从名字可以看出，该模式是将所有 Pipeline 汇集于一处的处理模式，如下图所示：

![](https://ws1.sinaimg.cn/large/6 ... 1y.jpg)

其配置参考如下：

```

config/pipelines.yml
pipeline.id: beats
config.string: |
input { beats { port => 5044 } }
output { pipeline { send_to => [commonOut] } }
pipeline.id: kafka
config.string: |
input { kafka { ... } }
output { pipeline { send_to => [commonOut] } }
pipeline.id: partner

This common pipeline enforces the same logic whether data comes from Kafka or Beats

config.string: |
input { pipeline { address => commonOut } }
filter {

Always remove sensitive data from all input sources

mutate { remove_field => 'sensitive-data' }
}
output { elasticsearch { } }
```

3. 总结

本文简单给大家讲解了 Pipeline-to-Pipeline的使用方法及官方推荐的几种模式，希望可以给大家有所帮助。另外这个机制目前还处于 Beta 阶段，尝鲜需谨慎！

Day 5 - Elasticsearch 存储设备全解析

cyberdak 发表了文章 • 0 个评论 • 5409 次浏览 • 2018-12-05 09:57 • 来自相关话题

day5 - es存储设备全解析

Elastic Search 作为一个分布式系统，它的最小单元(shard)实现基于 lucene , lucene是一个io密集cpu密集的系统。cpu密集可以通过使用更多核，更快的cpu以及优化算法来解决。而io密集部分需要搭配高性能的存储设备以及存储策略来解决。

传统的服务器硬盘分为SATA,SAS硬盘以及现在最高性能的SSD硬盘，其中SSD硬盘又分为 SATA SSD,PCI-E SSD ,M.2 SSD（性能依次提升）。

两者的区别在于 SATA 最高可以提供 7200转的。著名的HADOOP集群中，一半都会选择企业级SATA盘来降低存储成本。而SATA盘容易损坏以及恢复速度的问题，则交给10g高速网卡以及三副本策略来解决。

如果是了解数据库领域的同学就会知道，MySQL 之类的数据库严重推荐使用SSD来做存储。TiDB这种新时代的分布式数据库甚至在安装过程中会见存储是否是高性能设备，当时低速设备时，安装将失败。

如何查看io压力

iostat -x 1 100

可以根据 iowait , ioutil 等值来综合判断. 当iowait长期接近100%基本代表io系统出现瓶颈了。这时候可以用iotop命令来诊断出具体是什么进程在消耗io资源。

如何测试硬盘性能

通过 fio 测试顺序读/写，随机读/写性能。

顺序读
fio -name iops -rw=read -bs=4k -runtime=60 -iodepth 32 -filename /dev/sda -ioengine libaio -direct=1
随机读
fio -name iops -rw=randread -bs=4k -runtime=60 -iodepth 32 -filename /dev/sda -ioengine libaio -direct=1
顺序写
fio -name iops -rw=write -bs=4k -runtime=60 -iodepth 32 -filename /dev/sda -ioengine libaio -direct=1
随机写
fio -name iops -rw=randwrite -bs=4k -runtime=60 -iodepth 32 -filename /dev/sda -ioengine libaio -direct=1

更具体的测试可以参考[磁盘性能指标--IOPS、吞吐量及测试](http://blog.51cto.com/wushank/1708168)

RAID

RAID 0

将数据分布在N块盘中，速度最快，可以享受磁盘的并行读取和写入；安全性最低，一块盘损坏，将导致所有数据丢失。

![raid0.png](https://i.loli.net/2018/12/04/5c0673f515db4.png)

RAID 1

将数据同时保存在N块盘中，写入速度最慢（需要同时写多块盘）。安全性最高。

![raid1.png](https://i.loli.net/2018/12/04/5c0673f411bdd.png)

RAID 10 ?

将RAID 1 和 RAID 0 结合起来，获得高安全性和高性能。最常用的RAID策略。同时也是TiDB，MySQL等数据库推荐的RAID策略。

![raid10.png](https://i.loli.net/2018/12/04/5c0673f475d50.png)

RAID 5

RAID 5 最低三块盘，存储数据的异或编码，在一块盘损坏时，可以提供编码恢复出数据。

![raid5.png](https://i.loli.net/2018/12/04/5c0673f454d91.png)

ElasticSearch 使用低速设备的 Tips

修改index.merge.scheduler.max_thread_count参数为1；该参数影响lucene后台的合并线程数量，默认设置只适合SDD。多个合并线程可能导致io压力过大，触发 (linux 120s timeout)[https://cyberdak.github.io/es/ ... -down].

存储策略

避免单机存储过多数据，如果单机故障，将导致集群需要大量数据，影响集群的吞吐量，特别是发生在高峰时候更会影响业务。千兆网卡每小时可以同步的数据为463gb，可以参考这个速度结合资深集群网卡以及存储来调节每个节点存储的数据量。
存储有条件使用RAID10，增加单节点性能以及避免单节点存储故障

RAID卡策略

根据服务器RAID卡的等级不同，高级的RAID卡可以使用 write-back 写策略，数据写入会直接写入到缓存中，随后刷新到硬盘上。当主机掉电时，由RAID卡带的电池来保证数据成功写入到硬盘中。write back的设置需要电池有电才能支持，而某些场景可以设置为force write-back(即使电池没电了，也要写缓存),从而提高写入性能。

通知设置 新通知

Advent

背景

使用场景

举例

公司正门卡机数据

架构部门禁数据

写在最后