文档

Push推送Demo

更新时间:

配置环境变量

配置环境变量ALIBABA_CLOUD_ACCESS_KEY_IDALIBABA_CLOUD_ACCESS_KEY_SECRET

重要
  • 阿里云账号AccessKey拥有所有API的访问权限,建议您使用RAM用户进行API访问或日常运维,具体操作,请参见创建RAM用户

  • 创建AccessKey ID和AccessKey Secret,请参考创建AccessKey

  • 如果您使用的是RAM用户的AccessKey,请确保主账号已授权AliyunServiceRoleForOpenSearch服务关联角色,请参考OpenSearch-行业算法版服务关联角色,相关文档参考访问鉴权规则

  • 请不要将AccessKey ID和AccessKey Secret保存到工程代码里,否则可能导致AccessKey泄露,威胁您账号下所有资源的安全。

  • LinuxmacOS系统配置方法:

    执行以下命令,其中, <access_key_id>需替换为您RAM用户的AccessKey ID,<access_key_secret>替换为您RAM用户的AccessKey Secret。

    export ALIBABA_CLOUD_ACCESS_KEY_ID=<access_key_id> 
    export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<access_key_secret>
  • Windows系统配置方法

    1. 新建环境变量文件,添加环境变量ALIBABA_CLOUD_ACCESS_KEY_IDALIBABA_CLOUD_ACCESS_KEY_SECRET,并写入已准备好的AccessKey ID和AccessKey Secret。

    2. 重启Windows系统生效。

V3.1 SDK Push方式新增文档样例代码

Push 推送数据方式,主要是预先生成符合我们规定格式的待推送数据集合,最后在调用Push方法时,将这些数据集合一次性批量推送到应用中。

注意

  • 推送的数据集合必须要符合我们规定的数据集合格式才行,可参考应用控制台->上传文件->参考样例数据,文件中的数据集合格式。

  • 也可以在程序中通过JSONObject 及 JSONArray对象,预先拼接生成符合我们规定格式的数据集合,再调用Push方法一次性将这些数据集合批量推送到应用中。

  • 批量推送文档个数不能太大,不能超过我们规定限制,否则可能会导致推送报错。

适用场景

  • 动态拼接数据推送场景

  • 单个文档推送场景

  • 预先生成待推送数据集合,再批量推送数据集合场景

  • 批量数据推送场景

package com.aliyun.opensearch;

import com.aliyun.opensearch.sdk.dependencies.com.google.common.collect.Lists;
import com.aliyun.opensearch.sdk.dependencies.com.google.common.collect.Maps;
import com.aliyun.opensearch.sdk.dependencies.org.json.JSONArray;
import com.aliyun.opensearch.sdk.dependencies.org.json.JSONObject;
import com.aliyun.opensearch.sdk.generated.OpenSearch;
import com.aliyun.opensearch.sdk.generated.commons.OpenSearchClientException;
import com.aliyun.opensearch.sdk.generated.commons.OpenSearchException;
import com.aliyun.opensearch.sdk.generated.commons.OpenSearchResult;
import com.aliyun.opensearch.sdk.generated.document.Command;
import com.aliyun.opensearch.sdk.generated.document.DocumentConstants;
import com.aliyun.opensearch.sdk.generated.search.*;
import com.aliyun.opensearch.sdk.generated.search.general.SearchResult;

import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.util.Map;
import java.util.Random;

public class testPushSearch2 {

    private String appName = "替换为opensearch应用名";
    private String tableName = "替换opensearch应用表名";
    private String host = "替换应用的API访问地址";

