go使用es

elasticsearch有官方的golang驱动go-elasticsearch这个项目比较新， 另外一个常用的是 elastic，这两个驱动文档和demo都比较少。es的查询语法也相对复杂，很多查询方式去翻翻它们的test文件才能发现方式。本小节使用elastic做演示，注意不同elasticsearch版本对于不同的client版本，例如elasticsearch 5.5.3对应的client版本为gopkg.in/olivere/elastic.v5。如果这个对应关系错误，很可能程序会出错，这个在https://github.com/olivere/elastic的readme文档也有介绍。 本小节的demo主要基于 elasticsearch 5.5.3，client为gopkg.in/olivere/elastic.v5。

go链接es

var client *elastic.Client
func init()  {
    var err error
    client, err = elastic.NewClient(elastic.SetURL("http://localhost:9200"))
    if err != nil {
        log.Fatal(err)
    }
}

CURD

测试数据结构体

type Item struct {
    Id               int64  `json:"id"`
    Appid            string `json:"appid"`
    AppBAutoId       string `json:"app_b_auto_id"`
}

添加

func add()  {
    //add one
    item := Item{Id:int64(21),Appid:fmt.Sprintf("app_%d",21),AppBAutoId:fmt.Sprintf("app_%d",21+200)}
    put, err := client.Index().
        Index("es_test").
        Type("test").
        Id("1").       //这个id也可以指定,不指定的话 es自动生成一个
        BodyJson(item).
        Do(context.Background())
    if err != nil {
        // Handle error
        panic(err)
    }
    fmt.Println(put)
    //add many
    bulkRequest := client.Bulk()
    for i:=0;i<20;i++ {
        item := Item{Id:int64(i),Appid:fmt.Sprintf("app_%d",i),AppBAutoId:fmt.Sprintf("app_%d",i+200)}
        bulkRequest.Add(elastic.NewBulkIndexRequest().Index("es_test").Type("test").Doc(item))
    }
    bulkRequest.Do(context.TODO())
}

查找

func find()  {
    //find one
    get1, err := client.Get().
        Index("es_test").
        Type("test").
        Id("1").
        Do(context.Background())
    if err != nil {
        // Handle error
        panic(err)
    }
    if get1.Found {
        fmt.Printf("Got document %s in version %d from index %s, type %s\n", get1.Id, get1.Version, get1.Index, get1.Type)
    }
    var ttyp Item
    json.Unmarshal(*get1.Source,&ttyp)
    fmt.Println("item",ttyp)

    //find many
    searchResult, err := client.Search().
        Index("es_test"). Sort("id", true).
        Type("test").From(0).Size(100).
        Do(context.TODO())
    if err != nil {
        panic(err)
    }
    if searchResult.Hits.TotalHits >0 {
        var ttyp Item
        for _, item := range searchResult.Each(reflect.TypeOf(ttyp)) {
            t := item.(Item)
            fmt.Println("item ",t)
        }

    }
}

更新

func update() {
    fmt.Println(client.Update().Index("es_test").Type("test").Id("1").
        Doc(map[string]interface{}{"appid": "app_23"}).Do(context.TODO()))
}

删除

func delete()  {
    fmt.Println(client.Delete().Index("es_test").Type("test").Id("1").Do(context.TODO()))
}

统计查询

func agg() {
    //获取最大的id
    searchResult, err := client.Search().
        Index("es_test").Type("test").
        Aggregation("max_id", elastic.NewMaxAggregation().Field("id")).Size(0).Do(context.TODO())
    if err != nil {
        panic(err)
    }
    var a map[string]float32
    if searchResult != nil {
        if v, found := searchResult.Aggregations["max_id"]; found {
            json.Unmarshal([]byte(*v), &a)
            fmt.Println(a)
        }
    }
    //统计id相同的文档数
    searchResult, err = client.Search().
        Index("es_test").Type("test").
        Aggregation("count", elastic.NewTermsAggregation().Field("id")).Size(0).Do(context.TODO())
    if err != nil {
        panic(err)
    }
    if searchResult != nil {
        if v, found := searchResult.Aggregations["count"]; found {
            var ar elastic.AggregationBucketKeyItems
            err := json.Unmarshal(*v, &ar)
            if err != nil {
                fmt.Printf("Unmarshal failed: %v\n", err)
                return
            }

            for _, item := range ar.Buckets {
                fmt.Printf("id ：%v: count ：%v\n", item.Key, item.DocCount)

            }
        }
    }
}

参考资料