Elastic Stack 7.0.0 正式版已经发布了，包含很多新的功能，大家可以去下载使用了。  https://www.elastic.co/blog/el ... eased       下载链接： Elasticsearch: https://www.elastic.co/downloa ... 7-0-0 Logstash: https://www.elastic.co/downloa ... 7-0-0 Kibana： https://www.elastic.co/downloa ... 7-0-0   其他模块及产品的下载： https://www.elastic.co/downloads  

1.GitHub为Elasticsearch发布golang库。 http://t.cn/EIshhxH 2.通过简单的步骤将Elasticsearch节点添加到集群。 http://t.cn/EIs7apw 3.(自备梯子)Apple Watch是否让您更健康。 http://t.cn/EIs4DKb 编辑：至尊宝 归档：https://elasticsearch.cn/article/6557 订阅：https://tinyletter.com/elastic-daily

平安科技诚招ES运维工程师，待遇福利丰厚，工作地点上海张江，有意者可以将简历发送至 CAIZHEN616@pingan.com.cn   岗位要求 1. 大学本科以上学历，计算机相关专业，至少3年以上的工作经验； 2. 熟悉linux操作系统、网络、sql等方面的知识，具有大型应用平台运维经验，有过运维自动化项目经验者优先； 3. 熟悉Python、shell语言,并且能够熟练使用Python、Shell进行脚本编写； 4..熟悉ansible、saltstack或其他运维自动化工具，了解elasticsearch、logstash、filebeat等开源软件优先 5. 熟悉docker、kubernetes、swarm等容器及编排技术，熟悉相关网络、存储解决方案，有相关经验优先 平安科技是平安集团的全资子公司，致力于运用人工智能、智能认知、云计算、区块链等前沿科技，为人们打造全新云生活。对内，平安科技是平安集团的高科技内核和科技企业孵化器，负责开发并运营集团的关键平台和服务。对外，平安科技以智慧科技为手段、以智造未来为蓝图，聚焦于医疗、金融、智慧城市三大领域，将国际权威认证的技术能力应用到实际业务场景中，打造生态闭环，积极践行科技改变生活的企业理念。 超过10000名专业IT技术人员和管理专家组成的高级研发团队，为平台的运营稳定和可靠，提供了专家级的技术保障。目前所建立的云生态圈已经承载过5亿的互联网用户，并拓展至海外市场，包括美国、新加坡、香港等国家和地区

1.选择日志管理解决方案时需要考虑的6件事。 http://t.cn/EIqkgoY 2.使用Elastic查询慢日志。 http://t.cn/EI5O93f 3.(自备梯子)学习Docker。 http://t.cn/EI5hnwf 编辑：至尊宝 归档：https://elasticsearch.cn/article/6498 订阅：https://tinyletter.com/elastic-daily

1.ElasticSearch如何与Mysql集成。 http://t.cn/EffVjlc 2.将数据从MySQL导入Elasticsearch并使用Kibana对其进行可视化。 http://t.cn/EfffRLT 3.(自备梯子)计算的未来是模拟的。 http://t.cn/EffxMjV 编辑：至尊宝 归档：https://elasticsearch.cn/article/6397 订阅：https://tinyletter.com/elastic-daily

1.在Elasticsearch数据库上公开2400万个信用和抵押记录。 http://t.cn/EtdTYAs 2.使用Elasticsearch构建长期安全运营平台。 http://t.cn/EtdHXAV 3.(自备梯子)飞行汽车比您想象的更接近现实。 http://t.cn/EtdQtfV 编辑：至尊宝 归档：https://elasticsearch.cn/article/6356 订阅：https://tinyletter.com/elastic-daily

本文将详细介绍利用 ES 与 Hive 直接的数据交互；通过 Hive 外部表的方式，可以快速将 ES 索引数据映射到 Hive 中，使用易于上手的 Hive SQL 实现对数据的进一步加工。

一、开发环境

1、组件版本

