介绍

本次想和大家分享一款Elasticsearch分词插件，该插件是基于天津海量信息股份有限公司的中文分词核心开发的。海量分词针对大数据检索场景专门做了定制和优化，更贴近搜索需求，整体分词的性能也是非常高效。

本文章有广告成分。但希望将公司研究成果分享出来，给大家实际工作中多一种选择...

海量分词检索优化点

• 地名方面海量分词5.0可以识别并检索出关于地名后缀的结果

  可以通过搜索“河南”得到“河南省”的结果，搜索“天津”得到“天津市”的搜索结果，而不是简单河南、天津的识别。

• 著名人物的人名识别更精准，如刘翔、傅莹等

  部分分词器处理中文分词只有两种方式：一种是单字（unigrams）形式，即简单粗暴的将中文的每一个汉字作为一个词（token）分开；另一种是两字（bigrams）的，也就是任意相邻的两个汉字作为一个词分开。这种简单粗暴的切分方式无法实现时效性较新的人名识别，如刘翔、傅莹等会被识别为单字切开。

• 外国人名识别方面海量可以将人名识别智能识别

  “玛利亚 凯利”、“乔治·史密斯”、“玛丽·戴维斯”将完整的外国人名识别出姓氏和名，如“乔治·史密斯”可以被识别为“乔治”和 “史密斯”。

• 常见词的品牌名称识别方面，海量分词5.0识别的结果中包含实际意义的品牌名称

  如“乐高”，“吉米作为简单的词，可以被识别，但是词放在文档语境中有其品牌的属性，海量分词识别的结果中可以准确搜索出品牌的结果。

• 机构名识别方面

  海量分词5.0可以识别完整的机构名称，如“天津海量信息技术股份有限公司”，可以完整的识别出全称。

海量分词性能评测

评测用例

本次评测选取的语料一共三个。一个是2MB的海量测试语料，一个是4MB的北大语料（新版旧版各2MB），一个是9.4GB海量的线上实际数据

评测指标

本次评测是在开源评测程序上修改而来，评测指标有分词速度、行数完美率、字数完美率（该指标仅供参考）、内存消耗

评测结果

2MB海量测试语料

可以看到海量分词的行数完美率是最高的，而且速度十分优异；仅有的一个比海量分词速度快的算法是一个追求极限性能舍弃准确率的算法

4MB北大语料

（注：THULAC和stanford由于速度问题，不纳入评测）

可以看到海量的速度和行数完美率都很优异而且达到了兼顾，行数完美率只落后更高的ltp算法1.4个百分点，速度却是它的三十多倍

9.4GB线上数据

（注：THULAC和stanford由于速度问题，不纳入评测）

本表格中分词顺序按（4MB北大语料的）行数完美率进行排序，越靠前的（4MB北大语料的）行数完美率越高

可以看出海量的分词速度十分优秀，分词速度拉开了大多数分词数倍，相比于行数完美率小幅领先的ltp要快几十倍

海量分词插件使用方法

安装使用

• 下载安装 - 地址: https://github.com/HylandaOpen/elasticsearch-analysis-hlseg/releases

  unzip plugin to folder `your-es-root/plugins/`

• 使用 elasticsearch-plugin 安装

  ./bin/elasticsearch-plugin install https://github.com/HylandaOpen/elasticsearch-analysis-hlseg/releases/download/v6.4.2/elasticsearch-analysis-hlseg-6.4.2.zip

• 重启es集群

实例（借用github-ik分词插件的实例）

1.创建index

curl -XPUT http://localhost:9200/hylanda_seg

2.配置mapping

curl -XPOST http://localhost:9200/hylanda_seg/data/_mapping -H 'Content-Type:application/json' -d'
{
  "properties": {
    "msg": {
      "type": "text",
      "analyzer": "hlseg_search"
    }
  }
}'

3.插入测试数据

curl -XPOST http://localhost:9200/hylanda_seg/data/1 -H 'Content-Type:application/json' -d'
{"content":"美国留给伊拉克的是个烂摊子吗"}
'

curl -XPOST http://localhost:9200/hylanda_seg/data/2 -H 'Content-Type:application/json' -d'
{"content":"公安部：各地校车将享最高路权"}
'

curl -XPOST http://localhost:9200/hylanda_seg/data/3 -H 'Content-Type:application/json' -d'
{"content":"中韩渔警冲突调查：韩警平均每天扣1艘中国渔船"}
'

curl -XPOST http://localhost:9200/hylanda_seg/data/4 -H 'Content-Type:application/json' -d'
{"content":"中国驻洛杉矶领事馆遭亚裔男子枪击 嫌犯已自首"}
'

4.查询

curl -XPOST http://localhost:9200/hylanda_seg/data/_search  -H 'Content-Type:application/json' -d'
{
  "query": {
    "match": {
      "content": "中国"
    }
  },
  "highlight": {
    "fields": {
      "content": {}
    }
  }
}
'

返回结果

{
  "took" : 11,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 2,
    "max_score" : 0.5754429,
    "hits" : [
      {
        "_index" : "hylanda_seg",
        "_type" : "data",
        "_id" : "4",
        "_score" : 0.5754429,
        "_source" : {
          "content" : "中韩渔警冲突调查：韩警平均每天扣1艘中国渔船"
        },
        "highlight" : {
          "content" : [
            "中韩渔警冲突调查：韩警平均每天扣1艘<em>中国</em>渔船"
          ]
        }
      },
      {
        "_index" : "hylanda_seg",
        "_type" : "data",
        "_id" : "5",
        "_score" : 0.2876821,
        "_source" : {
          "content" : "中国驻洛杉矶领事馆遭亚裔男子枪击 嫌犯已自首"
        },
        "highlight" : {
          "content" : [
            "<em>中国</em>驻洛杉矶领事馆遭亚裔男子枪击 嫌犯已自首"
          ]
        }
      }
    ]
  }
}

字典配置

海量分词分为基础词词典CoreDict.dat和自定义词典userDict_utf8.txt。基础词词典在dictionary目录下，需要将CoreDict.zip解压后放在config目录下，可以通过修改config下的userDict_utf8.txt来更新自定义词典

自定义词典格式如下

1.用户自定义词典采用文本格式，utf-8编码，每行一个词

2.每个词包含三列属性，分别是词串、词的属性以及idf值的加权等级，并以Tab作为分隔，其中除了词串必填外，其他列可以不填，不填写则系统采用默认值

3.“#”表示注释，会在加载时被忽略

4.词的属性以西文逗号分隔，可以是词性、停止词标志或者自定义属性

5.词性标记参考北大标准，用于词性标注时参考，该项不填则默认为名词

6.停止词标志为：stopword，由SegOption.outputStopWord来控制是否输出停止词

7.自定义属性不参与分词过程，分词结果中若Token.userTag不为空，则可以获取到该词的自定义属性。

