Elasticsearch基础文档

1
{
2
  "properties": {
3
    "author": {
4
      "type": "object",
5
      "properties": {
6
        "name": {
7
          "type": "keyword"
8
        },
9
        "age": {
10
          "type": "integer"
11
        }
12
      }
13
    }
14
  }
15
}

1
PUT /user
2
{
3
  "mappings": {
4
    "properties": {
5
      "info": {
6
        "type": "text",
7
        "analyzer": "ik_smart"
8
      },
9
      "email": {
10
        "type": "keyword",
11
        "index": false
12
      },
13
      "name": {
14
        "type": "object",
15
        "properties": {
16
          "firstName": {
17
            "type": "keyword"
18
          },
19
          "lastName": {
20
            "type": "keyword"
21
          }
22
        }
23
      }
24
    }
25
  }
26
}

1
CREATE TABLE `tb_hotel`  (
2
  `id` bigint(20) NOT NULL COMMENT '酒店id',
3
  `name` varchar(255) NOT NULL COMMENT '酒店名称',
4
  `address` varchar(255) NOT NULL COMMENT '酒店地址',
5
  `price` int(10) NOT NULL COMMENT '酒店价格',
6
  `score` int(2) NOT NULL COMMENT '酒店评分',
7
  `brand` varchar(32) NOT NULL COMMENT '酒店品牌',
8
  `city` varchar(32) NOT NULL COMMENT '所在城市',
9
  `star_name` varchar(16) NULL DEFAULT NULL COMMENT '酒店星级，1星到5星，1钻到5钻',
10
  `business` varchar(255) NULL DEFAULT NULL COMMENT '商圈',
11
  `latitude` varchar(32) NOT NULL COMMENT '纬度',
12
  `longitude` varchar(32) NOT NULL COMMENT '经度',
13
  `pic` varchar(255) NULL DEFAULT NULL COMMENT '酒店图片',
14
  PRIMARY KEY (`id`) USING BTREE
15
);

1
PUT /hotel
2
{
3
  "mappings": {
4
    "properties": {
5
      "id": {
6
        "type": "keyword"
7
      },
8
      "name": {
9
        "type": "text",
10
        "analyzer": "ik_max_word",
11
        "copy_to": "all"
12
      },
13
      "address": {
14
        "type": "keyword",
15
        "index": false
16
      },
17
      "price": {
18
        "type": "integer"
19
      },
20
      "score": {
21
        "type": "integer"
22
      },
23
      "brand": {
24
        "type": "keyword",
25
        "copy_to": "all"
26
      },
27
      "city": {
28
        "type": "keyword"
29
      },
30
      "startName": {
31
        "type": "keyword"
32
      },
33
      "business": {
34
        "type": "keyword",
35
        "copy_to": "all"
36
      },
37
      "location": {
38
        "type": "geo_point"
39
      },
40
      "pic": {
41
        "type": "keyword",
42
        "index": false
43
      },
44
      "all": {
45
        "type": "text",
46
        "analyzer": "ik_max_word"
47
      }
48
    }
49
  }
50
}

1
{
2
  "name": {
3
    "type": "text",
4
    "analyzer": "ik_max_word",
5
    "copy_to": "all"
6
  },
7
  "brand": {
8
    "type": "keyword",
9
    "copy_to": "all"
10
  },
11
  "business": {
12
    "type": "keyword",
13
    "copy_to": "all"
14
  },
15
  "all": {
16
    "type": "text",
17
    "analyzer": "ik_max_word"
18
  }
19
}

1
// checkResponse 检查 Elasticsearch 响应是否为错误状态。
2
func checkResponse(res *esapi.Response, action string) error {
3
  if !res.IsError() {
4
    return nil
5
  }
6

7
  body, err := readResponseBody(res.Body)
8
  if err != nil {
9
    return fmt.Errorf("%s失败，状态码: %s，且读取响应体失败: %w", action, res.Status(), err)
10
  }
11

12
  return fmt.Errorf("%s失败，状态码: %s，响应: %s", action, res.Status(), body)
13
}
14

15
// readResponseBody 读取 Elasticsearch 响应体并转成字符串。
16
func readResponseBody(body io.Reader) (string, error) {
17
  data, err := io.ReadAll(body)
18
  if err != nil {
19
    return "", err
20
  }
21

22
  return string(data), nil
23
}