• CDH 集群版本：6.0.1
• ES 版本：6.5.1
• Hive 版本：2.1.1
• ES-Hadoop 版本：6.5.1

2、Hive 简介

Hive 在 Hadoop 生态系统中扮演着数据仓库的角色，借助 Hive 可以方便地进行数据汇总、即席查询以及分析存储在 Hadoop 文件系统中的大型数据集。

Hive 通过类 SQL 语言（HSQL）对 Hadoop 上的数据进行抽象，这样用户可以通过 SQL 语句对数据进行定义、组织、操作和分析；在 Hive 中，数据集是通过表（定义了数据类型相关信息）进行定义的，用户可以通过内置运算符或用户自定义函数（UDF）对数据进行加载、查询和转换。

3、Hive 安装 ES-Hadoop

官方推荐的安装方式：

使用 add jar

add jar /path/elasticsearch-hadoop.jar

使用 hive.aux.jars.path

$ bin/hive --auxpath=/path/elasticsearch-hadoop.jar

修改配置(hive-site.xml)

<property>
  <name>hive.aux.jars.path</name>
  <value>/path/elasticsearch-hadoop.jar</value>
  <description>A comma separated list (with no spaces) of the jar files</description>
</property>

CDH6.X 推荐的安装方法

将 elasticsearch-hadoop.jar 复制到 Hive 的 auxlib 目录中，然后重启 Hive 即可。

cp elasticsearch-hadoop.jar /opt/cloudera/parcels/CDH/lib/hive/auxlib/

二、Hive 与 ElasticSearch 的数据交互

1、数据类型对照表

请务必注意，ES 中的类型是 index/_mapping 中对应的数据类型，非 _source 里面数据的类型。

Hive type	Elasticsearch type
void	null
boolean	boolean
tinyint	byte
smallint	short
int	int
bigint	long
double	double
float	float
string	string
binary	binary
timestamp	date
struct	map
map	map
array	array
union	not supported (yet)
decimal	string
date	date
varchar	string
char	string

2、建立 Hive 外部表

CREATE EXTERNAL TABLE default.surface(
    water_type STRING,
    water_level STRING,
    monitor_time TIMESTAMP,
    sitecode STRING,
    p492 DOUBLE,
    p311 DOUBLE,
    status STRING
)
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES(
    'es.resource'='ods_data_day_surface*/doc',
    'es.query'='?q=status:001'
    'es.nodes'='sky-01','es.port'='9200',
    'es.net.http.auth.user'='sky',
    'es.net.http.auth.pass'='jointsky',
    'es.date.format'='yyyy-MM-dd HH:mm:ss',
    'es.ser.reader.value.class'='com.jointsky.bigdata.hive.EsValueReader'
    'es.mapping.names'='waterType:water_type,monitortime:monitor_time'
);

3、配置项说明

es.resource

es.resource 用于设置 ES 资源的位置，默认该配置项同时设置了读和写的索引，当然也可以分别设置读写索引名称：

• es.resource.read：设置读取位置；
• es.resource.write：设置写入位置。

es.query

es.query 设置查询过滤条件，目前支持 uri query、query dsl、external resource 三种设置方式。

# uri (or parameter) query
es.query = ?q=costinl

# query dsl
es.query = { "query" : { "term" : { "user" : "costinl" } } }

# external resource
es.query = org/mypackage/myquery.json

es.mapping.names

es.mapping.names 用于设置 Hive 与 ES 的字段映射关系，如果不设置，则默认字段名不发生变化（即为 data type 区域定义的字段名）；此外该部分还用于定义 Hive 到 ES 的数据映射类型。

'es.mapping.names' = 'date:@timestamp , url:url_123 ')

其他通用字段的说明请参考文章：使用 ES-Hadoop 将 Spark Streaming 流数据写入 ES

4、自定义日期类型解析

目前将 ES 的 date 类型映射到 Hive 的 TIMESTAMP 类型时，ES-Hadoop 组件只能识别时间戳格式或者标准的 XSD 格式的日期字符串：