8.idf值的加权分5级，从低到高的定义是idf-lv1 — idf-lv5，等级越高则该词在关键词计算时的权重会越大，若不填写该值则系统默认是idf-lv3(中等权重）

	BasicDataSource basicDataSource = new BasicDataSource();

	// 创建连接池 一次性创建多个连接池



	// 连接池 创建连接 ---需要四个参数

	basicDataSource.setDriverClassName("com.elasticsearch.jdbc.ElasticsearchDriver");

	basicDataSource.setUrl("jdbc:elasticsearch://127.0.0.1:5000");



	// 从连接池中获取连接

	Connection conn = basicDataSource.getConnection();

	String sql = "select SRC_IP,SRC_PORT from \"my_test-*\"";

	Statement stmt = conn.createStatement();

	ResultSet rs = stmt.executeQuery(sql);

	while (rs.next()) {

		System.out.println(rs.getString("SRC_IP"));

	}

	basicDataSource.close();





	String sql = "select SRC_IP,SRC_PORT from my_test* where SRC_PORT between 10 and 100 limit 1000";

	String url = "jdbc:elasticsearch://127.0.0.1:5000";

	Connection connection = DriverManager.getConnection(url, "test", null);

	Statement statement = connection.createStatement();

	ResultSet rs = statement.executeQuery(sql);

	ResultSetMetaData meta = rs.getMetaData();

	String columns = "|";

	for (int i = 0; i < meta.getColumnCount(); i++) {

		columns += meta.getColumnLabel(i) + " | ";

	}

	System.out.println(columns);

	while (rs.next()) {

		String row = "|";

		for (int i = 0; i < meta.getColumnCount(); i++) {

			row += rs.getString(i) + " | ";

		}

		System.out.println(row);

	}   

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/downloads/elasticsearch-plugins/repository-hdfs/repository-hdfs-5.0.2.zip 保存在/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 http://192.168.10.74:9200/_snapshot/backup
  {  
  "type": "hdfs",  
    "settings": {  
            "uri": "hdfs://192.168.10.170:9000",  
            "path": "/es",  
            "conf_location": "/usr/local/hadoop/etc/hadoop/hdfs-site.xml"  
    }
  }

• 备份快照

  PUT http://192.168.10.74:9200/_snapshot/backup/snapshot_users?wait_for_completion=true
  {
  "indices": "users",  //备份users的index，注意不设置这个属性，默认是备份所有index
  "ignore_unavailable": true,
  "include_global_state": false
  }

• 恢复快照

  POST http://192.168.10.74:9200/_snapshot/backup/snapshot_users/_restore
  {
  "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中进行查询

引入依赖

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-hadoop</artifactId>
    <version>5.0.2</version>
</dependency>

具体流程

• 数据在HDFS上，数据存储在HDFS的每个DataNode的block上

• 数据加载到Spark SQL

  • 数据从HDFS加载到Spark SQL中，以RDD形式存储

  JavaRDD<String> textFile = spark.read().textFile("hdfs://192.168.10.170:9000/csv/user.csv")

  • 添加数据结构信息转换为新的RDD

  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

  DataSet<Row> ds = spark.createDataFrame(dataSplits, UserItem.class);

• 由Dataset创建索引，并写入ES

  JavaEsSparkSQL.saveToEs(ds, "es_spark/users");

• 数据在ES中建立索引

• Spark SQL通过索引对ES中的数据进行查询

  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的强大检索性能帮助我们快速分析海量数据。

{

  "仓ku": "华南",

   "originName":  "",

   "Code":  "23248",

   "BrandName":  "",

   "originCode":  null,

   "CategoryName":  "原厂"

}

1. 背景

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

2. 日志处理基本流程

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

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

使用Elastic Stack处理日志的详细过程，用户可参考官方文章Getting started with the Elastic Stack，这里不展开介绍。

3. 日志场景调优

       对于Elasticsearch的通用调优，之前分享的文章Elasticsearch调优实践，详细介绍了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及之后的版本，支持关键字查询时自动选择目标字段，用户没有必要再使用原始日志字段提供不指定字段进行查询的能力。

4. 小结

日志的使用方式比较灵活，本文结合常见的客户使用方式，从整体上对性能、成本进行优化。用户也可结合自身业务场景，参考文章Elasticsearch调优实践进行更细致的优化。

当我们搭建elasitc stack集群时，大多数时候会在我们的架构中加入kafka作为消息缓冲区，即从beats -> kafka -> logstash -> elasticsearch这样的一个消息流。使用kafka可以给我们带来很多便利，但是也让我们需要额外多维护一套组件，elasitc stack本身已经提供了monitoring的功能，我们可以方便的从kibana上监控各个组件中各节点的可用性，吞吐和性能等各种指标，但kafka作为架构中的组件之一却游离在监控之外，相当不合理。

幸而elastic真的是迭代的相当快，在metricbeat上很早就有了对kafka的监控，但一直没有一个直观的dashboard，终于在6.5版本上，上新了kafka dashboard。我们来看一下吧。

安装和配置metricbeat

安装包下载地址，下载后，自己安装。 然后，将/etc/metricbeat/modules.d/kafka.yml.disable文件重命名为/etc/metricbeat/modules.d/kafka.yml。(即打开kafka的监控)。稍微修改一下文件内容, 注意，这里需填入所有你需要监控的kafka服务器的地址：

# Module: kafka
# Docs: https://www.elastic.co/guide/en/beats/metricbeat/6.4/metricbeat-module-kafka.html

- module: kafka
  metricsets:
    - partition
    - consumergroup
  period: 20s
  hosts: ["10.*.*.*:9092","10.*.*.*:9092","10.*.*.*:9092","10.*.*.*:9092"]

  #client_id: metricbeat
  #retries: 3
  #backoff: 250ms

  # List of Topics to query metadata for. If empty, all topics will be queried.
  #topics: []

  # 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"

  # SASL authentication
  #username: ""
  #password: ""

运行metricbeat，这里，一定要注意enable kibana dashboard。

然后就可以在kibana里面看到：

这样，我们就可以通过sentinl等类似的插件，自动做kafka的告警等功能了

@[toc] 搭建Elasitc stack集群时，我们往往把大部分注意力放在集群的搭建，索引的优化，分片的设置上等具体的调优参数上，很少有人会去关心Elasitc stack的日志配置的问题，大概是觉得，日志应该是一个公共的问题，默认的配置应该已经为我们处理好了。但很不幸，在不同的机器配置或者不同的运营策略下，如果采用默认的配置，会给我们带来麻烦。

默认配置带来的麻烦

以下例子是默认情况下，当Elasitc stack集群运行超过3个月之后的情况：

elasticsearch

elasticsearch默认情况下会每天rolling一个文件，当到达2G的时候，才开始清除超出的部分，当一个文件只有几十K的时候，文件会一直累计下来。

logstash

一直增长的gc文件和不停增多的rolling日志文件

kibana

默认日志输出到kibana.out文件当中，这个文件会变得越来越大

kafka

这里提到kafka是因为在大部分的架构当中，我们都会用到kafka作为中间件数据缓冲区，因此不得不维护kafka集群。同样，如果不做特定的配置，也会遇到日志的问题：不停增多的rolling日志文件

原因是kafka的默认log4j配置是使用DailyRollingFileAppender每隔一个小时生成一个文件 '.'yyyy-MM-dd-HH：

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.kafkaAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log
log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.requestAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.requestAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log
log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.cleanerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.cleanerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log
log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.controllerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.controllerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log
log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

解决方案

因此，对于我们需要维护的这几个组件，需要配置合理的日志rotate策略。一个比较常用的策略就是时间+size，每天rotate一个日志文件或者每当日志文件大小超过256M，rotate一个新的日志文件，并且最多保留7天之内的日志文件。

elasticsearch 

通过修改log4j2.properties文件来解决。该文件在/etc/elasticsesarch目录下（或者config目录）。 默认配置是：

appender.rolling.type = RollingFile 
appender.rolling.name = rolling
appender.rolling.fileName = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}.log 
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c{1.}] [%node_name]%marker %.-10000m%n
appender.rolling.filePattern = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}-%d{yyyy-MM-dd}-%i.log.gz 
appender.rolling.policies.type = Policies
appender.rolling.policies.time.type = TimeBasedTriggeringPolicy 
appender.rolling.policies.time.interval = 1 
appender.rolling.policies.time.modulate = true 
appender.rolling.policies.size.type = SizeBasedTriggeringPolicy 
appender.rolling.policies.size.size = 256MB 
appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.fileIndex = nomax
appender.rolling.strategy.action.type = Delete 
appender.rolling.strategy.action.basepath = ${sys:es.logs.base_path}
appender.rolling.strategy.action.condition.type = IfFileName 
appender.rolling.strategy.action.condition.glob = ${sys:es.logs.cluster_name}-* 
appender.rolling.strategy.action.condition.nested_condition.type = IfAccumulatedFileSize 
appender.rolling.strategy.action.condition.nested_condition.exceeds = 2GB 

以上默认配置，会保存2GB的日志，只有累计的日志大小超过2GB的时候，才会删除旧的日志文件。 建议改为如下配置，仅保留最近7天的日志

appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.action.type = Delete
appender.rolling.strategy.action.basepath = ${sys:es.logs.base_path}
appender.rolling.strategy.action.condition.type = IfFileName
appender.rolling.strategy.action.condition.glob = ${sys:es.logs.cluster_name}-*
appender.rolling.strategy.action.condition.nested_condition.type = IfLastModified
appender.rolling.strategy.action.condition.nested_condition.age = 7D

这里必须注意，log4j2会因为末尾的空格导致无法识别配置

logstash

与elasticsearch类似，通过修改log4j2.properties文件来解决。该文件在/etc/logstash目录下（或者config目录）。 默认配置是不会删除历史日志的：

status = error
name = LogstashPropertiesConfig

appender.console.type = Console
appender.console.name = plain_console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n

appender.json_console.type = Console
appender.json_console.name = json_console
appender.json_console.layout.type = JSONLayout
appender.json_console.layout.compact = true
appender.json_console.layout.eventEol = true

appender.rolling.type = RollingFile
appender.rolling.name = plain_rolling
appender.rolling.fileName = ${sys:ls.logs}/logstash-${sys:ls.log.format}.log
appender.rolling.filePattern = ${sys:ls.logs}/logstash-${sys:ls.log.format}-%d{yyyy-MM-dd}.log
appender.rolling.policies.type = Policies
appender.rolling.policies.time.type = TimeBasedTriggeringPolicy
appender.rolling.policies.time.interval = 1
appender.rolling.policies.time.modulate = true
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %-.10000m%n

需手动加上：

appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.action.type = Delete
appender.rolling.strategy.action.basepath = ${sys:ls.logs}
appender.rolling.strategy.action.condition.type = IfFileName
appender.rolling.strategy.action.condition.glob = ${sys:ls.logs}/logstash-${sys:ls.log.format}
appender.rolling.strategy.action.condition.nested_condition.type = IfLastModified
appender.rolling.strategy.action.condition.nested_condition.age = 7D

kibana

在kibana的配置文件中，只有以下几个选项：

logging.dest:
Default: stdout Enables you specify a file where Kibana stores log output.
logging.quiet:
Default: false Set the value of this setting to true to suppress all logging output other than error messages.
logging.silent:
Default: false Set the value of this setting to true to suppress all logging output.
logging.verbose:
Default: false Set the value of this setting to true to log all events, including system usage information and all requests. Supported on Elastic Cloud Enterprise.
logging.timezone
Default: UTC Set to the canonical timezone id (e.g. US/Pacific) to log events using that timezone. A list of timezones can be referenced at https://en.wikipedia.org/wiki/List_of_tz_database_time_zones.

