Elasticsearch SQL ORM查询案例

bboss

bboss ES SQL是针对es jdbc的替代解决方案

the best elasticsearch highlevel java rest api——-bboss

bboss 提供一组sql和fetchQuery API，可替代官方es jdbc模块；采用bboss即可拥有bboss的客户端自动发现和容灾能力、对es、jdk、spring boot的兼容性能力，又可以拥有es jdbc的所有功能，同时还解决了因为引入es jdbc导致项目对es版本的强依赖和兼容性问题，参考demo：

orm查询
https://gitee.com/bboss/eshelloword-booter/blob/master/src/test/java/org/bboss/elasticsearchtest/sql/SQLOrmTest.java

分页查询
https://gitee.com/bboss/eshelloword-booter/blob/master/src/test/java/org/bboss/elasticsearchtest/sql/SQLPagineTest.java

目前官方es sql提供的功能有限，也可以在bboss中使用Elasticsearch-sql插件提供的功能，下面有专门的章节介绍。

本文详细说明上面的案例：

1 orm查询

1.1 定义orm查询的实体bean

  1. package org.bboss.elasticsearchtest.sql;
  3. import com.frameworkset.orm.annotation.Column;
  5. import java.util.Date;
  7. public class DocObject {
  8.     private int isnew;
  9.     private Date createtime;
  10.     private String content;
  11.     private int documentId;
  12.     private int channelId;
  14.     /**
  15.      * 通过column指定索引文档和对象属性的映射关系
  16.      * 通过column注解还可以指定日期格式和时区信息
  17.      * @Column(name="docInfo.author",dataformat = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'",timezone = "Etc/UTC",locale = "zh")
  18.      *
  19.      */
  20.     @Column(name="docInfo.author")
  21.     private String docInfoAuthor;
  23.     public int getIsnew() {
  24.         return isnew;
  25.     }
  27.     public void setIsnew(int isnew) {
  28.         this.isnew = isnew;
  29.     }
  31.     public Date getCreatetime() {
  32.         return createtime;
  33.     }
  35.     public void setCreatetime(Date createtime) {
  36.         this.createtime = createtime;
  37.     }
  39.     public String getContent() {
  40.         return content;
  41.     }
  43.     public void setContent(String content) {
  44.         this.content = content;
  45.     }
  47.     public int getDocumentId() {
  48.         return documentId;
  49.     }
  51.     public void setDocumentId(int documentId) {
  52.         this.documentId = documentId;
  53.     }
  55.     public int getChannelId() {
  56.         return channelId;
  57.     }
  59.     public void setChannelId(int channelId) {
  60.         this.channelId = channelId;
  61.     }
  63.     public String getDocInfoAuthor() {
  64.         return docInfoAuthor;
  65.     }
  67.     public void setDocInfoAuthor(String docInfoAuthor) {
  68.         this.docInfoAuthor = docInfoAuthor;
  69.     }
  70. }

实体定义说明：

通过column指定索引文档和对象属性的映射关系，指定日期格式和时区信息,示例如下：

  1. @Column(name="docInfo.author")
  2. private String docInfoAuthor;

指定属性的映射关系、日期格式和时区信息,示例如下：

@Column(name=”docInfo.author”,dataformat = “yyyy-MM-dd’T’HH:mm:ss.SSS’Z’”,timezone = “Etc/UTC”,locale = “zh”)

1.2 执行orm查询