@Override
protected Object parseDate(Long value, boolean richDate) {
    return (richDate ? new TimestampWritable(new Timestamp(value)) : processLong(value));
}

@Override
protected Object parseDate(String value, boolean richDate) {
    return (richDate ? new TimestampWritable(new Timestamp(DatatypeConverter.parseDateTime(value).getTimeInMillis())) : parseString(value));
}

关于 XSD（XML Schema Date/Time Datatypes）可用参考文章：https://www.w3schools.com/xml/schema_dtypes_date.asp

为了兼容自定义的日期格式，需要编写自定义的日期读取类：

import org.apache.hadoop.hive.serde2.io.TimestampWritable;
import org.elasticsearch.hadoop.cfg.Settings;
import org.elasticsearch.hadoop.hive.HiveValueReader;

import java.sql.Timestamp;
import java.text.ParseException;
import java.text.ParsePosition;
import java.text.SimpleDateFormat;
import java.util.Date;

public class EsValueReader extends HiveValueReader {
    private String dateFormat;
    private static final String DEFAULT_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    private static final String DEFAULT_DATE_FORMAT_MIN = "yyyy-MM-dd HH:mm";
    private static final String DEFAULT_DATE_FORMAT_HOUR = "yyyy-MM-dd HH";
    private static final String DEFAULT_DATE_FORMAT_DAY = "yyyy-MM-dd";

    @Override
    public void setSettings(Settings settings) {
        super.setSettings(settings);
        dateFormat = settings.getProperty("es.date.format");
    }

    @Override
    protected Object parseDate(String value, boolean richDate) {
        if (value != null && value.trim().length() > 0 && DEFAULT_DATE_FORMAT.equalsIgnoreCase(dateFormat)) {
            if (richDate){
                if (value.length() == 16){
                    return new TimestampWritable(new Timestamp(parseDate(value, DEFAULT_DATE_FORMAT_MIN).getTime()));
                }
                if (value.length() == 13){
                    return new TimestampWritable(new Timestamp(parseDate(value, DEFAULT_DATE_FORMAT_HOUR).getTime()));
                }
                if (value.length() == 10){
                    return new TimestampWritable(new Timestamp(parseDate(value, DEFAULT_DATE_FORMAT_DAY).getTime()));
                }
                return new TimestampWritable(new Timestamp(parseDate(value, DEFAULT_DATE_FORMAT).getTime()));
            }
            return parseString(value);
        }
        return super.parseDate(value, richDate);
    }

    /**
     * 解析日期,根據指定的格式進行解析.<br>
     * 如果解析錯誤,則返回null
     * @param stringDate 日期字串
     * @param format 日期格式
     * @return 日期型別
     */
    private static Date parseDate(String stringDate, String format) {
        if (stringDate == null) {
            return null;
        }
        try {
            return parseDate(stringDate, new String[] { format });
        } catch (ParseException e) {
            return null;
        }
    }

    public static Date parseDate(String str, String... parsePatterns) throws ParseException {
        return parseDateWithLeniency(str, parsePatterns, true);
    }

    private static Date parseDateWithLeniency(
            String str, String[] parsePatterns, boolean lenient) throws ParseException {
        if (str == null || parsePatterns == null) {
            throw new IllegalArgumentException("Date and Patterns must not be null");
        }

        SimpleDateFormat parser = new SimpleDateFormat();
        parser.setLenient(lenient);
        ParsePosition pos = new ParsePosition(0);
        for (String parsePattern : parsePatterns) {
            String pattern = parsePattern;
            if (parsePattern.endsWith("ZZ")) {
                pattern = pattern.substring(0, pattern.length() - 1);
            }
            parser.applyPattern(pattern);
            pos.setIndex(0);
            String str2 = str;
            if (parsePattern.endsWith("ZZ")) {
                str2 = str.replaceAll("([-+][0-9][0-9]):([0-9][0-9])$", "$1$2");
            }
            Date date = parser.parse(str2, pos);
            if (date != null && pos.getIndex() == str2.length()) {
                return date;
            }
        }
        throw new ParseException("Unable to parse the date: " + str, -1);
    }
}