我们可以指定输出的日志文件与日志内容，但是却不可以配置日志的rotate。这时，我们需要使用logrotate，这个linux默认安装的工具。 首先，我们要在配置文件里面指定生成pid文件：

pid.file: "pid.log"

然后，修改/etc/logrotate.conf:

/var/log/kibana {
    missingok
    notifempty
    sharedscripts
    daily
    rotate 7
    copytruncate
    /bin/kill -HUP $(cat /usr/share/kibana/pid.log 2>/dev/null) 2>/dev/null
    endscript
}

kafka

如果不想写脚本清理过多的文件的话，需要修改config/log4j.properties文件。使用RollingFileAppender代替DailyRollingFileAppender，同时设置MaxFileSize和MaxBackupIndex。即修改为：

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.kafkaAppender=org.apache.log4j.RollingFileAppender
log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log
log4j.appender.kafkaAppender.MaxFileSize=10MB
log4j.appender.kafkaAppender.MaxBackupIndex=10
log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.stateChangeAppender=org.apache.log4j.RollingFileAppender
log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
log4j.appender.stateChangeAppender.MaxFileSize=10M
log4j.appender.stateChangeAppender.MaxBackupIndex=10
log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.requestAppender=org.apache.log4j.RollingFileAppender
log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log
log4j.appender.requestAppender.MaxFileSize=10MB
log4j.appender.requestAppender.MaxBackupIndex=10
log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.cleanerAppender=org.apache.log4j.RollingFileAppender
log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log
log4j.appender.cleanerAppender.MaxFileSize=10MB
log4j.appender.cleanerAppender.MaxBackupIndex=10
log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.controllerAppender=org.apache.log4j.RollingFileAppender
log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log
log4j.appender.controllerAppender.MaxFileSize=10MB
log4j.appender.controllerAppender.MaxBackupIndex=10
log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.authorizerAppender=org.apache.log4j.RollingFileAppender
log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
log4j.appender.authorizerAppender.MaxFileSize=10MB
log4j.appender.authorizerAppender.MaxBackupIndex=10
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

# Turn on all our debugging info
#log4j.logger.kafka.producer.async.DefaultEventHandler=DEBUG, kafkaAppender
#log4j.logger.kafka.client.ClientUtils=DEBUG, kafkaAppender
#log4j.logger.kafka.perf=DEBUG, kafkaAppender
#log4j.logger.kafka.perf.ProducerPerformance$ProducerThread=DEBUG, kafkaAppender
#log4j.logger.org.I0Itec.zkclient.ZkClient=DEBUG
log4j.logger.kafka=INFO, kafkaAppender

log4j.logger.kafka.network.RequestChannel$=WARN, requestAppender
log4j.additivity.kafka.network.RequestChannel$=false

#log4j.logger.kafka.network.Processor=TRACE, requestAppender
#log4j.logger.kafka.server.KafkaApis=TRACE, requestAppender
#log4j.additivity.kafka.server.KafkaApis=false
log4j.logger.kafka.request.logger=WARN, requestAppender
log4j.additivity.kafka.request.logger=false

log4j.logger.kafka.controller=TRACE, controllerAppender
log4j.additivity.kafka.controller=false

log4j.logger.kafka.log.LogCleaner=INFO, cleanerAppender
log4j.additivity.kafka.log.LogCleaner=false

log4j.logger.state.change.logger=TRACE, stateChangeAppender
log4j.additivity.state.change.logger=false

#Change this to debug to get the actual audit log for authorizer.
log4j.logger.kafka.authorizer.logger=WARN, authorizerAppender
log4j.additivity.kafka.authorizer.logger=false

使用过Kibana的同学应该都知道，当我们在kibana的配置文件中打开或者关闭功能，或者安装、卸载额外的插件后，重启kibana会触发一个优化的过程（optimize），如下图：

这个过程或长或短，视你电脑的性能而定。这里简单介绍一下该过程所要完成的事情。

Kibana是一个单页Web应用

首先，Kibana是一个单页的web应用。何为单页web应用？即所有的页面的读取都是在浏览器上完成，而与后台服务器无关。与后台服务器的通信只关乎数据，而非页面。所以，应用上所有的UI都被打包在一起，一次性的发送到了浏览器端，而不是通过URL到后台进行获取。所以，我们看到kibana的首页是下面这样的： http://localhost:5601/app/kibana#/ 注意这里的#后，代表#后面的内容会被浏览器提取，不往服务器端进行url的情况，而是在浏览器上进行内部重新渲染。因为所有的页面都是存储在浏览器的，所有在初次访问的时候，会加载大量的代码到浏览器端，这些代码都是被压缩过的bundle文件：

而optimize的过程，就是把这些原本可读性的源代码压缩为bundle.js的过程。因此，每当你对Kibana进行裁剪之后重启，因为前端的部分是完全由浏览器负责的，所有bundle文件需要重新生成后再发给浏览器，所以会触发optimize的过程。

Kibana在6.2.0版本之后，常规版本已经默认自带了xpack（当然，你还是可以直接下载不带xpack的开源社区版），导致Kibana的size已经到了200M左右，而且越往后的版本，功能越多，代码量越大，每次optimize的过程都会耗费更多的时间。一般来说，我们会将Kibana部署在单独的机器上，因为这仅仅是一个web后端，通常我们不会分配比较优质的资源，（2C4G都算浪费的了），这种情况下面，每次我们裁剪后重启Kibana都会耗费半个小时～1个小时的时间，更有甚者直接hang住，查看系统日志才知道OOM了。

Nodejs的内存机制

Kibana是用Nodejs编写的程序，在一般的后端语言中，基本的内存使用上基本没有什么限制，但是在nodeJs中却只能使用部分内存。在64位系统下位约为1.4G，在32位系统下约为0.7G，造成这个问题的主要原因是因为nodeJs基于V8构建，V8使用自己的方式来管理和分配内存，这一套管理方式在浏览器端使用绰绰有余，但是在nodeJs中这却限制了开发者，在应用中如果碰到了这个限制，就会造成进程退出。

Nodejs内存机制对Kibana优化的影响

因为Kibana的代码体量越来越大，将所有的代码加载到内存之后，再解析语法树，进行bundle的转换所耗费的内存已经接近1.4G的限制了，当你安装更多插件，比如sentinl的时候，系统往往已经无法为继，导致Kibana无法启动

解决方案

这种情况下，我们需要在Kibana启动的时候，指定NodeJs使用更多的内存。这个可以通过设置Node的环境变量办到。

NODE_OPTIONS="--max-old-space-size=4096"

当然，我的建议是直接指定在kibana的启动脚本当中，修改/usr/share/kibana/bin/kibana文件为：

#!/bin/sh
SCRIPT=$0

# SCRIPT may be an arbitrarily deep series of symlinks. Loop until we have the concrete path.
while [ -h "$SCRIPT" ] ; do
  ls=$(ls -ld "$SCRIPT")
  # Drop everything prior to ->
  link=$(expr "$ls" : '.*-> \(.*\)$')
  if expr "$link" : '/.*' > /dev/null; then
    SCRIPT="$link"
  else
    SCRIPT=$(dirname "$SCRIPT")/"$link"
  fi
done

DIR="$(dirname "${SCRIPT}")/.."
NODE="${DIR}/node/bin/node"
test -x "$NODE" || NODE=$(which node)
if [ ! -x "$NODE" ]; then
  echo "unable to find usable node.js executable."
  exit 1
fi

NODE_ENV=production exec "${NODE}" $NODE_OPTIONS --max_old_space_size=3072 --no-warnings "${DIR}/src/cli" ${@}

改动在最后一句：NODE_ENV=production exec "${NODE}" $NODE_OPTIONS --max_old_space_size=3072 --no-warnings "${DIR}/src/cli" ${@}

这样，我们可以保证Kibana能顺利的完成optimize的过程

• 父子关系维护检索实战一 Elasticsearch 5.x 父子关系维护检索实战
• 父子关系维护检索实战二 Elasticsearch 6.x 父子关系维护检索实战

1. Elasticsearch 5.x 中父子关系mapping结构设计
2. Elasticsearch 5.x 中维护父子关系数据
3. Elasticsearch 5.x 中has_child和has_parent查询的基本用法
4. Elasticsearch 5.x 中如何在检索中同时返回父子数据

