面试题：ElasticSearch中upsert元素在日志管理场景的应用

Question

在日志管理系统中，假设需要记录用户操作日志，每条日志包含用户ID、操作时间、操作内容等信息。请描述如何使用ElasticSearch的upsert元素来实现，当新日志到来，如果该用户当天已有日志记录，则更新相关操作记录；若没有，则插入新日志记录。

Answer 1

1. 确定索引结构：

   • 首先在Elasticsearch中创建一个合适的索引来存储日志数据。例如，假设索引名为user_operation_logs，文档结构如下：

   {
       "user_id": "string",
       "operation_time": "date",
       "operation_content": "text",
       "date": "date" // 这里新增一个日期字段，方便按日期查找当天记录
   }
   

2. 使用脚本实现Upsert逻辑：

   • Elasticsearch中的upsert操作结合脚本可以实现上述需求。以下是一个使用update API结合upsert的示例，以Python的Elasticsearch客户端为例（假设已安装elasticsearch库）：

   from elasticsearch import Elasticsearch
   
   es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
   
   user_id = "123"
   operation_time = "2023 - 10 - 01T12:00:00"
   operation_content = "User logged in"
   current_date = "2023 - 10 - 01"
   
   doc = {
       "user_id": user_id,
       "operation_time": operation_time,
       "operation_content": operation_content,
       "date": current_date
   }
   
   script = """
   if (ctx._source) {
       ctx._source.operation_content += ';'+ params.operation_content;
       ctx._source.operation_time = params.operation_time;
   } else {
       ctx._source = params.doc;
   }
   """
   
   response = es.update(
       index='user_operation_logs',
       id=user_id + '-' + current_date,
       body={
           "script": {
               "source": script,
               "lang": "painless",
               "params": {
                   "operation_content": operation_content,
                   "operation_time": operation_time,
                   "doc": doc
               }
           },
           "upsert": doc
       }
   )
   print(response)
   

   • 在上述示例中：
     • 我们首先定义了要插入或更新的文档doc。
     • script部分是一个Painless脚本，它会在Elasticsearch服务器端执行。如果文档已经存在（ctx._source存在），则更新operation_content和operation_time字段；如果文档不存在，则将整个ctx._source设置为新的文档params.doc。
     • upsert部分则是当文档不存在时要插入的初始文档。
   • 在实际的Elasticsearch REST API调用中，请求体如下：

   {
       "script": {
           "source": "if (ctx._source) {ctx._source.operation_content += ';'+ params.operation_content;ctx._source.operation_time = params.operation_time;} else {ctx._source = params.doc;}",
           "lang": "painless",
           "params": {
               "operation_content": "User logged in",
               "operation_time": "2023 - 10 - 01T12:00:00",
               "doc": {
                   "user_id": "123",
                   "operation_time": "2023 - 10 - 01T12:00:00",
                   "operation_content": "User logged in",
                   "date": "2023 - 10 - 01"
               }
           }
       },
       "upsert": {
           "user_id": "123",
           "operation_time": "2023 - 10 - 01T12:00:00",
           "operation_content": "User logged in",
           "date": "2023 - 10 - 01"
       }
   }
   

   • 这里使用user_id + '-' + current_date作为文档的id，这样可以保证每个用户每天的日志记录有唯一的标识，方便进行upsert操作。如果使用其他方式确定唯一标识，也需要保证其唯一性。

3. 注意事项：

   • 日期格式：确保日期格式符合Elasticsearch的要求，例如yyyy - MM - dd或yyyy - MM - ddTHH:mm:ss等格式。
   • 脚本语言：上述示例使用Painless脚本，这是Elasticsearch内置的脚本语言。要熟悉其语法和特性，以确保脚本正确执行。
   • 性能考虑：在大规模使用时，批量操作可以提高性能，同时注意索引的设置，如分片数、副本数等，以平衡读写性能。