OpenSearch：Push推送Demo

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_VALUE);

    //定義MapObject Storage Service上傳文檔資料,此為文檔1
    Map<String, Object> doc1 = Maps.newLinkedHashMap();
    doc1.put("id", value1);

    String title_string = "新增資料Push方式文檔1"; // utf-8
    byte[] bytes;
    try {
      bytes = title_string.getBytes("utf-8");
      String utf8_string = new String(bytes, "utf-8");
      doc1.put("name", utf8_string);
    } catch (UnsupportedEncodingException e) {
      e.printStackTrace();
    }

    doc1.put("phone", "137****1111");

    int[] int_arr = { 11, 11 };
    doc1.put("int_arr", int_arr);

    String[] literal_arr1 = { "Push方式新增文檔1", "測試Push方式新增文檔1]" };
    doc1.put("literal_arr", literal_arr1);

    float[] float_arr = { (float) 1.1, (float) 1.1 };
    doc1.put("float_arr", float_arr);

    doc1.put("cate_id", 1);

    JSONObject json1 = new JSONObject();
    json1.put(DocumentConstants.DOC_KEY_CMD, Command.ADD.toString());
    json1.put(DocumentConstants.DOC_KEY_FIELDS, doc1);

    //定義MapObject Storage Service上傳文檔資料,此為文檔2
    Map<String, Object> doc2 = Maps.newLinkedHashMap();
    doc2.put("id", value2);

    String title_string2 = "新增資料Push方式文檔2"; // utf-8
    byte[] bytes2;
    try {
      bytes2 = title_string2.getBytes("utf-8");
      String utf8_string2 = new String(bytes2, "utf-8");
      doc2.put("name", utf8_string2);
    } catch (UnsupportedEncodingException e) {
      e.printStackTrace();
    }

    doc2.put("phone", "137****2222");

    int[] int_arr2 = { 22, 22 };
    doc2.put("int_arr", int_arr2);

    String[] literal_arr2 = { "Push方式新增文檔2", "測試Push方式新增文檔2" };
    doc2.put("literal_arr", literal_arr2);

    float[] float_arr2 = { (float) 2.2, (float) 2.2 };
    doc2.put("float_arr", float_arr2);

    doc2.put("cate_id", 1);

    //新增及更新都設為ADD，不支援update，當已存在相同主索引值文檔時做更新，否則新增，此處作為新增
    JSONObject json2 = new JSONObject();
    json2.put(DocumentConstants.DOC_KEY_CMD, Command.ADD.toString());
    json2.put(DocumentConstants.DOC_KEY_FIELDS, doc2);

    //定義Map對象測試更新文檔資料,此為文檔3
    Map<String, Object> doc3 = Maps.newLinkedHashMap();
    doc3.put("id", value2);

    String title_string3 = "更新Push文檔2為doc3"; // utf-8
    byte[] bytes3;
    try {
      bytes3 = title_string3.getBytes("utf-8");
      String utf8_string3 = new String(bytes3, "utf-8");
      doc3.put("name", utf8_string3);
    } catch (UnsupportedEncodingException e) {
      e.printStackTrace();
    }

    doc3.put("phone", "137****3333");

    int[] int_arr3 = { 33, 33 };
    doc3.put("int_arr", int_arr3);

    String[] literal_arr3 = { "更新Push文檔2為doc3", "更新Push文檔2為doc3" };
    doc3.put("literal_arr", literal_arr3);

    float[] float_arr3 = { (float) 3.3, (float) 3.3 };
    doc3.put("float_arr", float_arr3);

    doc3.put("cate_id", 1);

    //新增及更新都設為ADD，不支援update，需要已存在相同主索引值，此處作為更新測試
    JSONObject json3 = new JSONObject();
    json3.put(DocumentConstants.DOC_KEY_CMD, Command.ADD.toString());
    json3.put(DocumentConstants.DOC_KEY_FIELDS, doc3);

    //定義Map對象測試更新文檔資料,此為文檔4，刪除文檔只需要設定需刪除文檔主索引值即可，此處測試刪除文檔1
    Map<String, Object> doc4 = Maps.newLinkedHashMap();
    doc4.put("id", value1);

    //此處設定刪除文檔處理
    JSONObject json4 = new JSONObject();
    json4.put(DocumentConstants.DOC_KEY_CMD, Command.DELETE.toString());
    json4.put(DocumentConstants.DOC_KEY_FIELDS, doc4);

    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.toString();

    //建立並構造OpenSearch對象
    OpenSearch openSearch = new OpenSearch(accesskey, secret, host);

    //建立OpenSearchClient對象，並以OpenSearch對象作為構造參數
    OpenSearchClient serviceClient = new OpenSearchClient(openSearch);

    //定義DocumentClient對象添加json格式doc資料批量提交
    DocumentClient documentClient = new DocumentClient(serviceClient);

    try {
      //執行推送操作
      OpenSearchResult osr = documentClient.push(docsJson, appName, tableName);

      //判斷資料是否推送成功，主要通過判斷2處，第一處判斷使用者方推送是否成功，第二處是應用控制台中有無報錯日誌
      //使用者方推送成功後，也有可能在應用端執行失敗，此錯誤會直接在應用控制台錯誤記錄檔中產生，比如欄位內容轉換失敗
      if (osr.getResult().equalsIgnoreCase("true")) {
        System.out.println(
          "使用者方推送無報錯！\n以下為getTraceInfo推送請求Id:" +
          osr.getTraceInfo().getRequestId()
        );
      } 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(serviceClient);

    //定義Config對象，用於設定config子句參數，用於分頁或設定資料返回格式
    Config config = new Config(Lists.newArrayList(appName));
    config.setStart(0);
    config.setHits(30);
    //設定返回格式為json,目前只支援返回xml和json格式，暫不支援返回fulljson類型
    config.setSearchFormat(SearchFormat.JSON);

    // 設定搜尋結果返回應用中哪些欄位
    config.setFetchFields(
      Lists.newArrayList(
        "id",
        "name",
        "phone",
        "int_arr",
        "literal_arr",
        "float_arr",
        "cate_id"
      )
    );

    // 建立參數對象
    SearchParams searchParams = new SearchParams(config);

    // 設定查詢子句，若需多個索引組合查詢，需要setQuery處合并，否則若設定多個setQuery後面的會替換前面查詢
    searchParams.setQuery("id:'" + value1 + "'|'" + value2 + "'");

    // 設定查詢過濾條件
    searchParams.setFilter("cate_id<=3");

    // 設定sort條件
    Sort sorter = new Sort();
    sorter.addToSortFields(new SortField("id", Order.DECREASE)); //設定id欄位降序
    sorter.addToSortFields(new SortField("RANK", Order.INCREASE)); //若id相同則以RANK升序

    //添加Sort對象參數
    searchParams.setSort(sorter);

    // 執行返回查詢結果
    SearchResult searchResult;
    try {
      searchResult = searcherClient.execute(searchParams);

      String result = searchResult.getResult();
      JSONObject obj = new JSONObject(result);

      // 輸出查詢結果
      System.out.println(obj.toString());
    } catch (OpenSearchException e) {
      e.printStackTrace();
    } catch (OpenSearchClientException e) {
      e.printStackTrace();
    }
  }
}