上述代码的 Maven 依赖

<dependencies>
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-exec</artifactId>
        <version>2.1.1</version>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch-hadoop</artifactId>
        <version>6.5.4</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

自定义日期解析包的部署

代码编写完成后，将代码进行打包，然后将打包好的 jar 包放置到 Hive 的 auxlib 目录中，然后重启 Hive 即可；该步骤与 ES-Hadoop 的安装步骤一样。

在编写 Spark 程序从 Hive 中读取数据的时候，需要添加对该包的依赖以及对 ES-Hadoop 的依赖。

三、总结

经过上述的步骤，Hive 与 ES 的映射已经不成问题，如果想从 ES 中导出数据，可用借助 HSQL insert into table XXX select * from XXXXX; 的方式从 ES 中读取数据写入到 HDFS；当然通过更为复杂的 HSQL 可以将数据进行处理，并将数据重新写入到 ES 或者存储到 HDFS。

充分利用 ES 的查询、过滤和聚合，可以很好的去服务数据标准化、数据清洗、数据分布情况等 ETL 流程。

---

Any Code，Code Any！

扫码关注『AnyCode』，编程路上，一起前行。

1.在现有Elasticsearch集群中添加或删除节点。 http://t.cn/EtUHo2i 2.运行400+节点Elasticsearch集群。 http://t.cn/EtUHqmF 3.(自备梯子)亚马逊变得比你意识到的更强大。 http://t.cn/EtUHBLe 编辑：至尊宝 归档：https://elasticsearch.cn/article/6348 订阅：https://tinyletter.com/elastic-daily

为ES构建Stanford NLP分词插件

Stanford NLP？

Stanford分词器是斯坦福大学NLP团队维护的一个开源分词器，支持了包括中文、英文…的语言，而且除了分词之外，它还支持了包括词性分析、情感分析…的各种功能。\ 这俩是这个project的项目主页

• Home page
• GitHub page

Why Stanford core NLP？

  市面上确实会有很多很有名的开源分词器，比如IK、Jieba，还有一些其他团队和公司提供的开源/商用的分词器，他们各有优劣。但是在各种分词器上比较了一大堆的分词case之后，我们发现Stanford NLP似乎是最适合我们当前需求的一个，因为我们不仅仅需要分词，还需要一些包括情感分析之类在内的更多的一些功能。

我们公司是做金融数据的搜索推荐的，在对比了各家分词器之后我们老板觉得Stanford NLP的效果最好，但是作为算法出身的人，他实现了一套非常重的分词、排序、搜索的服务。

在对比如研报、财报之类的信息进行搜索的时候确实会比较有效，但是在对经济类的新闻进行搜索的时候就会显得十分的笨重。

基于这个背景，我开始试图在ES里面引入老板推崇的Stanford 分词器来适应他的搜索、分词的需要，同时也能够不通过他那个笨重的分词排序服务来对我们系统中大量的经济、金融类的新闻进行分词、索引，并提供和他自己分词效果类似的分词和检索服务。

Why this project

我在包括百度、某谷姓404网站、GitHub以及国内的中文社区（Elastic中文社区)在内的各种地方搜过也问过了，但是似乎没有一个直接开箱可用的分词插件。所以，我只剩一条路了，就是搭建一个自己的插件来引用这个分词器。

How

对ES来说，插件主要分为两个部分：

1. 让ES可以看到的部分（class extends Plugin）
2. 自己行使职能的部分（functional part）

plugin

1. 为了让ES可以加载我们的plugin，我们需要先继承Plugin类，然后我们这个是个分词器插件，所以还要实现AnalysisPlugin类
2. 看过ES源码或者其他分词器源码的同学应该会知道，分词器插件需要实现两个方法，一个用来提供tokenizer，一个是analyzer分别对应分词器中的这俩。
   • 重写Map<String, AnalysisModule.AnalysisProvider<TokenizerFactory>>是为了可以提供搜索用分词器
   • 重写Map<String, AnalysisModule.AnalysisProvider<AnalyzerProvider<? extends Analyzer>>>是为了可以提供索引用分词器
