[大数据]ElasticSearch——进阶（二）

ElasticSearch 集成

Spring Data 集成

Spring Data是一个用于简化数据库、非关系型数据库、索引库访问，并支持云服务的开源框架。其主要目标是使得对数据的访问变得方便快捷，并支持map-reduce框架和云计算数据服务。 Spring Data可以极大的简化JPA（Elasticsearch…）的写法，可以在几乎不用写实现的情况下，实现对数据的访问和操作。除了CRUD外，还包括如分页、排序等一些常用的功能。类似于mybatisPlus。

Spring Data的官网：https://spring.io/projects/spring-data

Spring Data常见功能模块有 Spring Data JDBC、Spring Data Redis、Spring Data ElasticSearch、Spring for Apache Hadoop

Spring Data ElasticSearch 介绍

Spring Data Elasticsearch 基于 spring data API 简化 Elasticsearch操作，将原始操作Elasticsearch的客户端API 进行封装 。Spring Data为Elasticsearch项目提供集成搜索引擎。Spring Data Elasticsearch POJO的关键功能区域为中心的模型与Elastichsearch交互文档和轻松地编写一个存储索引库数据访问层。

官方网站: https://spring.io/projects/spring-data-elasticsearch

集成步骤

1、创建maven项目
springData基于springBoot开发，因此，直接创建一个springboot项目，引入其余依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

2、修改 application.properties 配置文件

# es服务地址
elasticsearch.host=127.0.0.1
# es服务端口
elasticsearch.port=9200

3、数据实体类 Product

@Data
@ToString
@Document(indexName = "product", shards = 3, replicas = 1) // 主分片3个，一个副本
public class Product {
    // 必须有id,这里的id是全局唯一的标识，等同于es中的"_id"
    @Id
    private Long id;// 商品唯一标识
    /**
     * type : 字段数据类型
     * analyzer : 分词器类型
     * index : 是否索引(默认:true)
     * Keyword : 短语,不进行分词
     */
    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    private String title;// 商品名称
    @Field(type = FieldType.Keyword)
    private String category;// 分类名称
    @Field(type = FieldType.Double)
    private Double price;// 商品价格
    @Field(type = FieldType.Keyword, index = false)
    private String images;// 图片地址
}

4、配置类

@ConfigurationProperties(prefix = "elasticsearch")
@Configuration
@Data
public class ElasticsearchConfig extends AbstractElasticsearchConfiguration {
    private String host ;
    private Integer port ;
 
    //重写父类方法
    @Override
    public RestHighLevelClient elasticsearchClient() {
        RestClientBuilder builder = RestClient.builder(new HttpHost(host, port));
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder);
        return restHighLevelClient;
    }
}

5、Dao 数据访问对象

    @Repository
    public interface ProductDao extends ElasticsearchRepository<Product,Long> {
     
    }

6、测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringDataESIndexTest {
    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;
 
    // 创建索引并增加映射配置
    @Test
    public void createIndex(){
        // 创建索引，系统初始化会自动创建索引
        System.out.println("创建索引");
    }
 
    @Test
    public void deleteIndex(){
        // 创建索引，系统初始化会自动创建索引
        boolean flg = elasticsearchRestTemplate.deleteIndex(Product.class);
        System.out.println("删除索引 = " + flg);
    }
}

这里，只演示了简单的索引操作，文档操作不在详细记录，因为使用都比较简单，如下

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringDataESProductDaoTest {
    @Autowired
    private ProductDao productDao;
 
    /**
     * 新增
     */
    @Test
    public void save(){
        Product product = new Product();
        product.setId(2L);
        product.setTitle("华为手机");
        product.setCategory("手机");
        product.setPrice(2999.0);
        product.setImages("http://www.atguigu/hw.jpg");
        productDao.save(product);
    }
    //productDao.save(product); // 修改
    //Product product = productDao.findById(1L).get(); // id查询
    //Iterable<Product> products = productDao.findAll(); // 查询所有
    //productDao.delete(product); // 删除
    //productDao.saveAll(productList); // 批量新增
    
    //PageRequest pageRequest = PageRequest.of(currentPage, pageSize,sort);  //设置查询分页
    //Page<Product> productPage = productDao.findAll(pageRequest); //分页查询
}

