linbin524 阅读(22) 评论(0)

在线文档:#/README
在线演示地址:#/login
源码github:https://github.com/linbin524/yc.boilerplate
源码gitee:

视频教程:
元磁之力框架开源初心和框架设计介绍(上):
元磁之力框架开源初心和框架设计介绍(下):
元磁之力框架数据库表和代码生成使用教程实战:

QQ群:1060819005

后续:关于框架demo和细节技巧,会在QQ群中发布,就不撰文说明。

大数据套件 ElasticSearch

简介

为了提升YC.Boierlate 在大数据量的处理能力,引入ES组件,封装对应的模块、实现租户拆分、仓储、集群、大数据上亿级别以上数据的检索、统计、分析,并提供千万级别分词搜索等演示示例。

ES基础介绍

Elasticsearch 是一个分布式、RESTful 风格的搜索和数据分析引擎,是PB级别大数据解决方案组件之一。
Elasticsearch是基于Lucense的搜索服务器,,基于RESTful web接口。Elasticsearch是Java语言开发的,并作为Apache许可条款下的开放源码发布,是一种流行的企业级搜索引擎。Elasticsearch用于云计算中,能够达到实时搜索,稳定,可靠,快速,安装使用方便。

ES解决什么问题

对海量数据进行近实时的处理

ES自动可以将海量数据分散到多台服务器上去存储和检索,通过内置搜索引擎、分词、实现
千万级别数据秒级查询、统计、分析等,相对传统关系型数据库的模糊查询在速度有着质的飞跃。

ES 适用场景

  • 维基百科,类似百度百科,牙膏,牙膏的维基百科,全文检索,高亮,搜索推荐

  • The Guardian(国外新闻网站),类似搜狐新闻,用户行为日志(点击,浏览,收藏,评论)+社交网络数据(对某某新闻的相关看法),数据分析,给到每篇新闻文章的作者,让他知道他的文章的公众反馈(好,坏,热门,垃圾,鄙视,崇拜)

  • Stack Overflow(国外的程序异常讨论论坛),IT问题,程序的报错,提交上去,有人会跟你讨论和回答,全文检索,搜索相关问题和答案,程序报错了,就会将报错信息粘贴到里面去,搜索有没有对应的答案

  • GitHub(开源代码管理),搜索上千亿行代码

  • 电商网站,检索商品

  • 日志数据分析,logstash采集日志,ES进行复杂的数据分析(ELK技术,elasticsearch+logstash+kibana)

  • 商品价格监控网站,用户设定某商品的价格阈值,当低于该阈值的时候,发送通知消息给用户,比如说订阅牙膏的监控,如果高露洁牙膏的家庭套装低于50块钱,就通知我,我就去买

  • BI系统,商业智能,Business Intelligence。比如说有个大型商场集团,BI,分析一下某某区域最近3年的用户消费金额的趋势以及用户群体的组成构成,产出相关的数张报表,**区,最近3年,每年消费金额呈现100%的增长,而且用户群体85%是高级白领,开一个新商场。ES执行数据分析和挖掘,Kibana进行数据可视化国内

  • 国内:站内搜索(电商,招聘,门户,等等),IT系统搜索(OA,CRM,ERP,等等),数据分析(ES热门的一个使用场景)

ES 常用组合

  • ELK :Elasticsearch是与名为Logstash的数据收集和日志解析引擎以及名为Kibana的分析和可视化平台一起开发的。这三个产品被设计成一个集成解决方案。
    Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。Elasticsearch是分布式的,这意味着索引可以被分成分片,每个分片可以有0个或多个副本。每个节点托管一个或多个分片,并充当协调器将操作委托给正确的分片。再平衡和路由是自动完成的。相关数据通常存储在同一个索引中,该索引由一个或多个主分片和零个或多个复制分片组成。一旦创建了索引,就不能更改主分片的数量。

  • 阿里巴巴开发的canal:基于Mysql的binlog日志订阅:binlog日志是Mysql用来记录数据实时的变化。这里主要的是binlog同步组件,目前实现的有国内的。
    github地址:https://github.com/alibaba/canal

  • go-mysql-elasticsearch:go-mysql-elasticsearch是一款使用go语言开发的同步数据到ES的工具。 go-mysql-elasticsearch也是基于Mysql的binlog订阅,也可以使用使用mysqldump的方式。目前还不支持ES6.x及以上的版本,也不支持mysql8.x版本,同时该项目目前还不够稳定,也在开发中。
    项目github地址:https://github.com/siddontang/go-mysql-elasticsearch