3. 在这个分词器里面我们主要是依靠Tokenizer来实现分词的

functional class

分词器，特别是Tokenizer主要是靠重写三个方法来实现分词的

1. incrementToken：用来确定每一个词元，输出每一个单词（字）以及它的位置、长度等
2. reset：用来重制分词结果
3. end：用来告诉ES，这段文本的分词已经结束了

所以我们主要需要重写的就是这仨方法，当然了，为了能让分词器正确的使用，我们还需要添加一些分词器的配置和初始化的内容，具体代码不写了可以参考我的git，主要讲两个坑：

1. ES是通过配置文件里的路径来寻找对应的插件类
2. 然后通过配置文件里的key和刚才提到的代码里的key来寻找对应的分词器，所以这俩地方不要写错了 #plugin-descriptor.properties: classname=org.elasticsearch.plugin.analysis.AnalysisSDPlugin #plugin-descriptor.properties: name=stanford-core-nlp
3. 在开发过程中由于有java-security的存在，所以需要通过AccessController来调用和加载我们需要的外部jar包

odds and ends

1. Stanford分词器里面包含了很多功能，目前我使用了分词的部分
2. 分词器自带词典文件，不过如果要做词典的修改可能需要解包，修改，再重新打包
3. 我现在hardcode了一大堆的标点符号在里面，后面可能会去优化一下部分逻辑
4. 待完成的功能还有其他功能包括情感分析之类的

also see

GitHub 地址

1.Java应用日志导入ELK。 http://t.cn/E5GiA1T 2.使用Tokens分发Cassandra数据。 http://t.cn/E5GIOd5 3.(自备梯子)为什么如此难以让计算机像人一样说话？ http://t.cn/EqFOf04 编辑：至尊宝 归档：https://elasticsearch.cn/article/6337 订阅：https://tinyletter.com/elastic-daily

1.创建DSL。 http://t.cn/EqcmScm 2.使用ANTLR从GSA迁移到Elasticsearch。 http://t.cn/Eqc3j1w 3.(自备梯子)将定义2019年的技术。 http://t.cn/EqUQ61i 编辑：至尊宝 归档：https://elasticsearch.cn/article/6329 订阅：https://tinyletter.com/elastic-daily

1.Elasticsearch查询与Spring Data。 http://t.cn/EGM5RCs 2.Spring Data Elasticsearch介绍。 http://t.cn/EAE698A 3.(自备梯子)区块链是互联网失败的提示。 http://t.cn/EbmQ42J 编辑：至尊宝 归档：https://elasticsearch.cn/article/6319 订阅：https://tinyletter.com/elastic-daily

本文将详细介绍利用 ES-Hadoop 将 Spark 处理的数据写入到 ES 中。

一、开发环境

1、组件版本

• CDH 集群版本：6.0.1
• Spark 版本：2.2.0
• Kafka 版本：1.0.1
• ES 版本：6.5.1

2、Maven 依赖

<!-- scala -->
<dependency>
  <groupId>org.scala-lang</groupId>
  <artifactId>scala-library</artifactId>
  <version>2.11.8</version>
</dependency>

<!-- spark 基础依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- spark-streaming 相关依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- spark-streaming-kafka 相关依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- zookeeper 相关依赖 -->
<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.4.5-cdh6.0.1</version>
</dependency>

<!-- Spark-ES 相关依赖 -->
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-spark-20_2.11</artifactId>
    <version>6.5.4</version>
</dependency>

<!-- Spark-ES 依赖的 HTTP 传输组件 -->
<dependency>
    <groupId>commons-httpclient</groupId>
    <artifactId>commons-httpclient</artifactId>
    <version>3.1</version>