1
// hotelIndexMapping 定义 hotel 索引库的 mapping。
2
const hotelIndexMapping = `{
3
  "mappings": {
4
    "properties": {
5
      "id": {
6
        "type": "keyword"
7
      },
8
      "name": {
9
        "type": "text",
10
        "analyzer": "ik_max_word",
11
        "copy_to": "all"
12
      },
13
      "address": {
14
        "type": "keyword",
15
        "index": false
16
      },
17
      "price": {
18
        "type": "integer"
19
      },
20
      "score": {
21
        "type": "integer"
22
      },
23
      "brand": {
24
        "type": "keyword",
25
        "copy_to": "all"
26
      },
27
      "city": {
28
        "type": "keyword"
29
      },
30
      "startName": {
31
        "type": "keyword"
32
      },
33
      "business": {
34
        "type": "keyword",
35
        "copy_to": "all"
36
      },
37
      "location": {
38
        "type": "geo_point"
39
      },
40
      "pic": {
41
        "type": "keyword",
42
        "index": false
43
      },
44
      "all": {
45
        "type": "text",
46
        "analyzer": "ik_max_word"
47
      }
48
    }
49
  }
50
}`
51

52
// CreateHotelIndex 创建 hotel 索引库。
53
func CreateHotelIndex(ctx context.Context, client *elasticsearch.Client) error {
54
  res, err := client.Indices.Create(
55
    "hotel",
56
    client.Indices.Create.WithContext(ctx),
57
    client.Indices.Create.WithBody(strings.NewReader(hotelIndexMapping)),
58
  )
59
  if err != nil {
60
    return fmt.Errorf("请求 Elasticsearch 创建 hotel 索引库失败: %w", err)
61
  }
62
  defer res.Body.Close()
63

64
  return checkResponse(res, "创建 hotel 索引库")
65
}

1
// GetHotelIndex 查询 hotel 索引库信息。
2
// 参数 ctx 用于控制请求生命周期，client 是已经初始化好的 Elasticsearch 客户端。
3
// 返回值为 Elasticsearch 返回的索引库 JSON 信息和可能出现的错误。
4
func GetHotelIndex(ctx context.Context, client *elasticsearch.Client) (string, error) {
5
  // 查询索引库信息，返回结果中包含 settings、mappings 等内容。
6
  res, err := client.Indices.Get(
7
    []string{"hotel"},
8
    client.Indices.Get.WithContext(ctx),
9
  )
10
  if err != nil {
11
    return "", fmt.Errorf("请求 Elasticsearch 查询 hotel 索引库失败: %w", err)
12
  }
13
  defer res.Body.Close()
14

15
  body, err := readResponseBody(res.Body)
16
  if err != nil {
17
    return "", fmt.Errorf("读取 hotel 索引库查询响应失败: %w", err)
18
  }
19

20
  if res.IsError() {
21
    return "", fmt.Errorf("查询 hotel 索引库失败，状态码: %s，响应: %s", res.Status(), body)
22
  }
23

24
  return body, nil
25
}

1
// hotelExtraMapping 定义追加到 hotel 索引库的字段 mapping。
2
const hotelExtraMapping = `{
3
  "properties": {
4
    "remark": {
5
      "type": "text",
6
      "analyzer": "ik_max_word"
7
    }
8
  }
9
}`
10

11
// UpdateHotelIndexMapping 给 hotel 索引库追加字段 mapping。
12
func UpdateHotelIndexMapping(ctx context.Context, client *elasticsearch.Client) error {
13
  // 已存在字段的类型通常不能修改，只能追加新的字段 mapping。
14
  res, err := client.Indices.PutMapping(
15
    []string{"hotel"},
16
    strings.NewReader(hotelExtraMapping),
17
    client.Indices.PutMapping.WithContext(ctx),
18
  )
19
  if err != nil {
20
    return fmt.Errorf("请求 Elasticsearch 修改 hotel 索引库 mapping 失败: %w", err)
21
  }
22
  defer res.Body.Close()
23

24
  return checkResponse(res, "修改 hotel 索引库 mapping")
25
}

1
// hotelDocForDemo 构造用于演示的酒店文档。
2
func hotelDocForDemo() HotelDoc {
3
  return HotelDoc{
4
    ID:        1,
5
    Name:      "如家精选酒店",
6
    Address:   "上海市黄浦区人民广场",
7
    Price:     399,
8
    Score:     45,
9
    Brand:     "如家",
10
    City:      "上海",
11
    StartName: "三钻",
12
    Business:  "人民广场",
13
    Location: GeoLocation{
14
      Lat: 31.2304,
15
      Lon: 121.4737,
16
    },
17
    Pic: "https://example.com/hotel.jpg",
18
  }
19
}
20

