Elasticsearch写入、读取、更新、删除以及批量操作（Golang）

1. 背景

加入自动驾驶事业部数仓团队后，工作围绕数据和检索展开，熟练掌握 Elasticsearch 的基本操作成为提高工作效率的关键。在项目中使用官方提供的包 github.com/elastic/go-elasticsearch，但其底层设计复杂，开发成本较高，因此使用了三方包 github.com/olivere/elastic（版本 7.0.32），以实现更简便的开发。

2. Elasticsearch基础操作

2.1 创建 Elasticsearch 客户端

定义一个 ES 实例结构：

type EsInstance struct {
  Client *elastic.Client    // es客户端
  Index  map[string]Indexes // 所有索引
}
 
type Indexes struct {
  Name    string `json:"name"`    // 索引名称
  Mapping string `json:"mapping"` // 索引结构
}

创建实例：

func NewEsInstance() (*EsInstance, error) {
  client, err := elastic.NewClient(
    elastic.SetURL("http://127.0.0.1:9200"), 
    elastic.SetBasicAuth("user_name", "user_password"),
    elastic.SetSniff(false), // 跳过ip检查
  )
  if err != nil {
    return &EsInstance{}, err
  }
  return &EsInstance{
    Client: client,
    Index:  map[string]Indexes{},
  }, nil
}

2.2 创建、删除索引

创建索引：

func (es *EsInstance) CreateIndex(index string, mapping string) error {
  exists, err := es.Client.IndexExists(index).Do(context.Background())
  if err != nil {
    return err
  }
  if exists {
    return nil
  }
  createIndex, err := es.Client.CreateIndex(index).BodyString(mapping).Do(context.Background())
  if err != nil {
    return err
  }
  if !createIndex.Acknowledged {
    return errors.New("create index failed")
  }
  es.Index[index] = Indexes{Name: index, Mapping: mapping}
  return nil
}

删除索引：

func (es *EsInstance) DeleteIndex(index string) error {
  exists, err := es.Client.IndexExists(index).Do(context.Background())
  if err != nil {
    return err
  }
  if !exists {
    return nil
  }
  deleteIndex, err := es.Client.DeleteIndex(index).Do(context.Background())
  if err != nil {
    return err
  }
  if !deleteIndex.Acknowledged {
    return errors.New("delete index failed")
  }
  return nil
}

2.3 插入文档

定义文档结构：

type Record struct {
  Index  string `json:"_index"`
  Type   string `json:"_type"`
  Id     string `json:"_id"`
  Source Source `json:"source"`
}

type Source struct {
  Id   string `json:"id"`
  Name string `json:"name"`
  Age  int    `json:"age"`
  Sex  string `json:"sex"`
}

插入文档：

func (es *EsInstance) InsertOneRecord(indexName string, record Record) error {
  _, err := es.Client.Index().Index(indexName).Id(record.Source.Id).BodyJson(record).Do(context.Background())
  return err
}

2.4 查询文档

根据文档 id 获取文档：

func (es *EsInstance) GetOneRecord(indexName, id string) (*Record, error) {
  record := &Record{}
  result, err := es.Client.Get().Index(indexName).Id(id).Do(context.Background())
  if err != nil {
    return record, err
  }
  if result.Found {
    err := json.Unmarshal(result.Source, &record.Source)
  }
  record.Id = result.Id
  record.Index = result.Index
  record.Type = result.Type
  return record, nil
}

2.5 删除文档

根据文档 id 删除文档：

func (es *EsInstance) DeleteOneRecord(indexName, id string) error {
  _, err := es.Client.Delete().Index(indexName).Id(id).Do(context.Background())
  return err
}

2.6 更新文档

根据文档 id 更新文档：

func (es *EsInstance) UpdateOneRecord(indexName, id string, record Source) error {
  _, err := es.Client.Update().Index(indexName).Id(id).Doc(record).Do(context.Background())
  return err
}

2.7 逻辑查询

自定义 DSL 查询：

func (es *EsInstance) Search(indexName string, size int, query elastic.Query) ([]Record, error) {
  records := make([]Record, 0)
  searchResult, err := es.Client.Search().Index(indexName).Query(query).Size(size).Do(context.Background())
  if err != nil {
    return records, err
  }
  if searchResult.Hits.TotalHits.Value > 0 {
    for _, hit := range searchResult.Hits.Hits {
      record := &Record{}
      json.Unmarshal(hit.Source, &record.Source)
      record.Id = hit.Id
      records = append(records, *record)
    }
  }
  return records, nil
}

2.8 滚动查询

滚动查询数据：

func (es *EsInstance) SearchScroll(indexName string, size int, query elastic.Query) ([]Record, error) {
  records := make([]Record, 0)
  searchResult, err := es.Client.Scroll(indexName).Query(query).Size(size).Do(context.Background())
  scrollId := searchResult.ScrollId

  for {
    scrollResult, err := es.Client.Scroll().ScrollId(scrollId).Do(context.Background())
    if err != nil || scrollResult.Hits.TotalHits.Value == 0 {
      break
    }
    scrollId = scrollResult.ScrollId
    for _, hit := range scrollResult.Hits.Hits {
      record := &Record{}
      json.Unmarshal(hit.Source, &record.Source)
      record.Id = hit.Id
      records = append(records, *record)
    }
  }
  es.Client.ClearScroll(scrollId).Do(context.Background())
  return records, err
}

2.9 批量插入

批量插入数据：

func (es *EsInstance) BatchInsertRecords(indexName string, records []Record) error {
  req := es.Client.Bulk().Index(indexName)
  for _, record := range records {
    doc := elastic.NewBulkIndexRequest().Id(record.Id).Doc(&record.Source)
    req.Add(doc)
  }
  _, err := req.Do(context.Background())
  return err
}

2.10 批量更新

批量更新数据：

func (es *EsInstance) BatchUpdateRecords(indexName string, records []Record) error {
  req := es.Client.Bulk().Index(indexName)
  for _, record := range records {
    doc := elastic.NewBulkUpdateRequest().Id(record.Id).Doc(&record.Source)
    req.Add(doc)
  }
  _, err := req.Do(context.Background())
  return err
}

2.11 批量删除

批量删除数据：

func (es *EsInstance) BatchDeleteRecords(indexName string, records []Record) error {
  req := es.Client.Bulk().Index(indexName)
  for _, record := range records {
    doc := elastic.NewBulkDeleteRequest().Id(record.Id)
    req.Add(doc)
  }
  _, err := req.Do(context.Background())
  return err
}

3. 检索条件设置

示例：

boolQuery := elastic.NewBoolQuery()
boolQuery.Filter(elastic.NewMatchQuery("tag_name", "turn_left"))
boolQuery.Filter(elastic.NewRangeQuery("start_time").Gte(1723737600000))

4. 测试

检索某辆车、某个时间段的数据：

boolQuery.Filter(elastic.NewMatchQuery("tag_name", "turn_left"))
boolQuery.Filter(elastic.NewMatchQuery("car_id", "京A0001"))
boolQuery.Filter(elastic.NewRangeQuery("start_time").Gte(1723737600000))
boolQuery.Filter(elastic.NewRangeQuery("end_time").Lte(1723823999000))

SearchScroll("index_2024", 4000, boolQuery)

5. 总结

本文介绍了 Elasticsearch 的基础操作，包括创建、删除、更新、查询、批量操作等。使用三方包 github.com/olivere/elastic 能有效降低开发成本，提升工作效率。