</dependency>

3、注意事项

如果使用 CDH 版本的 Spark，则在调试及实际部署运行的时候会出现下面的错误：

java.lang.ClassNotFoundException: org.apache.commons.httpclient.protocol.Protocol

很显然是缺少 httpclient 相关依赖造成的，对比开源版本与 CDH 版本的 Spark，发现开源版本多出了 commons-httpclient-3.1.jar，因此上述 Maven 的 pom 文件添加上对其依赖即可。

二、ES-Hadoop

1、简介

ES-Hadoop 实现了 Hadoop 生态（Hive、Spark、Pig、Storm 等）与 ElasticSearch 之间的数据交互，借助该组件可以将 Hadoop 生态的数据写入到 ES 中，然后借助 ES 对数据快速进行搜索、过滤、聚合等分析，进一步可以通过 Kibana 来实现数据的可视化。

同时，也可以借助 ES 作为数据存储层（类似数仓的 Stage 层或者 ODS 层），然后借助 Hadoop 生态的数据处理工具（Hive、MR、Spark 等）将处理后的数据写入到 HDFS 中。

使用 ES 做为原始数据的存储层，可以很好的进行数据去重、数据质量分析，还可以提供一些即时的数据服务，例如趋势展示、汇总分析等。

2、组成

ES-Hadoop 是一个整合性质的组件，它封装了 Hadoop 生态的多种组件与 ES 交互的 API，如果你只需要部分功能，可以使用细分的组件：

• elasticsearch-hadoop-mr
• elasticsearch-hadoop-hive
• elasticsearch-hadoop-pig
• elasticsearch-spark-20_2.10
• elasticsearch-hadoop-cascading
• elasticsearch-storm

三、elasticsearch-spark

1、配置

es-hadoop 核心是通过 es 提供的 restful 接口来进行数据交互，下面是几个重要配置项，更多配置信息请参阅官方说明：

• es.nodes：需要连接的 es 节点（不需要配置全部节点，默认会自动发现其他可用节点）；
• es.port：节点 http 通讯端口；
• es.nodes.discovery：默认为 true，表示自动发现集群可用节点；
• es.nodes.wan.only：默认为 false，设置为 true 之后，会关闭节点的自动 discovery，只使用 es.nodes 声明的节点进行数据读写操作；如果你需要通过域名进行数据访问，则设置该选项为 true，否则请务必设置为 false；
• es.index.auto.create：是否自动创建不存在的索引，默认为 true；
• es.net.http.auth.user：Basic 认证的用户名；
• es.net.http.auth.pass：Basic 认证的密码。

val conf = new SparkConf().setIfMissing("spark.app.name","rt-data-loader").setIfMissing("spark.master", "local[5]")
conf.set(ConfigurationOptions.ES_NODES, esNodes)
conf.set(ConfigurationOptions.ES_PORT, esPort)
conf.set(ConfigurationOptions.ES_NODES_WAN_ONLY, "true")
conf.set(ConfigurationOptions.ES_INDEX_AUTO_CREATE, "true")
conf.set(ConfigurationOptions.ES_NODES_DISCOVERY, "false")
conf.set(ConfigurationOptions.ES_NET_HTTP_AUTH_USER, esUser)
conf.set(ConfigurationOptions.ES_NET_HTTP_AUTH_PASS, esPwd)
conf.set("es.write.rest.error.handlers", "ignoreConflict")
conf.set("es.write.rest.error.handler.ignoreConflict", "com.jointsky.bigdata.handler.IgnoreConflictsHandler")

特别需要注意的配置项为 es.nodes.wan.only，由于在云服务器环境中，配置文件使用的一般为内网地址，而本地调试的时候一般使用外网地址，这样将 es.nodes 配置为外网地址后，最后会出现节点找不到的问题（由于会使用节点配置的内网地址去进行连接）：

org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: No data nodes with HTTP-enabled available; 
node discovery is disabled and none of nodes specified fit the criterion [xxx.xx.x.xx:9200]

