使用 JSON 模式和 ElasticSearch Java API 添加类型映射

Question

是否可以通过 Java API 使用 JSON 架构将类型 mapping 添加到 ElasticSearch Index？

我知道 ElasticSearch 使用第一个文档来创建 mapping，因此我可以使用 json 架构增强我的第一个文档。 但我想在索引文档之前创建类型。

Answer 1

你可以这样做：

String mapping = XContentFactory.jsonBuilder().startObject().startObject(typeName).startObject("properties")
                    .startObject("location").field("type", "geo_point").endObject()
                    .startObject("language").field("type", "string").field("index", "not_analyzed").endObject()
                    .startObject("user").startObject("properties").startObject("screen_name").field("type", "string").field("index", "not_analyzed").endObject().endObject().endObject()
                    .startObject("mention").startObject("properties").startObject("screen_name").field("type", "string").field("index", "not_analyzed").endObject().endObject().endObject()
                    .startObject("in_reply").startObject("properties").startObject("user_screen_name").field("type", "string").field("index", "not_analyzed").endObject().endObject().endObject()
                    .startObject("retweet").startObject("properties").startObject("user_screen_name").field("type", "string").field("index", "not_analyzed").endObject().endObject().endObject()
                    .endObject().endObject().endObject().string();
client.admin().indices().preparePutMapping(indexName).setType(typeName).setSource(mapping).execute().actionGet();

或者，如果您将映射作为字符串

String json = "{}";
PutMappingResponse response = client.admin().indices()
                    .preparePutMapping(index)
                    .setType(type)
                    .setSource(json)
                    .execute().actionGet();     

Answer 2

可以使用java api应用elasticsearch json映射，

步骤 1) 首先在 json 文件中为 Elasticsearch 类型创建映射，

例如。 resources/Customer.json

{
    "Customer": {
        "settings": {}, 
        "properties": { 
            "name": { "type":"String" , "index": "not_analyzed"}
        }
    }
}

步骤 2) 创建一个 java 方法以从 json 文件应用映射，（参见完整示例 here）

class EsUtils {

  public static Client client

  public static void applyMapping(String index, String type, String location) throws Exception {

            String source = readJsonDefn(location);

            if (source != null) {
                PutMappingRequestBuilder pmrb = client.admin().indices()
                                                      .preparePutMapping(index)
                                                      .setType(type);
                pmrb.setSource(source);
                MappingListener mappingListener = new MappingListener(pmrb)

                // Create type and mapping
                Thread thread = new Thread(mappingListener)

                thread.start();
                while (!mappingListener.processComplete.get()) {
                    System.out.println("not complete yet. Waiting for 100 ms")
                    Thread.sleep(100);

                }

            } else {
                   System.out.println("mapping error");
            }

       }

       public static String readJsonDefn(String url) throws Exception {
              //implement it the way you like 
              StringBuffer bufferJSON = new StringBuffer();

              FileInputStream input = new FileInputStream(new File(url).absolutePath);
              DataInputStream inputStream = new DataInputStream(input);
              BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));

              String line;

              while ((line = br.readLine()) != null) {
                             bufferJSON.append(line);
              }
              br.close();
              return bufferJSON.toString();
       }

    //runnable mapping listener
    static class MappingListener implements Runnable {
      PutMappingRequestBuilder requestBuilder;
      public AtomicBoolean processComplete;
      PutMappingActionListener actionListener;

      @Override
      void run() {
        try {
            requestBuilder.execute(actionListener)
        } catch (Exception e) {
            e.printStackTrace()
            this.processComplete.set(true)
        }
      }

      public MappingListener(PutMappingRequestBuilder requestBuilder) {
        this.processComplete = new AtomicBoolean(false);
        actionListener = new PutMappingActionListener(processComplete);
        this.requestBuilder = requestBuilder;
      }

     //action listener
     static class PutMappingActionListener implements ActionListener<PutMappingResponse> {
        public AtomicBoolean processComplete;

        public PutMappingActionListener(AtomicBoolean processComplete) {
            this.processComplete = processComplete;
        }

        void onResponse(PutMappingResponse response) {
            if (response.isAcknowledged()) {
                System.out.println("template successfully applied")
            }
            processComplete.set(true)
        }

        @Override
        void onFailure(Throwable throwable) {
            System.out.println("error applying mapping : " + throwable)
            throwable.printStackTrace()
            processComplete.set(true)
        }
      }
     } //end of mappinglistener

    }

STEP 3) 调用applyMapping() 方法通过你的es 客户端应用映射，

String index = "yourIndex"; //yourIndex
String type  = "Customer";
String location = "resources/Customer.json";

EsUtils.client = esClient; //pass your client
EsUtils.applyMapping(index, type, location);

STEP 4) 随意查询,

SearchRequestBuilder builder = client.prepareSearch("yourIndex");
builder.addAggregation(AggregationBuilders.terms("nameterms")
                                          .field("name").size(0))
SearchResponse response = builder.execute().actionGet();

完整参考

Elasticsearch apply mapping

Elasticsearch mapping settings 'not_analyzed' and grouping by field in Java

Answer 3

我已经像 Prayag 的回答那样做了，不同之处在于不使用 MappingListener。

import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequestBuilder;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;

public class MyElasticSearchService {

  private static final String INDEX = "myIndex";

  private Client client;

  private JsonSchema jsonSchema;

  ... 

  public void createIndexAndTypes() {
    createIndex();
    createType(TYPE_FOO, this.jsonSchema.getFooSchema());
    createType(TYPE_BAR, this.jsonSchema.getBarSchema());
  }