以rest sql api为例来介绍es 6.3.0的sql orm查询功能

  1. package org.bboss.elasticsearchtest.sql;
  3. import org.frameworkset.elasticsearch.ElasticSearchHelper;
  4. import org.frameworkset.elasticsearch.client.ClientInterface;
  5. import org.frameworkset.elasticsearch.entity.sql.SQLRestResponse;
  6. import org.frameworkset.elasticsearch.entity.sql.SQLRestResponseHandler;
  7. import org.junit.Test;
  9. import java.util.HashMap;
  10. import java.util.List;
  11. import java.util.Map;
  13. /**
  14.  * 以rest sql api为例来介绍es 6.3.0的sql orm查询功能
  15.  */
  16. public class SQLOrmTest {
  18.     /**
  19.      * 代码中的sql检索，返回Map类型集合，亦可以返回自定义的对象集合
  20.      */
  21.     @Test
  22.     public void testDemoQuery(){
  23.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  24.         List<Map> json = clientUtil.sql(Map.class,"{\"query\": \"SELECT * FROM demo\"}");
  27.         System.out.println(json);
  28.     }
  30.     /**
  31.      * 代码中的sql检索，返回Map类型集合，亦可以返回自定义的对象集合
  32.      */
  33.     @Test
  34.     public void testMapQuery(){
  35.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  36.         List<Map> json = clientUtil.sql(Map.class,"{\"query\": \"SELECT * FROM dbclobdemo\"}");
  39.         System.out.println(json);
  40.     }
  41.     /**
  42.      * 配置文件中的sql dsl检索,返回Map类型集合，亦可以返回自定义的对象集合
  43.      */
  44.     @Test
  45.     public void testMapSQLQueryFromDSL(){
  46.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  47.         //设置sql查询的参数
  48.         Map params = new HashMap();
  49.         params.put("channelId",1);
  50.         List<Map> json = clientUtil.sql(Map.class,"sqlQuery",params);
  51.         System.out.println(json);
  53.     }
  55.     /**
  56.      * 代码中的sql检索，返回Map类型对象，亦可以返回自定义的对象
  57.      */
  58.     @Test
  59.     public void testMapObjectQuery(){
  60.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  61.         Map json = clientUtil.sqlObject(Map.class,"{\"query\": \"SELECT * FROM dbclobdemo\"}");
  64.         System.out.println(json);
  65.     }
  66.     /**
  67.      * 配置文件中的sql dsl检索,返回Map类型对象，亦可以返回自定义的对象
  68.      */
  69.     @Test
  70.     public void testMapObjectSQLQueryFromDSL(){
  71.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  72.         //设置sql查询的参数
  73.         Map params = new HashMap();
  74.         params.put("channelId",1);
  75.         Map json = clientUtil.sqlObject(Map.class,"sqlQuery",params);
  76.         System.out.println(json);
  78.     }
  81.     /**
  82.      * 代码中的sql检索，返回DocObject 类型集合
  83.      */
  84.     @Test
  85.     public void testObjectListQuery(){
  86.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  87.         List<DocObject> json = clientUtil.sql(DocObject.class,"{\"query\": \"SELECT * FROM dbclobdemo\"}");
  90.         System.out.println(json);
  91.     }
  92.     /**
  93.      * 配置文件中的sql dsl检索,返回DocObject 类型集合
  94.      */
  95.     @Test
  96.     public void testObjectSQLQueryFromDSL(){
  97.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  98.         //设置sql查询的参数
  99.         Map params = new HashMap();
  100.         params.put("channelId",1);
  101.         List<DocObject> json = clientUtil.sql(DocObject.class,"sqlQuery",params);
  102.         System.out.println(json);
  104.     }
  106.     /**
  107.      * 代码中的sql检索，返回DocObject 类型对象
  108.      */
  109.     @Test
  110.     public void testObjectQuery(){
  111.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  112.         DocObject json = clientUtil.sqlObject(DocObject.class,"{\"query\": \"SELECT * FROM dbclobdemo where documentId = 1\"}");
  113.         System.out.println(json);
  114.     }
  115.     /**
  116.      * 配置文件中的sql dsl检索,返回DocObject 类型对象
  117.      */
  118.     @Test
  119.     public void testConditionObjectSQLQueryFromDSL(){
  120.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  121.         //设置sql查询的参数
  122.         Map params = new HashMap();
  123.         params.put("channelId",1);
  124.         DocObject json = clientUtil.sqlObject(DocObject.class,"sqlQuery",params);
  125.         System.out.println(json);
  127.     }
  128.     /**
  129.      * sql转换为dsl
  130.      */
  131.     @Test
  132.     public void testTranslate(){
  133.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  134.         String json = clientUtil.executeHttp("/_xpack/sql/translate",
  135.                 "{\"query\": \"SELECT * FROM dbclobdemo limit 5\",\"fetch_size\": 5}",
  136.                 ClientInterface.HTTP_POST
  137.         );
  138.         System.out.println(json);
  140.     }
  142.     /**
  143.      * 低阶的检索方法
  144.      */
  145.     @Test
  146.     public void testSQLRestResponse(){
  147.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  148.         SQLRestResponse sqlRestResponse = clientUtil.executeHttp("/_xpack/sql",
  149.                                                                     "{\"query\": \"SELECT * FROM dbclobdemo where documentId = 1\"}",
  150.                                                                     ClientInterface.HTTP_POST,
  151.                                                                         new SQLRestResponseHandler());
  152.         System.out.println(sqlRestResponse);
  153.     }
  156. }