<properties>

  

    <!--

   创建客户信息索引索引表

   -->

    <property name="createClientIndice">

        <![CDATA[{

            "settings": {

                "number_of_shards": 6,

                "index.refresh_interval": "5s"

            },

            "mappings": {

                "basic": {  ##基本信息

                    "properties": {

                        "party_id": {

                            "type": "keyword"

                        },

                        "sex": {

                            "type": "keyword"

                        },

                        "mari_sts": {

                            "type": "keyword"

                        },

                        "ethnic": {

                            "type": "text"

                        },

                        "prof": {

                            "type": "text"

                        },

                        "province": {

                            "type": "text"

                        },

                        "city": {

                            "type": "text"

                        },

                        "client_type": {

                            "type": "keyword"

                        },

                        "client_name": {

                            "type": "text"

                        },

                        "age": {

                            "type": "integer"

                        },

                        "id_type": {

                            "type": "keyword"

                        },

                        "idno": {

                            "type": "keyword"

                        },

                        "education": {

                            "type": "text"

                        },

                        "created_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "birth_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "last_modified_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "etl_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        }

                    }

                },

                "diagnosis": { ##结果分析

                    "_parent": {

                        "type": "basic"

                    },

                    "properties": {

                        "party_id": {

                            "type": "keyword"

                        },

                        "provider": {

                            "type": "text"

                        },

                        "subject": {

                            "type": "text"

                        },

                        "diagnosis_type": {

                            "type": "text"

                        },

                        "icd10_code": {

                            "type": "text",

                            "type": "keyword"

                        },

                        "sd_disease_name": {

                            "type": "text",

                            "type": "keyword"

                        },

                        "created_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "last_modified_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "etl_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        }

                    }

                },

                "medical": { ##医疗情况

                    "_parent": {

                        "type": "basic"

                    },

                    "properties": {

                        "party_id": {

                            "type": "keyword"

                        },

                        "hos_name_yb": {

                            "type": "text"

                        },

                        "eivisions_name": {

                            "type": "text"

                        },

                        "medical_type": {

                            "type": "text"

                        },

                        "medical_common_name": {

                            "type": "text"

                        },

                        "medical_sale_name": {

                            "type": "text"

                        },

                        "medical_code": {

                            "type": "text"

                        },

                        "specification": {

                            "type": "text"

                        },

                        "usage_num": {

                            "type": "text"

                        },

                        "unit": {

                            "type": "text"

                        },

                        "usage_times": {

                            "type": "text"

                        },

                        "created_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "last_modified_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "etl_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        }

                    }

                },

                "exam": { ##检查结果

                    "_parent": {

                        "type": "basic"

                    },

                    "properties": {

                        "party_id": {

                            "type": "keyword"

                        },

                        "hospital": {

                            "type": "text"

                        },

                        "dept": {

                            "type": "text"

                        },

                        "is_ok": {

                            "type": "text"

                        },

                        "exam_result": {

                            "type": "text"

                        },

                        "fld1": {

                            "type": "text"

                        },

                        "fld2": {

                            "type": "text"

                        },

                        "fld3": {

                            "type": "text"

                        },

                        "fld4": {

                            "type": "text"

                        },

                        "fld5": {

                            "type": "text"

                        },

                        "fld901": {

                            "type": "text"

                        },

                        "fld6": {

                            "type": "text"

                        },

                        "fld902": {

                            "type": "text"

                        },

                        "fld14": {

                            "type": "text"

                        },

                        "fld20": {

                            "type": "text"

                        },

                        "fld21": {

                            "type": "text"

                        },

                        "fld23": {

                            "type": "text"

                        },

                        "fld24": {

                            "type": "text"

                        },

                        "fld65": {

                            "type": "text"

                        },

                        "fld66": {

                            "type": "text"

                        },

                        "fld67": {

                            "type": "text"

                        },

                        "fld68": {

                            "type": "text"

                        },

                        "created_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "last_modified_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        },

                        "etl_date": {

                            "type": "date",

                            "format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd'T'HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||epoch_millis"

                        }

                    }

                }

            }

        }]]>

    </property>

</properties>

	public void createClientIndice(){

		//定义客户端实例，加载上面建立的dsl配置文件

		ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/Client_Info.xml");

		try {

			//client_info存在返回true，不存在返回false

			boolean exist = clientUtil.existIndice("client_info");



			//如果索引表client_info已经存在先删除mapping

			if(exist) {//先删除mapping client_info

				clientUtil.dropIndice("client_info");

			}

		} catch (ElasticSearchException e) {

			// TODO Auto-generated catch block

			e.printStackTrace();

		}

		//创建mapping client_info

		clientUtil.createIndiceMapping("client_info","createClientIndice");

		String client_info = clientUtil.getIndice("client_info");//获取最新建立的索引表结构client_info

		System.out.println("after createClientIndice clientUtil.getIndice(\"client_info\") response:"+client_info);

	}

• 定义对象

• Basic
• Medical
• Exam
• Diagnosis

public class Basic extends ESBaseData {

    /**

     *  索引_id

     */

    @ESId

    private String party_id;

    private String sex;                     // 性别

    ......

}    

通过注解@ESParentId指定Medical关联的基本信息文档_id，Medical文档_id由ElasticSearch自动生成

public class Medical extends ESBaseData {

    @ESParentId

    private String party_id;          //父id

    private String hos_name_yb;         //就诊医院

    ...

}

通过注解@ESParentId指定Exam关联的基本信息文档_id，Exam文档_id由ElasticSearch自动生成

public class Exam extends ESBaseData {

    @ESParentId

    private String party_id;          //父id

    private String  hospital;           // 就诊医院

    ....

}    

通过注解@ESParentId指定Diagnosis关联的基本信息文档_id，Diagnosis文档_id由ElasticSearch自动生成

public class Diagnosis extends ESBaseData {

    @ESParentId

    private String party_id;          //父id

    private String provider;            //诊断医院

    private String subject;             //科室

    ......

}    

• 通过api维护测试数据

	/**

	 * 录入体检医疗信息

	 */

	public void importClientInfoDataFromBeans()  {

		ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();



		//导入基本信息,并且实时刷新，测试需要，实际环境不要带refresh

		List<Basic> basics = buildBasics();

		clientUtil.addDocuments("client_info","basic",basics,"refresh");



		//导入医疗信息,并且实时刷新，测试需要，实际环境不要带refresh

		List<Medical> medicals = buildMedicals();

		clientUtil.addDocuments("client_info","medical",medicals,"refresh");



		//导入体检结果数据,并且实时刷新，测试需要，实际环境不要带refresh

		List<Exam> exams = buildExams();

		clientUtil.addDocuments("client_info","exam",exams,"refresh");



		//导入结果诊断数据,并且实时刷新，测试需要，实际环境不要带refresh

		List<Diagnosis> diagnosiss = buildDiagnosiss();

		clientUtil.addDocuments("client_info","diagnosis",diagnosiss,"refresh");

	}

	//构建基本信息集合

	private List<Basic> buildBasics() {

		List<Basic> basics = new ArrayList<Basic>();

		Basic basic = new Basic();

		basic.setParty_id("1");

		basic.setAge(60);

		basics.add(basic);

		//继续添加其他数据

		return basics;



	}

	//

构建医疗信息集合

	private List<Medical> buildMedicals() {

		List<Medical> medicals = new ArrayList<Medical>();

		Medical medical = new Medical();

		medical.setParty_id("1");//设置父文档id-基本信息文档_id

		medical.setCreated_date(new Date());

		medicals.add(medical);

		//继续添加其他数据

		return medicals;



	}

	//构建体检结果数据集合

	private List<Exam> buildExams() {

		List<Exam> exams = new ArrayList<Exam>();

		Exam exam = new Exam();

		exam.setParty_id("1");//设置父文档id-基本信息文档_id

		exams.add(exam);

		//继续添加其他数据

		return exams;

	}

	//构建结果诊断数据集合

	private List<Diagnosis> buildDiagnosiss() {

		List<Diagnosis> diagnosiss = new ArrayList<Diagnosis>();

		Diagnosis diagnosis = new Diagnosis();

		diagnosis.setParty_id("1");//设置父文档id-基本信息文档_id

		diagnosiss.add(diagnosis);

		//继续添加其他数据

		return diagnosiss;

	}