21
// CreateHotelDocument 新增或全量覆盖 hotel 文档。
22
func CreateHotelDocument(ctx context.Context, client *elasticsearch.Client) error {
23
  // 将 Go 结构体序列化为 JSON，作为 Elasticsearch 文档请求体。
24
  doc, err := json.Marshal(hotelDocForDemo())
25
  if err != nil {
26
    return fmt.Errorf("序列化 hotel 文档失败: %w", err)
27
  }
28

29
  res, err := client.Index(
30
    "hotel",
31
    strings.NewReader(string(doc)),
32
    client.Index.WithContext(ctx),
33
    client.Index.WithDocumentID("1"),
34
    client.Index.WithRefresh("true"),
35
  )
36
  if err != nil {
37
    return fmt.Errorf("请求 Elasticsearch 新增 hotel 文档失败: %w", err)
38
  }
39
  defer res.Body.Close()
40

41
  return checkResponse(res, "新增 hotel 文档")
42
}

1
// hotelDocsForBulkDemo 构造用于批量导入演示的酒店文档列表。
2
func hotelDocsForBulkDemo() []HotelDoc {
3
  // 先复用单条文档示例，保证字段结构和前面的 mapping 保持一致。
4
  first := hotelDocForDemo()
5

6
  // 再构造第二条文档，用不同 ID 演示 Bulk 一次写入多条数据。
7
  second := hotelDocForDemo()
8
  second.ID = 2
9
  second.Name = "汉庭酒店"
10
  second.Address = "上海市静安区南京西路"
11
  second.Price = 329
12
  second.Score = 43
13
  second.Brand = "汉庭"
14
  second.Business = "南京西路"
15
  second.Location = GeoLocation{Lat: 31.2336, Lon: 121.4563}
16

17
  return []HotelDoc{first, second}
18
}
19

20
// BulkCreateHotelDocuments 批量导入 hotel 文档。
21
func BulkCreateHotelDocuments(ctx context.Context, client *elasticsearch.Client, docs []HotelDoc) error {
22
  // 批量导入没有数据时直接返回，避免向 Elasticsearch 发送空的 bulk 请求。
23
  if len(docs) == 0 {
24
    return nil
25
  }
26

27
  // Bulk API 使用 NDJSON 格式，每条文档都需要先写操作行，再写数据行。
28
  var body strings.Builder
29
  for _, doc := range docs {
30
    // index 操作表示新增或覆盖文档，这里显式指定索引名和文档ID。
31
    // Bulk API 里的每一条操作，都需要一行操作元数据,表示：我要执行index操作，目标索引库是hotel，文档ID是doc.ID
32
    // 如果是单条新增，就不需要自己写外面这一层，因为索引名和文档ID已经通过API参数传了
33
    meta := map[string]map[string]string{
34
      "index": {
35
        "_index": "hotel",
36
        "_id":    fmt.Sprintf("%d", doc.ID),
37
      },
38
    }
39

40
    // 序列化操作元数据，避免手写 JSON 字符串导致格式错误。
41
    metaLine, err := json.Marshal(meta)
42
    if err != nil {
43
      return fmt.Errorf("序列化 hotel 批量导入元数据失败: %w", err)
44
    }
45

46
    // 序列化文档内容，作为 index 操作的下一行数据。
47
    docLine, err := json.Marshal(doc)
48
    if err != nil {
49
      return fmt.Errorf("序列化 hotel 批量导入文档失败: %w", err)
50
    }
51

52
    // 每一行都必须追加换行符，最后一行也不能省略换行。
53
    body.Write(metaLine)
54
    body.WriteByte('\n')
55
    body.Write(docLine)
56
    body.WriteByte('\n')
57
  }
58

59
  // 发送批量请求；学习环境使用 refresh=true，方便马上查询导入结果。
60
  res, err := client.Bulk(
61
    strings.NewReader(body.String()),
62
    client.Bulk.WithContext(ctx),
63
    client.Bulk.WithRefresh("true"),
64
  )
65
  if err != nil {
66
    return fmt.Errorf("请求 Elasticsearch 批量导入 hotel 文档失败: %w", err)
67
  }
68
  defer res.Body.Close()
69

70
  // Bulk 响应需要读取出来，因为 HTTP 200 也可能包含单条写入失败。
71
  respBody, err := readResponseBody(res.Body)
72
  if err != nil {
73
    return fmt.Errorf("读取 hotel 批量导入响应失败: %w", err)
74
  }
75

76
  // 先检查整个 Bulk 请求本身是否失败。
77
  if res.IsError() {
78
    return fmt.Errorf("批量导入 hotel 文档失败，状态码: %s，响应: %s", res.Status(), respBody)
79
  }
80

81
  // 再检查 Bulk 响应里的 errors 字段，确认是否存在单条文档失败。
82
  var result struct {
83
    Errors bool `json:"errors"`
84
  }
85
  if err := json.Unmarshal([]byte(respBody), &result); err != nil {
86
    return fmt.Errorf("解析 hotel 批量导入响应失败: %w", err)
87
  }
88
  if result.Errors {
89
    return fmt.Errorf("批量导入 hotel 文档存在失败项，响应: %s", respBody)
90
  }
91

92
  return nil
93
}