ES 和常规关系型数据库差异

ES中有几个基本概念:索引(index)、类型(type)、文档(document)、映射(mapping)等。我们将这几个概念与传统的关系型数据库中的库、表、行、列等概念进行对比,如下表:

常规问题

  • 内存:es 的默认配置在常规服务器上大部分都有内存使用率的问题,需要根据实际情况合理调优。

  • 版本:ES 每个版本配套组件有极强耦合,无法做到各个版本兼容,所以jdk、以及其他组件需要指定适配。

  • 分词:es除了内置standard分词,还可以其他分词组件,对中文支持比较好的有:es-ik。

  • 分片(shard): 因为 ES 是个分布式的搜索引擎, 所以索引通常都会分解成不同部分, 而这些分布在不同节点的数据就是分片. ES自动管理和组织分片, 并在必要的时候对分片数据进行再平衡分配, 所以用户基本上不用担心分片的处理细节.

  • 副本(replica): ES 默认为一个索引创建 5 个主分片, 并分别为其创建一个副本分片. 也就是说每个索引都由 5 个主分片成本, 而每个主分片都相应的有一个 copy。对于分布式搜索引擎来说, 分片及副本的分配将是高可用及快速搜索响应的设计核心.主分片与副本都能处理查询请求,它们的唯一区别在于只有主分片才能处理索引请求.副本对搜索性能非常重要,同时用户也可在任何时候添加或删除副本。额外的副本能给带来更大的容量, 更高的呑吐能力及更强的故障恢复能力。

  • 深度查询:在Elasticsearch中如果需要做分页查询,我们通常使用form和size实现。form指定从有序哪一行开始,size表示从当前开始读取多少行。但是我们发现查询结果最大只能到10000,这是因为Elasticsearch中的size的默认值在index.max_result_window 中设置,并且默认值就是10000,如果需要扩展,可以通过如下操作【扩大查询最大值】其中1000000是标识扩大为10万:

   put /tenant_1_books/_settings
    { 
        "index.max_result_window" :"1000000"
    }

还可以采用searchAfer、scroll等方案。

YC.ElasticSearch 模块实战

集群部署

在本地或者服务器上搭建3个es节点,形成集群,针对elasticsearch.yml 进行节点配置,最后启动服务,并安装对应的kibana组件【可视化】。

配置

在项目的YC.ServiceWebApi 中的配置文件 DefaultConfig.json,做如下配置,其中node是对应的es节点。


  "ElasticSearchSetting": {

    ///elasticSearch节点集群
    "Nodes": [
      { "node": "http://127.0.0.1:9200" },
      { "node": "http://127.0.0.1:9201" },
      { "node": "http://127.0.0.1:9202" }

    ]
  }

在项目的YC.ServiceWebApi 找到 ElasticSearchAutofacModule.cs 该文件是相关的IOC 注入配置,在Startup.cs中进行如下注入操作:

 // elasticSearch 注入
builder.RegisterModule(new ElasticSearchAutofacModule());

ES 模块调用

在示例演示 BookAppService 中可以直接使用对应的注入调用es组件。


      private IElasticSearchRepository<Book> _elasticSearchRepository;
       public BookAppService(
        IHttpContextAccessor httpContextAccessor, ICacheManager cacheManager, IMapper mapper, IElasticSearchRepository<Book> elasticSearchRepository) : base(httpContextAccessor, cacheManager)
        {
            _cacheManager = cacheManager;
            _mapper = mapper;
            _elasticSearchRepository = elasticSearchRepository;
        }

        /// <summary>
        /// 查查默认1页10条
        /// </summary>
        /// <returns>返回数据集合</returns>

        public async Task<ApiResult<List<BookAddOrEditDto>>> GetAllAsync()
        {
            var res = new ApiResult<List<BookAddOrEditDto>>();
            var data = await _elasticSearchRepository.GetAllAsync();

            var entityDtoList = _mapper.Map<List<BookAddOrEditDto>>(data);
            return res.Ok(entityDtoList);
        }

YC.ElasticSearch 模块介绍

模块包含有请求上下文、以及默认仓储,其中仓储封装了常规crud、聚合查询、searchAfter查询等常规操作异步方法,并在仓储上提供一个公开请求上下文对象,用于自定义化es操作,模块配套对应的单元测试,提供基础调用示例。

es 其他使用介绍

  • 分词

使用 kibana 操作,对指定的Index进行分词