    public static void main(String[] args) {
      //用户识别信息
      //从环境变量读取配置的AccessKey ID和AccessKey Secret,
      //运行代码示例前必须先配置环境变量,参考文档上面“配置环境变量”步骤
      String accesskey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
      String secret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
 		
      //查看文件和默认编码格式
      System.out.println(String.format("file.encoding: %s", System.getProperty("file.encoding")));
      System.out.println(String.format("defaultCharset: %s", Charset.defaultCharset().name
      //-------------数据推送示例代码-----------------
      //生成随机数,作为主键值
      Random rand = new Random();
      int value1 = rand.nextInt(Integer.MAX_VALUE);
      int value2 = rand.nextInt(Integer.MAX_VA
      //定义Map对象存储上传文档数据,此为文档1
      Map<String, Object> doc1 = Maps.newLinkedHashMap();
      doc1.put("id", val
      String title_string = "新增数据Push方式文档1";// utf-8
      byte[] bytes;
      
          bytes = title_string.getBytes("utf-8");
          String utf8_string = new String(bytes, "utf-8");
          doc1.put("name", utf8_str
      } catch (UnsupportedEncodingException e) {
          e.printStackTrace();
  
      doc1.put("phone", "137****11
      int[] int_arr = {11,11};
      doc1.put("int_arr", int_
      String[] literal_arr1 = {"Push方式新增文档1","测试Push方式新增文档1]"};
      doc1.put("literal_arr", literal_a
      float[] float_arr = {(float)1.1,(float)1.1};
      doc1.put("float_arr", float_
      doc1.put("cate_id"
      //标准版不支持update,需要使用ADD进行全字段更新;高级版支持update,部分字段更新。
      JSONObject json1 = new JSONObject();
      json1.put(DocumentConstants.DOC_KEY_CMD, Command.ADD.toString());
      json1.put(DocumentConstants.DOC_KEY_FIELDS, d
      //定义Map对象存储上传文档数据,此为文档2
      Map<String, Object> doc2 = Maps.newLinkedHashMap();
      doc2.put("id", val
      String title_string2 = "新增数据Push方式文档2";// utf-8
      byte[] bytes2;
      
          bytes2 = title_string2.getBytes("utf-8");
          String utf8_string2 = new String(bytes2, "utf-8");
          doc2.put("name", utf8_stri
      } catch (UnsupportedEncodingException e) {
          e.printStackTrace();
  
      doc2.put("phone", "137****22
      int[] int_arr2 = {22,22};
      doc2.put("int_arr", int_a
      String[] literal_arr2 = {"Push方式新增文档2","测试Push方式新增文档2"};
      doc2.put("literal_arr", literal_a
      float[] float_arr2 = {(float)2.2,(float)2.2};
      doc2.put("float_arr", float_a
      doc2.put("cate_id"
      //新增及更新都设为ADD,不支持update,当已存在相同主键值文档时做更新,否则新增,此处作为新增
      JSONObject json2 = new JSONObject();
      json2.put(DocumentConstants.DOC_KEY_CMD, Command.ADD.toString());
      json2.put(DocumentConstants.DOC_KEY_FIELDS, d
      //定义Map对象测试更新文档数据,此为文档3
      Map<String, Object> doc3 = Maps.newLinkedHashMap();
      doc3.put("id", val
      String title_string3 = "更新Push文档2为doc3";// utf-8
      byte[] bytes3;
      
          bytes3 = title_string3.getBytes("utf-8");
          String utf8_string3 = new String(bytes3, "utf-8");
          doc3.put("name", utf8_stri
      } catch (UnsupportedEncodingException e) {
          e.printStackTrace();
  
      doc3.put("phone", "137****33
      int[] int_arr3 = {33,33};
      doc3.put("int_arr", int_a
      String[] literal_arr3 = {"更新Push文档2为doc3","更新Push文档2为doc3"};
      doc3.put("literal_arr", literal_a
      float[] float_arr3 = {(float)3.3,(float)3.3};
      doc3.put("float_arr", float_a
      doc3.put("cate_id"
      //新增及更新都设为ADD,不支持update,需要已存在相同主键值,此处作为更新测试
      JSONObject json3 = new JSONObject();
      json3.put(DocumentConstants.DOC_KEY_CMD, Command.ADD.toString());
      json3.put(DocumentConstants.DOC_KEY_FIELDS, d
      //定义Map对象测试更新文档数据,此为文档4,删除文档只需要设置需删除文档主键值即可,此处测试删除文档1
      Map<String, Object> doc4 = Maps.newLinkedHashMap();
      doc4.put("id", val
      //此处设置删除文档处理
      JSONObject json4 = new JSONObject();
      json4.put(DocumentConstants.DOC_KEY_CMD, Command.DELETE.toString());
      json4.put(DocumentConstants.DOC_KEY_FIELDS, d
      JSONArray docsJsonArr = new JSONArray();
      docsJsonArr.put(json1);//新增文档1
      docsJsonArr.put(json2);//新增文档2
      docsJsonArr.put(json3);//更新Push文档2为doc3
      docsJsonArr.put(json4);//删除文档1
      String docsJson = docsJsonArr.toStri
      //创建并构造OpenSearch对象
      OpenSearch openSearch = new OpenSearch(accesskey, secret, h
      //创建OpenSearchClient对象,并以OpenSearch对象作为构造参数
      OpenSearchClient serviceClient = new OpenSearchClient(openSea
      //定义DocumentClient对象添加json格式doc数据批量提交
      DocumentClient documentClient = new DocumentClient(serviceCli
      try {
          //执行推送操作
          OpenSearchResult osr = documentClient.push(docsJson, appName, tableN
          //判断数据是否推送成功,主要通过判断2处,第一处判断用户方推送是否成功,第二处是应用控制台中有无报错日志
          //用户方推送成功后,也有可能在应用端执行失败,此错误会直接在应用控制台错误日志中生成,比如字段内容转换失败
          if(osr.getResult().equalsIgnoreCase("tru
              System.out.println("用户方推送无报错!\n以下为getTraceInfo推送请求Id:"+osr.getTraceInfo().getRequestI
          }else{
              System.out.println("用户方推送报错!"+osr.getTraceInfo());
      
      } catch (OpenSearchException e) {
          e.printStackTrace();
      } catch (OpenSearchClientException e) {
          e.printStackTrace();
  
      try {
          Thread.sleep(1000);//休眠1秒
      } catch (InterruptedException e) {
          e.printStackTrace();
  
      //-------------查询上面push的文档-----------------
      //创建SearcherClient对象,并以OpenSearchClient对象作为构造参数
      SearcherClient searcherClient = new SearcherClient(serviceCli
      //定义Config对象,用于设定config子句参数,用于分页或设置数据返回格式
      Config config = new Config(Lists.newArrayList(appName));
      config.setStart(0);
      config.setHits(30);
      //设置返回格式为json,目前只支持返回xml和json格式,暂不支持返回fulljson类型
      config.setSearchFormat(SearchFormat.J
      // 设置搜索结果返回应用中哪些字段
      config.setFetchFields(Lists.newArrayList("id", "name", "phone", "int_arr", "literal_arr", "float_arr", "cate_i
      // 创建参数对象
      SearchParams searchParams = new SearchParams(con
      // 设置查询子句,若需多个索引组合查询,需要setQuery处合并,否则若设置多个setQuery后面的会替换前面查询
      searchParams.setQuery("id:'" + value1 + "'|'"+value2 +
      // 设置查询过滤条件
      searchParams.setFilter("cate_id<
      // 设置sort条件
      Sort sorter = new Sort();
      sorter.addToSortFields(new SortField("id", Order.DECREASE)); //设置id字段降序
      sorter.addToSortFields(new SortField("RANK", Order.INCREASE)); //若id相同则以R
      //添加Sort对象参数
      searchParams.setSort(sor
      // 执行返回查询结果
      SearchResult searchResult;
      try {
          searchResult = searcherClient.execute(searchPar
          String result = searchResult.getResult();
          JSONObject obj = new JSONObject(res
          // 输出查询结果
          System.out.println(obj.toStrin
      } catch (OpenSearchException e) {
          e.printStackTrace();
      } catch (OpenSearchClientException e) {
          e.printStackTrace();
  
  }
}

重要

推送数据的时候,只能包含同表的字段,不能跨表推送。