  private void createType(final String type, final String schema) {
    final PutMappingRequestBuilder requestBuilder = this.client.admin().indices().preparePutMapping(INDEX).setType(type);
    final PutMappingResponse response = requestBuilder.setSource(schema).execute().actionGet();
  }
}



import lombok.Getter;

@Getter
public class JsonSchema {

  private String fooSchema;
  private String barSchema;

  public JsonSchema() {
    this.fooSchema = ClasspathUtil.getTextfileContent("foo_schema.json");
    this.barSchema = ClasspathUtil.getTextfileContent("bar_schema.json");
  }
}



import java.io.IOException;

import org.apache.commons.io.IOUtils;
import org.springframework.core.io.ClassPathResource;

public class ClasspathUtil {

    public static String getTextfileContent(final String filename) {
        try {
            return IOUtils.toString(new ClassPathResource(filename).getInputStream());
        } catch (final IOException e) {
            throw new IllegalStateException(e);
        }
    }
}

Answer 4

请在附件中找到针对您的问题的解决方案，在工作代码示例中进行了描述。我在 ElasticSearch 中使用术语过滤器时遇到问题。字段 indexrecord_uuid 包含一个 UUID 字符串值（例如 8457964d-72e4-4b96-9232-c0e90fccd57d），它没有正确响应使用 TermsFilterBuilder 的精确匹配查询。

通过调用 Elastcisearch 分析器 API，我发现这个字段的结构不是单个字符串，而是拆分为多个标记。该问题在here 中有详细描述。

    GET /_analyze?analyzer=standard
{
  "8457964d-72e4-4b96-9232-c0e90fccd57d"
}

返回结果：

{
   "tokens": [
      {
         "token": "8457964d",
         "start_offset": 5,
         "end_offset": 13,
         "type": "<ALPHANUM>",
         "position": 1
      },
      {
         "token": "72e4",
         "start_offset": 14,
         "end_offset": 18,
         "type": "<ALPHANUM>",
         "position": 2
      },
      {
         "token": "4b96",
         "start_offset": 19,
         "end_offset": 23,
         "type": "<ALPHANUM>",
         "position": 3
      },
      {
         "token": "9232",
         "start_offset": 24,
         "end_offset": 28,
         "type": "<NUM>",
         "position": 4
      },
      {
         "token": "c0e90fccd57d",
         "start_offset": 29,
         "end_offset": 41,
         "type": "<ALPHANUM>",
         "position": 5
      }
   ]
}

所以要启用这样的查询

    GET /backmeup/backup/_search?pretty
{
  "query": {
    "bool" : {
    "must" : [ {
      "match" : {
        "owner_id" : {
          "query" : "2",
          "type" : "boolean"
        }
      }
    },{
      "bool" : {
        "should" : {
          "bool" : {
            "must" : {
              "filtered" : {
                "query" : {
                  "match_all" : { }
                },
                "filter" : {
                  "terms" : {
                    "indexrecord_uuid" : [ "8457964d-72e4-4b96-9232-c0e90fccd57d"]
                  }
                }
              }
            }
          }
        },
        "minimum_should_match" : "1"
      }
    }]
  }
  }
}

我不得不删除索引并更新字段映射并将其设置为 not_analyzed。

{
"backup" : {
     "properties" : {
         "indexrecord_uuid" : {
             "type" : "string",
             "index" : "not_analyzed" 
         }
     }
 }
}

发出 PutMappingRequest

PutMappingRequestBuilder pmrb = this.client.admin().indices().preparePutMapping(INDEX_NAME)
                .setType("backup");
        pmrb.setSource(this.getIndexCustomFieldMapping());
        PutMappingResponse putMappingResponse = pmrb.execute().actionGet();
        if (!putMappingResponse.isAcknowledged()) {
            this.logger.error("Could not create index [" + INDEX_NAME + " ].");
        } else {
            this.logger.debug("Successfully created index [" + INDEX_NAME + " ].");
        }

要检查现有版本的索引是否已经配置为使用此映射配置，您可以这样做

 /**
 * Checks if the currently existing version of the index was configured properly e.g. if the indexrecord_uuid field
 * mapping was set etc
 * 
 * check if the field mapping for indexrecord_uuid was set to not analyzed as this causes issues using the terms
 * filter on UUID Strings https://www.elastic.co/guide/en/elasticsearch/guide/current/_finding_exact_values.html
 * https://www.elastic.co/guide/en/elasticsearch/guide/current/analysis-intro.html#analyze-api
 * https://www.elastic.co/guide/en/elasticsearch/guide/current/_finding_exact_values.html
 */
private boolean checkIsIndexFieldMappingSet() {
    GetMappingsResponse mapping = this.client.admin().indices().prepareGetMappings(INDEX_NAME).get();
    try {
        HashMap props = (HashMap) mapping.getMappings().get(INDEX_NAME).get("backup").getSourceAsMap();
        if (props != null) {
            if (props.containsKey("properties")) {
                HashMap fieldMappings = ((HashMap) props.get("properties"));
                //check if the field mapping for indexrecord_uuid was set to not analyzed 
                if (fieldMappings.containsKey("indexrecord_uuid")) {
                    HashMap fieldIndexRecordMapping = (HashMap) fieldMappings.get("indexrecord_uuid");
                    if (fieldIndexRecordMapping.containsKey("index")
                            && fieldIndexRecordMapping.get("index").toString().equals("not_analyzed")) {
                        return true;
                    }
                }

            }
        }
    } catch (Exception e) {
        this.logger.debug("Index does not contain a 'not_analyzed' field mapping for indexrecord_uuid");
    }
    return false;
}