• 通过json报文批量导入测试数据

    <!--

   导入基本信息：

   -->

    <property name="bulkImportBasicData" trim="false">

        <![CDATA[

            { "index": { "_id": "1" }}

            {  "party_id":"1", "sex":"男", "mari_sts":"不详", "ethnic":"蒙古族", "prof":"放牧","birth_date":"1966-2-14 00:00:00", "province":"内蒙古", "city":"赤峰市","client_type":"1", "client_name":"安", "age":52,"id_type":"1", "idno":"1", "education":"初中","created_date":"2013-04-24 00:00:00","last_modified_date":"2013-04-24 00:00:00", "etl_date":"2013-04-24 00:00:00"}

            { "index": { "_id": "2" }}

            { "party_id":"2", "sex":"女", "mari_sts":"已婚", "ethnic":"汉族", "prof":"公务员","birth_date":"1986-07-06 00:00:00", "province":"广东", "city":"深圳","client_type":"1", "client_name":"彭", "age":32,"id_type":"1", "idno":"2", "education":"本科", "created_date":"2013-05-09 15:49:47","last_modified_date":"2013-05-09 15:49:47", "etl_date":"2013-05-09 15:49:47"}

            { "index": { "_id": "3" }}

            { "party_id":"3", "sex":"男", "mari_sts":"未婚", "ethnic":"汉族", "prof":"无业","birth_date":"2000-08-15 00:00:00", "province":"广东", "city":"佛山","client_type":"1", "client_name":"浩", "age":18,"id_type":"1", "idno":"3", "education":"高中", "created_date":"2014-09-01 09:49:27","last_modified_date":"2014-09-01 09:49:27", "etl_date":"2014-09-01 09:49:27" }

             { "index": { "_id": "4" }}

             { "party_id":"4", "sex":"女", "mari_sts":"未婚", "ethnic":"满族", "prof":"工人","birth_date":"1996-03-14 00:00:00", "province":"江苏", "city":"扬州","client_type":"1", "client_name":"慧", "age":22,"id_type":"1", "idno":"4", "education":"高中", "created_date":"2014-09-16 09:30:37","last_modified_date":"2014-09-16 09:30:37", "etl_date":"2014-09-16 09:30:37" }

            { "index": { "_id": "5" }}

            { "party_id":"5", "sex":"女", "mari_sts":"已婚", "ethnic":"汉族", "prof":"教师","birth_date":"1983-08-14 00:00:00", "province":"宁夏", "city":"灵武","client_type":"1", "client_name":"英", "age":35,"id_type":"1", "idno":"5", "education":"本科", "created_date":"2015-09-16 09:30:37","last_modified_date":"2015-09-16 09:30:37", "etl_date":"2015-09-16 09:30:37" }

            { "index": { "_id": "6" }}

            { "party_id":"6", "sex":"女", "mari_sts":"已婚", "ethnic":"汉族", "prof":"工人","birth_date":"1959-07-04 00:00:00", "province":"山东", "city":"青岛","client_type":"1", "client_name":"岭", "age":59,"id_type":"1", "idno":"6", "education":"小学", "created_date":"2015-09-01 09:49:27","last_modified_date":"2015-09-01 09:49:27", "etl_date":"2015-09-01 09:49:27" }

            { "index": { "_id": "7" }}

            { "party_id":"7", "sex":"女", "mari_sts":"未婚", "ethnic":"汉族", "prof":"学生","birth_date":"1999-02-18 00:00:00", "province":"山东", "city":"青岛","client_type":"1", "client_name":"欣", "age":19,"id_type":"1", "idno":"7", "education":"高中", "created_date":"2016-12-01 09:49:27","last_modified_date":"2016-12-01 09:49:27", "etl_date":"2016-12-01 09:49:27" }

            { "index": { "_id": "8" }}

            { "party_id":"8", "sex":"女", "mari_sts":"未婚", "ethnic":"汉族", "prof":"学生","birth_date":"2007-11-18 00:00:00", "province":"山东", "city":"青岛","client_type":"1", "client_name":"梅", "age":10,"id_type":"1", "idno":"8", "education":"小学", "created_date":"2016-11-21 09:49:27","last_modified_date":"2016-11-21 09:49:27", "etl_date":"2016-11-21 09:49:27" }

            { "index": { "_id": "9" }}

            { "party_id":"9", "sex":"男", "mari_sts":"不详", "ethnic":"回族", "prof":"个体户","birth_date":"1978-03-29 00:00:00", "province":"北京", "city":"北京","client_type":"1", "client_name":"磊", "age":40,"id_type":"1", "idno":"9", "education":"高中", "created_date":"2017-09-01 09:49:27","last_modified_date":"2017-09-01 09:49:27", "etl_date":"2017-09-01 09:49:27" }

            { "index": { "_id": "10" }}

            { "party_id":"10", "sex":"男", "mari_sts":"已婚", "ethnic":"汉族", "prof":"农民","birth_date":"1970-11-14 00:00:00", "province":"浙江", "city":"台州","client_type":"1", "client_name":"强", "age":47,"id_type":"1", "idno":"10", "education":"初中", "created_date":"2018-09-01 09:49:27","last_modified_date":"2018-09-01 09:49:27", "etl_date":"2018-09-01 09:49:27" }

        ]]>

    </property>

    <!--

  导入诊断信息

  -->

    <property name="bulkImportDiagnosisData" trim="false">

        <![CDATA[

            { "index": { "parent": "1" }}

            { "party_id":"1", "provider":"内蒙古医院", "subject":"","diagnosis_type":"","icd10_code":"J31.0", "sd_disease_name":"鼻炎","created_date":"2013-07-23 20:56:44", "last_modified_date":"2013-07-23 20:56:44", "etl_date":"2013-07-23 20:56:44" }



            { "index": { "parent": "1" }}

            { "party_id":"1", "provider":"内蒙古医院", "subject":"","diagnosis_type":"","icd10_code":"M47.8", "sd_disease_name":"颈椎病","created_date":"2013-09-23 20:56:44", "last_modified_date":"2013-09-23 20:56:44", "etl_date":"2013-09-23 20:56:44" }



            { "index": { "parent": "1" }}

            { "party_id":"1", "provider":"内蒙古医院", "subject":"","diagnosis_type":"","icd10_code":"E78.1", "sd_disease_name":"甘油三脂增高","created_date":"2018-09-20 09:27:44", "last_modified_date":"2018-09-20 09:27:44", "etl_date":"2018-09-20 09:27:44" }



            { "index": { "parent": "4" }}

            { "party_id":"4", "provider":"江苏医院", "subject":"","diagnosis_type":"","icd10_code":"J00", "sd_disease_name":"感冒","created_date":"2011-05-19 15:52:55", "last_modified_date":"2011-05-19 15:52:55", "etl_date":"2011-05-19 15:52:55" }



            { "index": { "parent": "6" }}

            { "party_id":"6", "provider":"山东医院", "subject":"","diagnosis_type":"","icd10_code":"H44", "sd_disease_name":"眼疾","created_date":"2016-04-08 10:42:18", "last_modified_date":"2016-04-08 10:42:18", "etl_date":"2016-04-08 10:42:18" }



            { "index": { "parent": "6" }}

            { "party_id":"6", "provider":"山东医院", "subject":"","diagnosis_type":"","icd10_code":"M47.8", "sd_disease_name":"颈椎病","created_date":"2016-04-08 10:42:18", "last_modified_date":"2016-04-08 10:42:18", "etl_date":"2016-04-08 10:42:18" }



            { "index": { "parent": "7" }}

            { "party_id":"7", "provider":"山东医院", "subject":"","diagnosis_type":"","icd10_code":"J00", "sd_disease_name":"感冒","created_date":"2017-04-08 10:42:18", "last_modified_date":"2017-04-08 10:42:18", "etl_date":"2017-04-08 10:42:18" }



            { "index": { "parent": "8" }}

            { "party_id":"8", "provider":"山东医院", "subject":"","diagnosis_type":"","icd10_code":"J00", "sd_disease_name":"感冒","created_date":"2018-04-08 10:42:18", "last_modified_date":"2018-04-08 10:42:18", "etl_date":"2018-04-08 10:42:18" }



            { "index": { "parent": "9" }}

            { "party_id":"9", "provider":"朝阳医院", "subject":"","diagnosis_type":"","icd10_code":"A03.901", "sd_disease_name":"急性细菌性痢疾","created_date":"2015-06-08 10:42:18", "last_modified_date":"2015-06-08 10:42:18", "etl_date":"2015-06-08 10:42:18" }

        ]]>

    </property>



    <!--

 导入医疗信息

 -->

    <property name="bulkImportMedicalData" trim="false">

        <![CDATA[

                { "index": { "parent": "1" }}

                { "party_id":"1", "hos_name_yb":"内蒙古医院", "eivisions_name":"", "medical_type":"","medical_common_name":"氟化钠", "medical_sale_name":"", "medical_code":"A01AA01", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



                { "index": { "parent": "1" }}

                { "party_id":"1", "hos_name_yb":"内蒙古医院", "eivisions_name":"", "medical_type":"","medical_common_name":"四环素", "medical_sale_name":"", "medical_code":"A01AB13", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2016-05-31 00:00:00", "last_modified_date":"2016-05-31 00:00:00", "etl_date":"2016-05-31 00:00:00" }



                { "index": { "parent": "1" }}

                { "party_id":"1", "hos_name_yb":"内蒙古医院", "eivisions_name":"", "medical_type":"","medical_common_name":"", "medical_sale_name":"盐酸多西环素胶丸", "medical_code":"A01AB22", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2016-03-18 00:00:00", "last_modified_date":"2016-03-18 00:00:00", "etl_date":"2016-03-18 00:00:00" }



                { "index": { "parent": "1" }}

                { "party_id":"1", "hos_name_yb":"内蒙古医院", "eivisions_name":"", "medical_type":"","medical_common_name":"盐酸多西环素分散片", "medical_sale_name":"", "medical_code":"A01AB22", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2013-07-23 20:56:44", "last_modified_date":"2013-07-23 20:56:44", "etl_date":"2013-07-23 20:56:44" }



                { "index": { "parent": "1" }}

                { "party_id":"1", "hos_name_yb":"内蒙古医院", "eivisions_name":"", "medical_type":"","medical_common_name":"地塞米松", "medical_sale_name":"", "medical_code":"A01AC02", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2013-09-23 20:56:44", "last_modified_date":"2013-09-23 20:56:44", "etl_date":"2013-09-23 20:56:44" }



                { "index": { "parent": "1" }}

                { "party_id":"1", "hos_name_yb":"内蒙古医院", "eivisions_name":"", "medical_type":"","medical_common_name":"肾上腺素", "medical_sale_name":"", "medical_code":"A01AD01", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2018-09-20 09:27:44", "last_modified_date":"2018-09-20 09:27:44", "etl_date":"2018-09-20 09:27:44" }



                 { "index": { "parent": "4" }}

                { "party_id":"4", "hos_name_yb":"江苏医院", "eivisions_name":"", "medical_type":"","medical_common_name":"地塞米松", "medical_sale_name":"", "medical_code":"A01AC02", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2011-05-19 15:52:55", "last_modified_date":"2011-05-19 15:52:55", "etl_date":"2011-05-19 15:52:55" }



                { "index": { "parent": "4" }}

                { "party_id":"4", "hos_name_yb":"江苏医院", "eivisions_name":"", "medical_type":"","medical_common_name":"四环素", "medical_sale_name":"", "medical_code":"A01AB13", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2018-04-08 10:42:18", "last_modified_date":"2018-04-08 10:42:18", "etl_date":"2018-04-08 10:42:18" }



                { "index": { "parent": "4" }}

                { "party_id":"4", "hos_name_yb":"江苏医院", "eivisions_name":"", "medical_type":"","medical_common_name":"诺氟沙星胶囊", "medical_sale_name":"", "medical_code":"A01AD01", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2015-06-08 10:42:18", "last_modified_date":"2015-06-08 10:42:18", "etl_date":"2015-06-08 10:42:18" }



                { "index": { "parent": "6" }}

                { "party_id":"6", "hos_name_yb":"山东医院", "eivisions_name":"", "medical_type":"","medical_common_name":"盐酸异丙肾上腺素片", "medical_sale_name":"", "medical_code":"A01AD01", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2014-01-23 20:56:44", "last_modified_date":"2014-01-23 20:56:44", "etl_date":"2014-01-23 20:56:44" }



                { "index": { "parent": "6" }}

                { "party_id":"6", "hos_name_yb":"山东医院", "eivisions_name":"", "medical_type":"","medical_common_name":"甲硝唑栓", "medical_sale_name":"", "medical_code":"A01AB17", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2018-06-08 10:42:18", "last_modified_date":"2018-06-08 10:42:18", "etl_date":"2018-06-08 10:42:18" }



                { "index": { "parent": "9" }}

                { "party_id":"9", "hos_name_yb":"朝阳医院", "eivisions_name":"", "medical_type":"","medical_common_name":"复方克霉唑乳膏", "medical_sale_name":"", "medical_code":"A01AB18", "specification":"","usage_num":"", "unit":"", "usage_times":"","created_date":"2014-01-23 20:56:44", "last_modified_date":"2014-01-23 20:56:44", "etl_date":"2014-01-23 20:56:44"}

         ]]>

    </property>



    <!--

     导入体检信息

-->

    <property name="bulkImportExamData" trim="false">

        <![CDATA[

            { "index": { "parent": "1" }}

            { "party_id":"1", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"高血压","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "2" }}

            { "party_id":"2", "hospital":"", "dept":"", "is_ok":"Y", "exam_result":"轻度脂肪肝","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "3" }}

            { "party_id":"3", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"急性细菌性痢疾","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "4" }}

            { "party_id":"4", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"感冒","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "5" }}

            { "party_id":"5", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"感冒","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "6" }}

            { "party_id":"6", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"感冒","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "7" }}

            { "party_id":"7", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"颈椎病","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "8" }}

            { "party_id":"1", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"颈椎病","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



            { "index": { "parent": "9" }}

                { "party_id":"9", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"颈椎病","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }



                { "index": { "parent": "10" }}

                { "party_id":"10", "hospital":"", "dept":"", "is_ok":"N", "exam_result":"颈椎病","fld1":"158", "fld2":"63", "fld3":"94", "fld4":"85", "fld5":"131", "fld901":"89", "fld6":"4.9","fld902":"4.8","fld14":"78", "fld21":"78", "fld23":"", "fld24":"5.5", "fld65":"5.5", "fld66":"1.025","fld67":"", "fld68":"","created_date":"2014-03-18 00:00:00", "last_modified_date":"2014-03-18 00:00:00", "etl_date":"2014-03-18 00:00:00" }

        ]]>

    </property>

	/**

	 * 通过读取配置文件中的dsl json数据导入医疗数据

	 */

	public void importClientInfoFromJsonData(){

		ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/Client_Info.xml");



		clientUtil.executeHttp("client_info/basic/_bulk?refresh","bulkImportBasicData",ClientUtil.HTTP_POST);

		clientUtil.executeHttp("client_info/diagnosis/_bulk?refresh","bulkImportDiagnosisData",ClientUtil.HTTP_POST);

		clientUtil.executeHttp("client_info/medical/_bulk?refresh","bulkImportMedicalData",ClientUtil.HTTP_POST);

		clientUtil.executeHttp("client_info/exam/_bulk?refresh","bulkImportExamData",ClientUtil.HTTP_POST);

		

		long basiccount = clientUtil.countAll("client_info/basic");

		System.out.println(basiccount);

		long medicalcount = clientUtil.countAll("client_info/medical");

		System.out.println(medicalcount);

		long examcount = clientUtil.countAll("client_info/exam");

		System.out.println(examcount);

		long diagnosiscount = clientUtil.countAll("client_info/diagnosis");

		System.out.println(diagnosiscount);

	}

• 根据父查子-通过客户名称信息查询客户端体检结果

   <!--根据客户名称查询客户体检报告-->

    <property name="queryExamSearchByClientName">

        <![CDATA[

            {

              "query": {

                ## 最多返回size变量对应的记录条数

                "size":#[size],

                "has_parent": {

                  "type": "basic",

                  "query": {

                    "match": {

                      "client_name": #[clientName] ## 通过变量clientName设置客户名称

                    }

                  }

                }

              }

            }

        ]]>

    </property>

	/**

	 * 根据客户名称查询客户体检报告

	 */

	public void queryExamSearchByClientName(){

		ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/Client_info.xml");

		Map<String,Object> params = new HashMap<String,Object>();

		params.put("clientName","张三");

		params.put("size",1000);

		ESDatas<Exam> exams = clientUtil.searchList("client_info/exam/_search","queryExamSearchByClientName",params,Exam.class);

		List<Exam> examList = exams.getDatas();//获取符合条件的体检数据

		long totalSize = exams.getTotalSize();//符合条件的总记录数据

	}

• 根据子查父数据-通过医疗信息编码查找客户基本数据

    <!--通过医疗信息编码查找客户基本数据-->

    <property name="queryClientInfoByMedicalName">

        <![CDATA[

            {

              "query": {

                ## 最多返回size变量对应的记录条数

                "size":#[size],

                "has_child": {

                  "type":       "medical",

                  "score_mode": "max",

                  "query": {

                    "match": {

                      "medical_code": #[medicalCode] ## 通过变量medicalCode设置医疗编码

                    }

                  }

                }

              }

            }

        ]]>

    </property>

	/**

	 * 通过医疗信息编码查找客户基本数据

	 */

	public void queryClientInfoByMedicalName(){

		ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/Client_info.xml");

		Map<String,Object> params = new HashMap<String,Object>();

		params.put("medicalCode","A01AA01"); //通过变量medicalCode设置医疗编码

		params.put("size",1000); //最多返回size变量对应的记录条数

		ESDatas<Basic> bascis = clientUtil.searchList("client_info/basic/_search","queryClientInfoByMedicalName",params,Basic.class);

		List<Basic> bascisList = bascis.getDatas();//获取符合条件的客户信息

		long totalSize = bascis.getTotalSize();

	}

• 根据父条件查询所有子数据集合并返回父数据，根据客户名称查询所有体检数据，同时返回客户信息

    <!--根据客户名称获取客户体检诊断数据，并返回客户信息-->

    <property name="queryDiagnosisByClientName">

        <![CDATA[

            {

              "query": {

                ## 最多返回size变量对应的记录条数

                "size":#[size],

                "has_parent": {

                  "type": "basic",

                  "query": {

                    "match": {

                      "client_name": #[clientName] ## 通过变量clientName设置客户名称

                    }

                  },

                  "inner_hits": {}  ## 通过变量inner_hits表示要返回对应的客户信息

                }

              }

            }

        ]]>

    </property>

	/**

	 * 根据客户名称获取客户体检诊断数据，并返回客户数据

	 */

	public void queryDiagnosisByClientName(){



		ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/Client_info.xml");

		Map<String,Object> params = new HashMap<String,Object>();

		params.put("clientName","张三");

		params.put("size",1000);



		try {

			ESInnerHitSerialThreadLocal.setESInnerTypeReferences(Basic.class);//指定inner查询结果对应的客户基本信息类型,Basic只有一个文档类型，索引不需要显示指定basic对应的mapping type名称

			ESDatas<Diagnosis> diagnosiss = clientUtil.searchList("client_info/diagnosis/_search",

					"queryDiagnosisByClientName",params,Diagnosis.class);

			List<Diagnosis> diagnosisList = diagnosiss.getDatas();//获取符合条件的体检报告数据

			long totalSize = diagnosiss.getTotalSize();

			//遍历诊断报告信息，并查看报告对应的客户基本信息

			for(int i = 0;  diagnosisList != null && i < diagnosisList.size(); i ++) {

				Diagnosis diagnosis = diagnosisList.get(i);

				List<Basic> basics = ResultUtil.getInnerHits(diagnosis.getInnerHits(), "basic");

				if(basics != null) {

					System.out.println(basics.size());

				}

			}

		}

		finally{

			ESInnerHitSerialThreadLocal.clean();//清空inner查询结果对应的客户基本信息类型

		}

	}

•  根据子条件查询父数据并返回符合条件的父的子数据集合，查询客户信息，同时返回客户对应的所有体检报告、医疗记录、诊断记录

    <!--查询客户信息，同时返回客户对应的所有体检报告、医疗记录、诊断记录-->

    <property name="queryClientAndAllSons">

        <![CDATA[

        {

          "query": {

            "bool": {

              "should": [

                {

                    "match_all":{}

                }

              ]

              ,"must": [

                {

                  "has_child": {

                    "score_mode": "none",

                    "type": "diagnosis"

                    ,"query": {

                      "bool": {

                        "must": [

                          {

                            "term": {

                              "icd10_code": {

                                "value": "J00"

                              }

                            }

                          }

                        ]

                      }

                    },"inner_hits":{}

                  }

                }

                ]

              ,"should": [

                  {

                  "has_child": {

                    "score_mode": "none",

                    "type": "medical"

                    ,"query": {

                      "match_all": {}



                    },"inner_hits":{}

                  }

                }

              ]

              ,"should": [

                {

                  "has_child": {

                    "type": "exam",

                    "query": {

                      "match_all": {}

                    },"inner_hits":{}

                  }

                }

              ]

            }

          }

        }

        ]]>

    </property>

	/**

	 * 查询客户信息，同时返回客户对应的所有体检报告、医疗记录、诊断记录

	 */

	public void queryClientAndAllSons(){

		ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/Client_Info.xml");

		Map<String,Object> params = null;//没有检索条件，构造一个空的参数对象



		try {

			//设置子文档的类型和对象映射关系

			ESInnerHitSerialThreadLocal.setESInnerTypeReferences("exam",Exam.class);//指定inner查询结果对于exam类型和对应的对象类型Exam

			ESInnerHitSerialThreadLocal.setESInnerTypeReferences("diagnosis",Diagnosis.class);//指定inner查询结果对于diagnosis类型和对应的对象类型Diagnosis

			ESInnerHitSerialThreadLocal.setESInnerTypeReferences("medical",Medical.class);//指定inner查询结果对于medical类型和对应的对象类型Medical

			ESDatas<Basic> escompanys = clientUtil.searchList("client_info/basic/_search",

					"queryClientAndAllSons",params,Basic.class);

			//String response = clientUtil.executeRequest("client_info/basic/_search","queryClientAndAllSons",params);直接获取原始的json报文

//			escompanys = clientUtil.searchAll("client_info",Basic.class);

			long totalSize = escompanys.getTotalSize();

			List<Basic> clientInfos = escompanys.getDatas();//获取符合条件的数据

			//查看公司下面的雇员信息（符合检索条件的雇员信息）

			for (int i = 0; clientInfos != null && i < clientInfos.size(); i++) {

				Basic clientInfo = clientInfos.get(i);

				List<Exam> exams = ResultUtil.getInnerHits(clientInfo.getInnerHits(), "exam");

				if(exams != null)

					System.out.println(exams.size());

				List<Diagnosis> diagnosiss = ResultUtil.getInnerHits(clientInfo.getInnerHits(), "diagnosis");

				if(diagnosiss != null)

					System.out.println(diagnosiss.size());

				List<Medical> medicals = ResultUtil.getInnerHits(clientInfo.getInnerHits(), "medical");

				if(medicals != null)

					System.out.println(medicals.size());



			}

		}

		finally{

			ESInnerHitSerialThreadLocal.clean();//清空inner查询结果对于各种类型信息

		}

	}

	@Test

	public void testMutil(){

		this.createClientIndice();//创建indice client_info

//		this.importClientInfoDataFromBeans(); //通过api添加测试数据

		this.importClientInfoFromJsonData();//导入测试数据

		this.queryExamSearchByClientName(); //根据客户端名称查询提交报告

		this.queryClientInfoByMedicalName();//通过医疗信息编码查找客户基本数据

		this.queryDiagnosisByClientName();//根据客户名称获取客户体检诊断数据，并返回客户数据

		this.queryClientAndAllSons();//查询客户信息，同时返回客户对应的所有体检报告、医疗记录、诊断记录

	}

logstash作为一个数据管道中间件，支持对各种类型数据的采集与转换，并将数据发送到各种类型的存储库，比如实现消费kafka数据并且写入到Elasticsearch, 日志文件同步到对象存储S3等，mysql数据同步到Elasticsearch等。

logstash内部主要包含三个模块：

* input: 从数据源获取数据
* filter: 过滤、转换数据
* output: 输出数据

不同类型的数据都可以通过对应的input-plugin， output-plugin完成数据的输入与输出。如需要消费kafka中的数据并写入到Elasticsearch中，则需要使用logstash的kafka-input-plugin完成数据输入，logstash-output-elasticsearch完成数据输出。如果需要对输入数据进行过滤或者转换，比如根据关键词过滤掉不需要的内容，或者时间字段的格式转换，就需要又filter-plugin完成了。

logstash的input插件目前已经有几十种了，支持大多数比较通用或开源的数据源的输入。但如果公司内部开发的数据库或其它存储类的服务不能和开源产品在接口协议上兼容，比如腾讯自研的消息队列服务CMQ不依赖于其它的开源消息队列产品，所以不能直接使用logstash的logstash-input-kafka或logstash-input-rabbitmq同步CMQ中的数据；腾讯云对象存储服务COS， 在鉴权方式上和AWS的S3存在差异，也不能直接使用logstash-input-s3插件从COS中读取数据，对于这种情况，就需要自己开发logstash的input插件了。

本文以开发logstash的cos input插件为例，介绍如何开发logstash的input插件。

logstash官方提供了有个简单的input plugin example可供参考: https://github.com/logstash-plugins/logstash-input-example/

环境准备

logstash使用jruby开发，首先要配置jruby环境：

1. 安装rvm:

   rvm是一个ruby管理器，可以安装并管理ruby环境，也可以通过命令行切换到不同的ruby版本。

   gpg --keyserver hkp://keys.gnupg.net --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB

   \curl -sSL https://get.rvm.io | bash -s stable

   source /etc/profile.d/rvm.sh

2. 安装jruby

   rvm install jruby

   rvm use jruby

3. 安装包管理工具bundle和测试工具rspec

   gem install bundle
   gem install rspec

从example开始

1. clone logstash-input-example

   git clone https://github.com/logstash-plugins/logstash-input-example.git

2. 将clone出来的logstash-input-example源码copy到logstash-input-cos目录，并删除.git文件夹，目的是以logstash-input-example的源码为参考进行开发，同时把需要改动名称的地方修改一下：

   mv logstash-input-example.gemspec logstash-input-cos.gemspec
   mv lib/logstash/inputs/example.rb lib/logstash/inputs/cos.rb
   mv spec/inputs/example_spec.rb spec/inputs/cos_spec.rb

3. 建立的源码目录结构如图所示：

其中，重要文件的作用说明如下：

• cos.rb: 主文件，在该文件中编写logstash配置文件的读写与源数据获取的代码，需要继承LogStash::Inputs::Base基类
• cos_spec.rb: 单元测试文件，通过rspec可以对cos.rb中的代码进行测试
• logstash-input-cos.gemspec: 类似于maven中的pom.xml文件，配置工程的版本、名称、licene,包依赖等，通过bundle命令可以下载依赖包

配置并下载依赖

因为腾讯云COS服务没有ruby sdk, 因为只能依赖其Java sdk进行开发，首先添加对cos java sdk的依赖。在logstash-input-cos.gemspec中Gem dependencies配置栏中增加以下内容：

# Gem dependencies
  s.requirements << "jar 'com.qcloud:cos_api', '5.4.4'"
  s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
  s.add_runtime_dependency 'logstash-codec-plain'
  s.add_runtime_dependency 'stud', '>= 0.0.22'
  s.add_runtime_dependency 'jar-dependencies'
  s.add_development_dependency 'logstash-devutils', '1.3.6'

相比logstash-input-example.gemspec，增加了对com.qcloud:cos_api包以及jar-dependencies包的依赖,jar-dependencies用于在ruby环境中管理jar包，并且可以跟踪jar包的加载状态。

然后，在logstash-input-cos.gemspec中增加配置：

s.platform = 'java'

这样可以成功下载java依赖包，并且可以在ruby代码中直接调用java代码。

最后，执行以下命令下载依赖：

bundle install

编写代码

logstash-input-cos的代码逻辑其实比较简单，主要是通过执行定时任务，调用cos java sdk中的listObjects方法，获取到指定bucket里的数据，并在每次定时任务执行结束后设置marker保存在本地，再次执行时从marker位置获取数据，以实现数据的增量同步。

jar包的引用

因为要调用cos java sdk中的代码，先引用该jar包：

require 'cos_api-5.4.4.jar'
java_import com.qcloud.cos.COSClient;
java_import com.qcloud.cos.ClientConfig;
java_import com.qcloud.cos.auth.BasicCOSCredentials;
java_import com.qcloud.cos.auth.COSCredentials;
java_import com.qcloud.cos.exception.CosClientException;
java_import com.qcloud.cos.exception.CosServiceException;
java_import com.qcloud.cos.model.COSObjectSummary;
java_import com.qcloud.cos.model.ListObjectsRequest;
java_import com.qcloud.cos.model.ObjectListing;
java_import com.qcloud.cos.region.Region;

读取配置文件

logstash配置文件读取的代码如图所示：

config_name为cos,其它的配置项读取代码按照ruby的代码规范编写，添加类型校验与默认值，就可以从以下配置文件中读取配置项：

input {
    cos {
        "endpoint" => "cos.ap-guangzhou.myqcloud.com"
        "access_key_id" => "*****"
        "access_key_secret" => "****"
        "bucket" => "******"
        "region" => "ap-guangzhou"
        "appId" => "**********"
        "interval" => 60
    }
}

output {
    stdout {
        codec=>rubydebug
    }
}

实现register方法

logstash input插件必须实现另个方法：register 和run

register方法类似于初始化方法，在该方法中可以直接使用从配置文件读取并赋值的变量，完成cos client的初始化，代码如下：

    # 1 初始化用户身份信息(appid, secretId, secretKey)
    cred = com.qcloud.cos.auth.BasicCOSCredentials.new(@access_key_id, @access_key_secret)
    # 2 设置bucket的区域, COS地域的简称请参照 https://www.qcloud.com/document/product/436/6224
    clientConfig = com.qcloud.cos.ClientConfig.new(com.qcloud.cos.region.Region.new(@region))
    # 3 生成cos客户端
    @cosclient = com.qcloud.cos.COSClient.new(cred, clientConfig)
    # bucket名称, 需包含appid
    bucketName = @bucket + "-"+ @appId
    @bucketName = bucketName

    @listObjectsRequest = com.qcloud.cos.model.ListObjectsRequest.new()
    # 设置bucket名称
    @listObjectsRequest.setBucketName(bucketName)
    # prefix表示列出的object的key以prefix开始
    @listObjectsRequest.setPrefix(@prefix)
    # 设置最大遍历出多少个对象, 一次listobject最大支持1000
    @listObjectsRequest.setMaxKeys(1000)
    @listObjectsRequest.setMarker(@markerConfig.getMarker)

示例代码中设置了@cosclient和@listObjectRequest为全局变量， 因为在run方法中会用到这两个变量。

注意在ruby中调用java代码的方式：没有变量描述符；不能直接new Object()，而只能Object.new().

实现run方法

run方法获取数据并将数据流转换成event事件

最简单的run方法为：

def run(queue)
    Stud.interval(@interval) do
      event = LogStash::Event.new("message" => @message, "host" => @host)
      decorate(event)
      queue << event
    end # loop
  end # def run

代码说明：

• 通过Stud ruby模块执行定时任务，interval可自定义，从配置文件中读取
• 生成event, 示例代码生成了一个包含两个字段数据的event
• 调用decorate()方法， 给该event打上tag，如果配置的话
• queue<<event, 将event插入到数据管道中，发送给filter处理

logstash-input-cos的run方法实现为：

def run(queue)
    @current_thread = Thread.current
    Stud.interval(@interval) do
      process(queue)
    end
end

def process(queue)
    @logger.info('Marker from: ' + @markerConfig.getMarker)

    objectListing = @cosclient.listObjects(@listObjectsRequest)
    nextMarker = objectListing.getNextMarker()
    cosObjectSummaries = objectListing.getObjectSummaries()
    cosObjectSummaries.each do |obj|
       # 文件的路径key
       key = obj.getKey()

       if stop?
         @logger.info("stop while attempting to read log file")
         break
       end
       # 根据key获取内容
       getObject(key) { |log|
         # 发送消息
         @codec.decode(log) do |event|
           decorate(event)
           queue << event
         end
       }

       #记录 marker
       @markerConfig.setMarker(key)
       @logger.info('Marker end: ' + @markerConfig.getMarker)
    end
  end

  # 获取下载输入流
 def getObject(key, &block)
    getObjectRequest = com.qcloud.cos.model.GetObjectRequest.new(@bucketName, key)
    cosObject = @cosclient.getObject(getObjectRequest)
    cosObjectInput = cosObject.getObjectContent()
    buffered =BufferedReader.new(InputStreamReader.new(cosObjectInput))
    while (line = buffered.readLine())
      block.call(line)
    end
  end

测试代码

在spec/inputs/cos_spec.rb中增加如下测试代码:

# encoding: utf-8
require "logstash/devutils/rspec/spec_helper"
require "logstash/inputs/cos"

describe LogStash::Inputs::Cos do

  it_behaves_like "an interruptible input plugin" do
    let(:config) { {
        "endpoint" => 'cos.ap-guangzhou.myqcloud.com',
        "access_key_id" => '*',
        "access_key_secret" => '*',
        "bucket" => '*',
         "region" => 'ap-guangzhou',
         "appId" => '*',
        "interval" => 60 } }
  end
end

rspec是一个ruby测试库，通过bundle命令执行rspec：

bundle exec rspec

如果cos.rb中的代码没有语法或运行时错误，则会出现如果信息表明测试成功：

Finished in 0.8022 seconds (files took 3.45 seconds to load)
1 example, 0 failures

构建并测试input-plugin-cos

build

使用gem对input-plugin-cos插件源码进行build:

gem build logstash-input-cos.gemspec

构建完成后会生成一个名为logstash-input-cos-0.0.1-java.gem的文件

test

在logstash的解压目录下，执行一下命令安装logstash-input-cos plugin:

./bin/logstash-plugin install /usr/local/githome/logstash-input-cos/logstash-input-cos-0.0.1-java.gem

执行结果为：

Validating /usr/local/githome/logstash-input-cos/logstash-input-cos-0.0.1-java.gem
Installing logstash-input-cos
Installation successful

另外，可以通过./bin/logstash-plugin list命令查看logstash已经安装的所有input/output/filter/codec插件。

生成配置文件cos.logstash.conf,内容为：

input {
    cos {
        "endpoint" => "cos.ap-guangzhou.myqcloud.com"
        "access_key_id" => "*****"
        "access_key_secret" => "****"
        "bucket" => "******"
        "region" => "ap-guangzhou"
        "appId" => "**********"
        "interval" => 60
    }
}

output {
    stdout {
        codec=>rubydebug
    }
}

该配置文件使用腾讯云官网账号的secret_id和secret_key进行权限验证，拉取指定bucket里的数据，为了测试，将output设置为标准输出。

执行logstash:

./bin/logstash -f cos.logstash.conf

输出结果为：

Sending Logstash's logs to /root/logstash-5.6.4/logs which is now configured via log4j2.properties
[2018-07-30T19:26:17,039][WARN ][logstash.runner          ] --config.debug was specified, but log.level was not set to 'debug'! No config info will be logged.
[2018-07-30T19:26:17,048][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/root/logstash-5.6.4/modules/netflow/configuration"}
[2018-07-30T19:26:17,049][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/root/logstash-5.6.4/modules/fb_apache/configuration"}
[2018-07-30T19:26:17,252][INFO ][logstash.inputs.cos      ] Using version 0.1.x input plugin 'cos'. This plugin isn't well supported by the community and likely has no maintainer.
[2018-07-30T19:26:17,341][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500}
[2018-07-30T19:26:17,362][INFO ][logstash.inputs.cos      ] Registering cos input {:bucket=>"bellengao", :region=>"ap-guangzhou"}
[2018-07-30T19:26:17,528][INFO ][logstash.pipeline        ] Pipeline main started
[2018-07-30T19:26:17,530][INFO ][logstash.inputs.cos      ] Marker from:
log4j:WARN No appenders could be found for logger (org.apache.http.client.protocol.RequestAddCookies).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
[2018-07-30T19:26:17,574][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2018-07-30T19:26:17,714][INFO ][logstash.inputs.cos      ] Marker end: access.log
{
       "message" => "77.179.66.156 - - [25/Oct/2016:14:49:33 +0200] \"GET / HTTP/1.1\" 200 612 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36\"",
      "@version" => "1",
    "@timestamp" => 2018-07-30T11:26:17.710Z
}
{
       "message" => "77.179.66.156 - - [25/Oct/2016:14:49:34 +0200] \"GET /favicon.ico HTTP/1.1\" 404 571 \"http://localhost:8080/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36\"",
      "@version" => "1",
    "@timestamp" => 2018-07-30T11:26:17.711Z
}

在cos中的bucket里上传了名为access.log的nginx日志，上述输出结果中最后打印出来的每个json结构体构成一个event， 其中message消息即为access.log中每一条日志。