1
// GetHotelDocument 根据文档 ID 查询 hotel 文档。
2
func GetHotelDocument(ctx context.Context, client *elasticsearch.Client, id string) (string, error) {
3
  res, err := client.Get(
4
    "hotel",
5
    id,
6
    client.Get.WithContext(ctx),
7
  )
8
  if err != nil {
9
    return "", fmt.Errorf("请求 Elasticsearch 查询 hotel 文档失败: %w", err)
10
  }
11
  defer res.Body.Close()
12

13
  body, err := readResponseBody(res.Body)
14
  if err != nil {
15
    return "", fmt.Errorf("读取 hotel 文档查询响应失败: %w", err)
16
  }
17

18
  if res.IsError() {
19
    return "", fmt.Errorf("查询 hotel 文档失败，状态码: %s，响应: %s", res.Status(), body)
20
  }
21

22
  return body, nil
23
}

1
// UpdateHotelDocument 局部修改 hotel 文档。
2
func UpdateHotelDocument(ctx context.Context, client *elasticsearch.Client, id string) error {
3
  body := `{
4
  "doc": {
5
    "price": 429,
6
    "remark": "靠近人民广场，适合出差住宿"
7
  }
8
}`
9

10
  res, err := client.Update(
11
    "hotel",
12
    id,
13
    strings.NewReader(body),
14
    client.Update.WithContext(ctx),
15
    client.Update.WithRefresh("true"),
16
  )
17
  if err != nil {
18
    return fmt.Errorf("请求 Elasticsearch 修改 hotel 文档失败: %w", err)
19
  }
20
  defer res.Body.Close()
21

22
  return checkResponse(res, "修改 hotel 文档")
23
}

1
// ReplaceHotelDocument 全量覆盖 hotel 文档。
2
func ReplaceHotelDocument(ctx context.Context, client *elasticsearch.Client, id string) error {
3
  // 先构造一份完整文档；全量覆盖必须提交完整 _source，避免重要字段被意外清空。
4
  doc := hotelDocForDemo()
5
  doc.Price = 459
6
  doc.Score = 48
7

8
  // 将完整文档序列化为 JSON，作为覆盖写入的请求体。
9
  body, err := json.Marshal(doc)
10
  if err != nil {
11
    return fmt.Errorf("序列化 hotel 覆盖文档失败: %w", err)
12
  }
13

14
  // 对同一个文档 ID 再次执行 Index，Elasticsearch 会用请求体覆盖该文档的 _source。
15
  res, err := client.Index(
16
    "hotel",
17
    strings.NewReader(string(body)),
18
    client.Index.WithContext(ctx),
19
    client.Index.WithDocumentID(id),
20
    client.Index.WithRefresh("true"),
21
  )
22
  if err != nil {
23
    return fmt.Errorf("请求 Elasticsearch 全量覆盖 hotel 文档失败: %w", err)
24
  }
25
  defer res.Body.Close()
26

27
  return checkResponse(res, "全量覆盖 hotel 文档")
28
}

1
// DeleteHotelDocument 根据文档 ID 删除 hotel 文档。
2
func DeleteHotelDocument(ctx context.Context, client *elasticsearch.Client, id string) error {
3
  res, err := client.Delete(
4
    "hotel",
5
    id,
6
    client.Delete.WithContext(ctx),
7
    client.Delete.WithRefresh("true"),
8
  )
9
  if err != nil {
10
    return fmt.Errorf("请求 Elasticsearch 删除 hotel 文档失败: %w", err)
11
  }
12
  defer res.Body.Close()
13

14
  return checkResponse(res, "删除 hotel 文档")
15
}