每个orm查询方法的都有对应的功能注释说明，可以根据需要使用相关的方法。

代码中用到的sql dsl脚本配置文件及内容：esmapper/sql.xml

  1. <properties>
  2.     <!--
  3.         sql query
  5.     -->
  6.     <property name="sqlQuery">
  7.         <![CDATA[
  8.          {"query": "SELECT * FROM dbclobdemo where channelId=#[channelId]"}
  9.         ]]>
  10.     </property>
  13.     <property name="sqlQueryWithStringParam">
  14.         <![CDATA[
  15.          {"query": "SELECT * FROM dbclobdemo where channelId=#[channelId] and docTitle='#[channelId,quoted=false]'"} ## sql中特定的字符串参数语法
  16.         ]]>
  17.     </property>
  18.     <!--
  19.         分页sql query
  20.         每页显示 fetch_size对应的记录条数
  22.     -->
  23.     <property name="sqlPagineQuery">
  24.         <![CDATA[
  25.          {
  26.          ## 指示sql语句中的回车换行符会被替换掉开始符,注意dsl注释不能放到sql语句中，否则会有问题，因为sql中的回车换行符会被去掉，导致回车换行符后面的语句变道与注释一行
  27.          ##  导致dsl模板解析的时候部分sql段会被去掉
  28.             "query": #"""
  29.                     SELECT * FROM dbclobdemo
  33.                         where channelId=#[channelId]
  34.              """,
  35.              ## 指示sql语句中的回车换行符会被替换掉结束符
  36.             "fetch_size": #[fetchSize]
  37.          }
  38.         ]]>
  39.     </property>
  41. </properties>

我们将配置文件放到工程resources目录下面即可。sql配置说明：

sql中特定的字符串参数语法

  1. {"query": "SELECT * FROM dbclobdemo where channelId=#[channelId] and docTitle='#[channelId,quoted=false]'"}

我们使用#[xxx]类型变量传递sql参数时，如果是字符串内容会自动在值的两边带上双引号，但是在sql语句是字符串值是用单引号’来标识的，所以通过qutoed=false来指示解析引擎不要在值的两边加双引号，然后在外部手动添加单引号：

‘#[channelId,quoted=false]’

如果sql语句比较长，可能要换行，es暂时不支持多行sql语句的执行，bboss通过下面特定的语法，来包围多行sql，sql解析引擎在第一次解析sql的时候讲其中的多行sql解析为一行：

#“””

…

…

“””

例如：

  1. {
  2. ## 指示sql语句中的回车换行符会被替换掉开始符,注意dsl注释不能放到sql语句中，否则会有问题，因为sql中的回车换行符会被去掉，导致回车换行符后面的语句变道与注释一行
  3. ##  导致dsl模板解析的时候部分sql段会被去掉
  4.    "query": #"""
  5.            SELECT * FROM dbclobdemo
  9.                where channelId=#[channelId]
  10.     """,
  11.     ## 指示sql语句中的回车换行符会被替换掉结束符
  12.    "fetch_size": #[fetchSize]
  13. }

1.3 通过fetch_size实现分页查询

  1. package org.bboss.elasticsearchtest.sql;
  3. import org.frameworkset.elasticsearch.ElasticSearchHelper;
  4. import org.frameworkset.elasticsearch.client.ClientInterface;
  5. import org.frameworkset.elasticsearch.entity.sql.SQLResult;
  6. import org.junit.Test;
  8. import java.util.HashMap;
  9. import java.util.List;
  10. import java.util.Map;
  12. public class SQLPagineTest {
  13.     /**
  14.      * 代码中的sql检索，返回Map类型集合，亦可以返回自定义的对象集合
  15.      */
  16.     @Test
  17.     public void testMapQuery(){
  18.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  19.         SQLResult<Map> sqlResult = clientUtil.fetchQuery(Map.class,"{\"query\": \"SELECT * FROM dbclobdemo\",\"fetch_size\": 1}");
  21.         do{
  22.             List<Map> datas = sqlResult.getDatas();
  23.             if(datas == null || datas.size() == 0){
  24.                 break;
  25.             }
  26.             else{
  27.                 System.out.println(datas.size());//处理数据
  28.                 sqlResult = sqlResult.nextPage();//获取下一页数据
  30.             }
  32.         }while(true);
  36.     }
  37.     /**
  38.      * 配置文件中的sql dsl检索,返回Map类型集合，亦可以返回自定义的对象集合
  39.      */
  40.     @Test
  41.     public void testMapSQLQueryFromDSL(){
  42.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  43.         //设置sql查询的参数
  44.         Map params = new HashMap();
  45.         params.put("channelId",1);
  46.         params.put("fetchSize",1);
  47.         SQLResult<Map> sqlResult = clientUtil.fetchQuery(Map.class,"sqlPagineQuery",params);
  49.         do{
  50.             List<Map> datas = sqlResult.getDatas();
  51.             if(datas == null || datas.size() == 0){
  52.                 break;
  53.             }
  54.             else{
  55.                 System.out.println(datas.size());//处理数据
  56.                 sqlResult = sqlResult.nextPage();//获取下一页数据
  58.             }
  60.         }while(true);
  62.     }
  67.     /**
  68.      * 代码中的sql检索，返回DocObject 类型集合 
  69.      */
  70.     @Test
  71.     public void testObjectListQuery(){
  72.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  74.         SQLResult<DocObject> sqlResult = clientUtil.fetchQuery(DocObject.class,"{\"query\": \"SELECT * FROM dbclobdemo\",\"fetch_size\": 1}");
  76.         do{
  77.             List<DocObject> datas = sqlResult.getDatas();
  78.             if(datas == null || datas.size() == 0){
  79.                 break;
  80.             }
  81.             else{
  82.                 System.out.println(datas.size());//处理数据
  83.                 sqlResult = sqlResult.nextPage();//获取下一页数据
  85.             }
  87.         }while(true);
  90.     }
  93.     /**
  94.      * 配置文件中的sql dsl检索,返回DocObject 类型集合 
  95.      */
  96.     @Test
  97.     public void testObjectSQLQueryFromDSL(){
  98.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  99.         //设置sql查询的参数
  101.         Map params = new HashMap();
  102.         params.put("channelId",1);
  103.         params.put("fetchSize",1);
  104.         SQLResult<DocObject> sqlResult = clientUtil.fetchQuery(DocObject.class,"sqlPagineQuery",params);
  106.         do{
  107.             List<DocObject> datas = sqlResult.getDatas();
  108.             if(datas == null || datas.size() == 0){
  109.                 System.out.println(0);//处理数据
  110.                 break;
  111.             }
  112.             else{
  113.                 System.out.println(datas.size());//处理数据
  114.                 sqlResult = sqlResult.nextPage();//获取下一页数据
  116.             }
  118.         }while(true);
  120.     }
  122.     /**
  123.      * 配置文件中的sql dsl检索,返回DocObject 类型集合 
  124.      */
  125.     @Test
  126.     public void testObjectSQLQueryFromDSL1(){
  127.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  128.         //设置sql查询的参数
  130.         Map params = new HashMap();
  131.         params.put("channelId",1);
  132.         params.put("fetchSize",1);
  133.         SQLResult<DocObject> sqlResult = clientUtil.fetchQuery(DocObject.class,"sqlPagineQuery",params);
  135.         do{
  136.             List<DocObject> datas = sqlResult.getDatas();
  137.             if(datas == null || datas.size() == 0){
  138.                 System.out.println(0);//处理数据
  139.                 break;
  140.             }
  141.             else{
  142.                 System.out.println(datas.size());//处理数据
  143.                 sqlResult = clientUtil.fetchQueryByCursor(DocObject.class,sqlResult);//获取下一页数据,通过api获取下一页数据
  145.             }
  147.         }while(true);
  149.     }
  151.     /**
  152.      * 配置文件中的sql dsl检索,返回DocObject 类型集合 
  153.      * 测试没有返回数据的情况
  154.      */
  155.     @Test
  156.     public void testNodataSQLQueryFromDSL1(){
  157.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  158.         //设置sql查询的参数
  160.         Map params = new HashMap();
  161.         params.put("channelId",2);
  162.         params.put("fetchSize",1);
  163.         SQLResult<DocObject> sqlResult = clientUtil.fetchQuery(DocObject.class,"sqlPagineQuery",params);
  166.         do{
  167.             List<DocObject> datas = sqlResult.getDatas();
  168.             if(datas == null || datas.size() == 0){
  169.                 System.out.println(0);//处理数据
  170.                 break;
  171.             }
  172.             else{
  173.                 System.out.println(datas.size());//处理数据
  174.                 sqlResult = clientUtil.fetchQueryByCursor(DocObject.class,sqlResult);//获取下一页数据,通过api获取下一页数据
  176.             }
  178.         }while(true);
  180.     }
  182.     /**
  183.      * 配置文件中的sql dsl检索,返回DocObject 类型集合 
  184.      */
  185.     @Test
  186.     public void testObjectSQLQueryFromDSL2(){
  187.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/sql.xml");//初始化一个加载sql配置文件的es客户端接口
  188.         //设置sql查询的参数
  190.         Map params = new HashMap();
  191.         params.put("channelId",1);
  192.         params.put("fetchSize",1);
  193.         SQLResult<DocObject> sqlResult = clientUtil.fetchQuery(DocObject.class,"sqlPagineQuery",params);
  195.         do{
  196.             List<DocObject> datas = sqlResult.getDatas();
  197.             if(datas == null || datas.size() == 0){
  198.                 System.out.println(0);//处理数据
  199.                 break;
  200.             }
  201.             else{
  202.                 System.out.println(datas.size());//处理数据
  203.                 sqlResult = clientUtil.fetchQueryByCursor(DocObject.class,sqlResult.getCursor(),sqlResult.getColumns());//获取下一页数据,通过api获取下一页数据
  205.             }
  207.         }while(true);
  209.     }
  211.     /**
  212.      * 代码中的sql检索，返回 DocObject类型集合
  213.      */
  214.     @Test
  215.     public void testCloseCursor(){
  216.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  218.         SQLResult<DocObject> sqlResult = clientUtil.fetchQuery(DocObject.class,"{\"query\": \"SELECT * FROM dbclobdemo\",\"fetch_size\": 1}");
  219.         List<DocObject> datas = sqlResult.getDatas();
  220.         System.out.println(datas.size());//处理数据
  221.         System.out.println(sqlResult.closeCursor());//只处理第一页数据，就主动关闭分页游标
  222.     }
  224.     /**
  225.      * 代码中的sql检索，返回DocObject类型集合
  226.      */
  227.     @Test
  228.     public void testCloseCursor1(){
  229.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  231.         SQLResult<DocObject> sqlResult = clientUtil.fetchQuery(DocObject.class,"{\"query\": \"SELECT * FROM dbclobdemo\",\"fetch_size\": 1}");
  232.         List<DocObject> datas = sqlResult.getDatas();
  233.         System.out.println(datas.size());//处理数据
  234.         String ret = clientUtil.closeSQLCursor(sqlResult.getCursor());
  235.         System.out.println(ret);//只处理第一页数据，就主动关闭分页游标
  237.     }
  238. }

2 Elasticsearch-sql查询

基于第三方Elasticsearch-sql插件的查询功能的使用方法和bboss提供的查询api使用方法一致，只是检索的rest服务换成/_sql服务即可：

  1.     /**
  2.      * Elasticsearch-SQL插件功能测试方法
  3.      */
  4.     public void testESSQL(){
  5.         ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  6.         ESDatas<Map> esDatas =  //ESDatas包含当前检索的记录集合，最多10条记录，由sql中的limit属性指定
  7.                 clientUtil.searchList("/_sql",//sql请求
  8.                         "select * from vem_order_index_2018 limit 0,10", //elasticsearch-sql支持的sql语句
  9.                         Map.class);//返回的文档封装对象类型
  10.         //获取结果对象列表
  11.         List<Map> demos = esDatas.getDatas();
  13.         //获取总记录数
  14.         long totalSize = esDatas.getTotalSize();
  15.         System.out.println(totalSize);
  16.     }

Elasticsearch-SQL插件sql转dsl功能:

  1. /**
  2.  * Elasticsearch-SQL插件功能:sql转dsl
  3.  */
  4. @Test
  5. public void testESSQLTranslate(){
  6.    ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();
  7.    String dsl =  //将sql转换为dsl
  8.          clientUtil.executeHttp("/_sql/_explain",//sql转dsl请求
  9.                "select operModule.keyword from dbdemo group by operModule.keyword ",ClientInterface.HTTP_POST);//返回的转换的结果
  11.    //获取总记录数
  12.    System.out.println(dsl);
  13. }

官方的ES-SQL功能目前还不成熟，必须Elasticsearch 6.3以上的版本，而且收费；Elasticsearch-SQL插件功能全，免费，可以在不同的Elasticsearch版本上运行。

3 开发交流

elasticsearch技术交流：166471282

elasticsearch：

bbossgroups