此时将 es.nodes.wan.only 设置为 true 即可。推荐开发测试时使用域名，集群部署的时候将该选项置为 false。

2、屏蔽写入冲突

如果数据存在重复，写入 ES 时往往会出现数据写入冲突的错误，此时有两种解决方法。

方法一：设置 es.write.operation 为 upsert，这样达到的效果为如果存在则更新，不存在则进行插入，该配置项默认值为 index。

方法二：自定义冲突处理类，类似上述配置中设置了自定义的 error.handlers，通过自定义类来处理相关错误，例如忽略冲突等：

public class IgnoreConflictsHandler extends BulkWriteErrorHandler {
    public HandlerResult onError(BulkWriteFailure entry, DelayableErrorCollector<byte[]> collector) throws Exception {
        if (entry.getResponseCode() == 409) {
            StaticLog.warn("Encountered conflict response. Ignoring old data.");
            return HandlerResult.HANDLED;
        }
        return collector.pass("Not a conflict response code.");
    }
}

方法二可以屏蔽写入版本比预期的小之类的版本冲突问题。

3、RDD 写入 ES

EsSpark 提供了两种主要方法来实现数据写入：

• saveToEs ：RDD 内容为 Seq[Map]，即一个 Map 对象集合，每个 Map 对应一个文档；
• saveJsonToEs：RDD 内容为 Seq[String]，即一个 String 集合，每个 String 是一个 JSON 字符串，代表一条记录（对应 ES 的 _source）。

数据写入可以指定很多配置信息，例如：

• es.resource：设置写入的索引和类型，索引和类型名均支持动态变量；
• es.mapping.id：设置文档 _id 对应的字段名；
• es.mapping.exclude：设置写入时忽略的字段，支持通配符。

val itemRdd = rdd.flatMap(line => {
    val topic = line.topic()
    println("正在处理：" + topic + " - " + line.partition() + " : " + line.offset())
    val jsonArray = JSON.parseArray(line.value()).toJavaList(classOf[JSONObject]).asScala
    val resultMap = jsonArray.map(jsonObj =>{
      var tmpId = "xxx"
      var tmpIndex = "xxxxxx"
      jsonObj.put("myTmpId", tmpId)
      jsonObj.put("myTmpIndex", tmpIndex)
      jsonObj.getInnerMap
    })
    resultMap
})
val mapConf = Map(
    ("es.resource" , "{myTmpIndex}/doc"),
    ("es.write.operation" , "upsert"),
    ("es.mapping.id" , "myTmpId"),
    ("es.mapping.exclude" , "myTmp*")
)
EsSpark.saveToEs(itemRdd, mapConf)

es.mapping.exclude 只支持 RDD 为 Map 集合（saveToEs），当为 Json 字符串集合时（saveJsonToEs）会提示不支持的错误信息；这个配置项非常有用，例如 myTmpId 作为文档 id，因此没有必要重复存储到 _source 里面了，可以配置到这个配置项，将其从 _source 中排除。

---

Any Code，Code Any！

扫码关注『AnyCode』，编程路上，一起前行。

1.使用ElasticSearch进行全文检索。 http://t.cn/Eb9Ahhr 2.使用Flink和Kafka构建数据管道。 http://t.cn/Eb92onY 3.(自备梯子)Jupyter Notebook 扩展。 http://t.cn/EyD8Ao5 编辑：至尊宝 归档：https://elasticsearch.cn/article/6312 订阅：https://tinyletter.com/elastic-daily

一、分词插件

1、分词器概念

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

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

2、选择分词器

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

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

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

二、HanLP 简介

HanLP 是一系列模型与算法组成的 NLP 工具包，具备功能完善、性能高效、架构清晰、语料时新、可自定义的特点,详情可参考 github 介绍：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：插件属性文件所在目录

2、TokenStream

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

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 <<ALL FILES>> 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/8/docs/technotes/guides/security/permissions.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/elasticsearch-analysis-hanlp