Spark Streaming 框架、Flink 框架集成

Spark Streaming和Apace Spark框架都是大数据的数据分析引擎，Flink更倾向实时数据的处理。

集成步骤

1、创建maven项目
2、导入相关jar
3、编写测试类
4、测试
具体过程就不记录，没接触过大数据，有机会在深入学习吧 -.-

ElasticSearch优化

硬件选择

ElasticSearch的基础是Lucene，所有的索引和文档数据是存储在本地的磁盘中，具体的路径可在 ES 的配置文件 …/config/elasticsearch.yml 中配置，磁盘在现代服务器上通常都是瓶颈。ElasticSearch 重度使用磁盘，磁盘能够处理的吞吐量越大，你的节点就越文档。优化磁盘 I/O 的技巧如下：

1、使用SSD（固态硬盘），它比普通的机械硬盘好很多
2、使用 RAID 0。条带化 RAID 会提高磁盘的 I/O
3、使用多块硬盘，并允许 ElasticSearch 通过多个 path.data 目录配置把数据条带化分配到它们上面
4、不要使用远程挂载的存储，比如 NFS 或者 SMB/CIFS

分片策略

分片和副本的设计为 ES 提供了支持分布式和故障转移的特性，但并不意味着分片和副本是可以无限分配的。而且索引的分片完成分配后由于索引的路由机制，我们是不能重新修改分片数的。

可能有人会说，我不知道这个索引将来会变得多大，并且过后我也不能更改索引的大小，所以为了保险起见，还是给它设为 1000 个分片吧。但是需要知道的是，一个分片并不是没有代价的。需要了解：

• 一个分片的底层即为一个 Lucene 索引，会消耗一定文件句柄、内存、以及 CPU 运转。
• 每一个搜索请求都需要命中索引中的每一个分片，如果每一个分片都处于不同的节点还好， 但如果多个分片都需要在同一个节点上竞争使用相同的资源就有些糟糕了。
• 用于计算相关度的词项统计信息是基于分片的。如果有许多分片，每一个都只有很少的数据会导致很低的相关度。

一个业务索引具体需要分配多少分片可能需要架构师和技术人员对业务的增长有个预先的判断，横向扩展应当分阶段进行。为下一阶段准备好足够的资源。 只有当你进入到下一个阶段，你才有时间思考需要作出哪些改变来达到这个阶段。一般来说，我们遵循一些原则：

• 控制每个分片占用的硬盘容量不超过 ES 的最大 JVM 的堆空间设置（一般设置不超过 32G，参考下文的 JVM 设置原则），因此，如果索引的总容量在 500G 左右，那分片大小在 16 个左右即可；当然，最好同时考虑原则 2。
• 考虑一下 node 数量，一般一个节点有时候就是一台物理机，如果分片数过多，大大超过了节点数，很可能会导致一个节点上存在多个分片，一旦该节点故障，即使保持了 1 个以上的副本，同样有可能会导致数据丢失，集群无法恢复。所以， 一般都设置分片数不超过节点数的 3 倍。
  -主分片，副本和节点最大数之间数量，我们分配的时候可以参考以下关系：节点数<=主分片数*（副本数+1）

推迟分片分配

对于节点瞬时中断的问题，默认情况，集群会等待一分钟来查看节点是否会重新加入，如果这个节点在此期间重新加入，重新加入的节点会保持其现有的分片数据，不会触发新的分片分配。这样就可以减少 ES 在自动再平衡可用分片时所带来的极大开销。

通过修改参数 delayed_timeout ，可以延长再均衡的时间，可以全局设置也可以在索引级别进行修改:

PUT /_all/_settings 
{
 "settings": {
 "index.unassigned.node_left.delayed_timeout": "5m" 
  } 
}

路由选择

当我们查询文档的时候，Elasticsearch 如何知道一个文档应该存放到哪个分片中呢？它其实是通过下面这个公式来计算出来:

shard = hash(routing) % number_of_primary_shards
routing 默认值是文档的 id，也可以采用自定义值，比如用户 id

不带 routing 查询

在查询的时候因为不知道要查询的数据具体在哪个分片上，所以整个过程分为 2 个步骤：

分发：请求到达协调节点后，协调节点将查询请求分发到每个分片上。
聚合: 协调节点搜集到每个分片上查询结果，在将查询的结果进行排序，之后给用户返回结果。

带 routing 查询
查询的时候，可以直接根据 routing 信息定位到某个分配查询，不需要查询所有的分配，经过协调节点排序。像上面自定义的用户查询，如果 routing 设置为 userid 的话，就可以直接查询出数据来，效率提升很多。

写入速度

ES 的默认配置，是综合了数据可靠性、写入速度、搜索实时性等因素。实际使用时，我们需要根据公司要求，进行偏向性的优化。
针对于搜索性能要求不高，但是对写入要求较高的场景，我们需要尽可能的选择恰当写优化策略。综合来说，可以考虑以下几个方面来提升写索引的性能：

• 加大 Translog Flush ，目的是降低 Iops、Writeblock。增加 Index Refresh 间隔，目的是减少 Segment Merge 的次数。
• 调整 Bulk 线程池和队列。
• 优化节点间的任务分布。
• 优化 Lucene 层的索引建立，目的是降低 CPU 及 IO。

内存设置

ES 默认安装后设置的内存是 1GB，对于任何一个现实业务来说，这个设置都太小了。如果是通过解压安装的 ES，则在 ES 安装文件中包含一个 jvm.option 文件，添加如下命令来设置 ES 的堆大小，Xms 表示堆的初始大小，Xmx 表示可分配的最大内存，都是 1GB。

确保 Xmx 和 Xms 的大小是相同的，其目的是为了能够在 Java 垃圾回收机制清理完堆区后不需要重新分隔计算堆区的大小而浪费资源，可以减轻伸缩堆大小带来的压力。

假设你有一个 64G 内存的机器，按照正常思维思考，你可能会认为把 64G 内存都给ES 比较好，但现实是这样吗， 越大越好？虽然内存对 ES 来说是非常重要的，但是答案是否定的！
因为 ES 堆内存的分配需要满足以下两个原则：

1、不要超过物理内存的 50%：Lucene 的设计目的是把底层 OS 里的数据缓存到内存中。Lucene 的段是分别存储到单个文件中的，这些文件都是不会变化的，所以很利于缓存，同时操作系统也会把这些段文件缓存起来，以便更快的访问。如果我们设置的堆内存过大，Lucene 可用的内存将会减少，就会严重影响降低 Lucene 的全文本查询性能。
2、堆内存的大小最好不要超过 32GB：在 Java 中，所有对象都分配在堆上，然后有一个 Klass Pointer 指针指向它的类元数据。这个指针在 64 位的操作系统上为 64 位，64 位的操作系统可以使用更多的内存（264）。在 32 位的系统上为 32 位，32 位的操作系统的最大寻址空间为 4GB（232）。但是 64 位的指针意味着更大的浪费，因为你的指针本身大了。浪费内存不算，更糟糕的是，更大的指针在主内存和缓存器（例如 LLC, L1 等）之间移动数据的时候，会占用更多的带宽。

假设你有个机器有 128 GB 的内存，你可以创建两个节点，每个节点内存分配不超过 32 GB。 也就是说不超过 64 GB 内存给 ES 的堆内存，剩下的超过 64 GB 的内存给 Lucene

重要配置

相关面试

都是文字上的记录，需要的时候在来详细学习下吧面试题

end…

**如果总结的还行，就点个赞呗 @_@ 如有错误，欢迎指点！