//创建表 对应的分片,需要表还没创建时候设置
put /tenant_1_books_0
{
  "settings":{
    
    "number_of_shards":2
  }
}

  • 数据结构修改和迁移
 //创建新的索引数据库,并指定字段映射类型,tenant_1_books_0 中的bookName 类型改为keyword
PUT tenant_1_books_0
 {
  "mappings": {
    "properties": {
      
      "bookName":{
        "type":"keyword"
      }
    }
    
  }
  
 }

  //迁移数据,将tenant_1_books 数据迁移到tenant_1_books_1
 POST _reindex
 {
  "source": {
    "index": "tenant_1_books"
    
  },
  "dest": {
     "index": "tenant_1_books_1"
    
  }
 }
  • 分页查询
//查看tenant_1_books所有数据【默认会分页】
GET tenant_1_books/_search
{
  "track_total_hits": true,
  "query": {
    "match_all": {}
  }
}
//深度分页方案1 扩大分页限制
//允许深度分页,限制在10w
put /tenant_1_books/_settings
{
     "index.max_result_window" :"1000000"
}
//查询数据 分页,track_total_hits=真实的总数
GET tenant_1_books/_search
{
  "track_total_hits": true,
  "from" : 99000, "size" : 100,
  "query": {
    "match_all": {}
  }
}

结果如下:

  • 深度查询 searchAfer

配套的net单元测试代码如下:

        /// <summary>
        /// 深度分页查询 searchAfter
        /// </summary>
        /// <returns></returns>
        [Fact]
        public async Task GetPageByQuerySearchAfterTest()
        {
            int size = 100;
            // "bookName" : {
            //"type" : "keyword"
            //},
            //1、BookName 修改为keyword 所有必须完整匹配,不分词
            Func<QueryContainerDescriptor<Book>, QueryContainer> query1 = q => q.Term(t => t.BookName, "吞噬星空");
            Func<QueryContainerDescriptor<Book>, QueryContainer> query2 = q => q.Match(mq =>
              mq.Field(f => f.BookName).Query("哈利波特").Operator(Operator.And)
              );//由于类型为 keyword,所以Match 查找不出来,只能使用Term 精确查询

            //2.全字匹配+ 分词查询
            Func<QueryContainerDescriptor<Book>, QueryContainer> query3 = q => q
                           .Term(t => t.BookName, "吞噬星空")
                           || q.Match(mq =>
               mq.Field(f => f.BookContent).Query("哈利波特").Operator(Operator.And)
              );
            //排序,按照时间升序,再按照
            Func<SortDescriptor<Book>, IPromise<IList<ISort>>> sort = s => s.Ascending(a => a.CreateDate).Descending(d=>d.Price);

          var result1=await  _elasticSearchRepository.GetPageByQuerySearchAfterAsync(query3, sort, 100, null);
            //使用上一次查询得到SearchAfter 作为下一次查询的游标
            var result2 = await _elasticSearchRepository.GetPageByQuerySearchAfterAsync(query3, sort, 100, result1.SearchAfter);

            Assert.NotNull(result2.List);
        }
  • scroll 查询
//深度分页方案3 scroll,者每次查询大量的文档,但是对实时性要求并不高,
//后面的每次滚屏(或者叫翻页)都是基于这个快照的结果,也就是即使有新的数据进来也不会别查询到。
//1. 查询
POST tenant_1_books/_search?scroll=1m
{
    "size": 1000,
    "query": {
        "match_all" : {
        }
    }
   
  }
//2. 上一次查询所得到结果,作为游标
POST _search/scroll
{
    "scroll" : "1m",
    "scroll_id" : "FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFmJpVWRZVWQ3UkItejk2UUx5bC15bFEAAAAAAAMv2xZyZURRWkowelF6S0NnRjMzWjhfQTh3"
  
}
  • 聚合查询
//聚合获取该字段的所有统计
get /tenant_1_books/_search
{
   "aggs":{
      "extended_stats_price":{"extended_stats":{"field":"price"}}
   }
}
//聚合 总和统计
get /tenant_1_books/_search
{
   "aggs":{
      "total_price":{"sum":{"field":"price"}}
   }
}

YC.ElasticSearch 大数据检索示例

演示站点中,默认使用租户1 作为es 检索演示,内置1000多万条测试数据,通过 书名、书内容关键词、发布时间范围等可进行查询, 价格 查询在演示站点中关闭了,无法查询,请注意。

备注:演示站点默认使用10000条数据查询上限边界。