go 语言操作es示例
2024-10-21 20:43:50
使用高度封装的 orm 查询
package main
import (
"context"
"fmt"
"log"
"os"
"reflect"
"time"
"github.com/olivere/elastic/v7"
)
type esObj struct {
db *elastic.Client
index string
}
type Blog struct {
Title string `json:"title"`
Author string `json:"author"`
Content string `json:"content"`
PageView int64 `json:"pageview"`
ReleaseTime time.Time `json:"release_time"`
IsDel bool `json:"is_del"`
}
const blogMap = `{
"mappings": {
"properties": {
"title": {
"type": "text",
"analyzer": "ik_max_word",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"author": {
"type": "keyword"
},
"content": {
"type": "text",
"analyzer": "ik_max_word"
},
"pageview": {
"type": "long"
},
"release_time": {
"type": "date"
},
"isdel": {
"type": "boolean"
}
}
}
}`
func main() {
log.SetFlags(log.Ltime | log.Lshortfile)
ips := []string{"http://127.0.0.1:9200", "http://127.0.0.1:9201"}
esDb, err := New(ips, "", "")
if err != nil {
return
}
ctx := context.Background()
if exist, err := esDb.IndexExist(ctx); err != nil {
return
} else if !exist {
log.Printf("index of %v not exist", esDb.index)
if err = esDb.CreateIndex(ctx); err != nil {
return
}
}
log.Println("es 查询成功")
defer esDb.DeleteIndex(ctx)
blogs := getBlogDemo()
if err := esDb.InsertPost(ctx, blogs[0]); err != nil {
return
}
_ = esDb.InsertPostList(ctx, blogs)
_ = esDb.QueryMathAll(ctx, false)
_ = esDb.QueryBlogNumbByAuthor(ctx, "李白")
_ = esDb.QueryBlogByContent(ctx, "黄河", false)
_ = esDb.AggsSumPageViesByAuthorName(ctx, "李白")
esDb.BoolQueryBlogByAuthor(ctx, "李白", false)
esDb.BucketAggs(ctx, true)
log.Println("插入成功")
}
func New(ips []string, uname, password string) (es *esObj, err error) {
esOptions := make([]elastic.ClientOptionFunc, 0)
if len(ips) == 0 {
log.Fatalf("ip地址输入错误")
}
esOptions = append(esOptions, elastic.SetURL(ips...))
if len(uname) > 0 && len(password) > 0 {
esOptions = append(esOptions, elastic.SetBasicAuth(uname, password))
}
esOptions = append(
// 设置健康检查时间
esOptions, elastic.SetHealthcheckInterval(10*time.Second),
// 启动压缩算法
elastic.SetGzip(true),
// 设置错误日志输出
elastic.SetErrorLog(log.New(os.Stdout, "es--error ", log.LstdFlags)),
// 设置info日志输出
elastic.SetInfoLog(log.New(os.Stdout, "es--info ", log.LstdFlags)),
)
esClient, err := elastic.NewClient(esOptions...)
if err != nil {
log.Printf("es 启动失败 err is %v", err)
return nil, fmt.Errorf("es 启动失败 err is %v", err)
}
es = &esObj{
db: esClient,
index: "blog",
}
return es, nil
}
// IndexExist 判断索引是否存在
func (es *esObj) IndexExist(ctx context.Context) (bool, error) {
exist, err := es.db.IndexExists(es.index).Do(ctx)
if err != nil {
log.Printf("es indexExist fail; index is %s, err is %v", es.index, err)
return false, err
}
return exist, nil
}
// CreateIndex 创建索引
func (es *esObj) CreateIndex(ctx context.Context) (err error) {
result, err := es.db.CreateIndex(es.index).BodyString(blogMap).Do(ctx)
if err != nil {
log.Fatalf("create index fail , index is %s, err is %v", es.index, err)
return err
}
if !result.Acknowledged {
log.Fatalf("not createIndex.Acknowledged ; index is %s", es.index)
return nil
}
log.Printf("es index of %s create success maping is %v", es.index, blogMap)
return nil
}
// DeleteIndex 删除索引
func (es *esObj) DeleteIndex(ctx context.Context) (bool, error) {
res, err := es.db.DeleteIndex(es.index).Do(ctx)
if err != nil {
log.Fatalf("delete index of %s fail, err is %v", es.index, err)
return false, err
}
if !res.Acknowledged {
log.Printf("delete index of %s not acknowledged", es.index)
return false, nil
}
log.Printf("delet index of %s success", es.index)
return true, nil
}
// InsertPost 添加数据
func (es *esObj) InsertPost(ctx context.Context, blog Blog) error {
res, err := es.db.Index().Index(es.index).BodyJson(blog).Do(ctx)
if err != nil {
log.Fatalf("es insert in %s fail, blog is %+v; err is %v", es.index, blog, err)
return err
}
log.Printf("insert result %v", res)
return nil
}
// InsertPostList 批量添加数据
func (es *esObj) InsertPostList(ctx context.Context, blogs []Blog) error {
bulkService := es.db.Bulk().Index(es.index).Refresh("true")
for _, blog := range blogs {
bulkService.Add(elastic.NewBulkCreateRequest().Index(es.index).Doc(blog))
}
res, err := bulkService.Do(ctx)
if err != nil {
log.Fatalf("index of %s bulk insert err %v", es.index, err)
return err
}
log.Printf("bulk insert success %v", res.Errors)
return nil
}
// QueryMathAll match_all 查询
func (es *esObj) QueryMathAll(ctx context.Context, print bool) error {
res, err := es.db.Search().
Index(es.index).
Query(elastic.NewMatchAllQuery()).
Do(ctx)
if err != nil {
log.Fatalf("queryMathALL fail err is %v", err)
return err
}
log.Printf("查询花费的时间 %v", res.TookInMillis)
log.Println("查询到的总数", res.TotalHits())
var ttyp Blog
blogs := make([]Blog, 0)
for _, v := range res.Each(reflect.TypeOf(ttyp)) {
blog := v.(Blog)
blogs = append(blogs, blog)
}
if print {
log.Printf("**********Blogs*****************")
fmt.Println(blogs)
log.Println("*******************")
}
return nil
}
// QueryBlogNumbByAuthor term 查询-查询某个作者的作品有多少
func (es *esObj) QueryBlogNumbByAuthor(ctx context.Context, author string) error {
term := elastic.NewTermQuery("author", author)
res, err := es.db.Search().Index(es.index).Query(term).Do(ctx)
if err != nil {
log.Fatalf("QueryBlogNumbByAuthor fail ,author is %v, err is %v", author, err)
return err
}
log.Printf("author of %s all numb blog is %v", author, res.TotalHits())
return nil
}
// QueryBlogByContent 查询带有黄河关键字的博客
func (es *esObj) QueryBlogByContent(ctx context.Context, content string, print bool) (err error) {
match := elastic.NewMatchQuery("content", content)
res, err := es.db.Search().Index(es.index).Query(match).Do(ctx)
if err != nil {
log.Fatalf("es query fail when index is %s ,content is %v; match is %v; err is %v", es.index, content, match, err)
return err
}
log.Printf("sum numb is %d", res.TotalHits())
// 输出所有的内容
var bty Blog
blogs := make([]Blog, 0)
for _, v := range res.Each(reflect.TypeOf(bty)) {
blog := v.(Blog)
blogs = append(blogs, blog)
}
if print {
log.Printf("es打开 %+v", blogs)
}
return nil
}
func (es *esObj) BoolQueryBlogByAuthor(ctx context.Context, author string, print bool) {
termQuery := elastic.NewTermQuery("author", author)
rangeQuery := elastic.NewRangeQuery("release_time").Gte("0160").Lte("0172").Format("dd/MM/yyyy||yyyy")
boolQuery := elastic.NewBoolQuery().Must(termQuery, rangeQuery)
res, err := es.db.Search().Index(es.index).Query(boolQuery).Do(ctx)
if err != nil {
log.Fatalf("bool query err %v", err)
}
var bty Blog
blogs := make([]Blog, 0)
for _, v := range res.Each(reflect.TypeOf(bty)) {
blog := v.(Blog)
blogs = append(blogs, blog)
}
if print {
log.Printf("es 的 bool 查询结果 %+v", blogs)
}
}
// AggsSumPageViesByAuthorName 聚合索引 李白的书的总的阅读量
func (es *esObj) AggsSumPageViesByAuthorName(ctx context.Context, author string) (err error) {
query := elastic.NewTermQuery("author", author)
maxPageViews := elastic.NewMaxAggregation().Field("pageview")
totalPageViews := elastic.NewSumAggregation().Field("pageview")
res, err := es.db.Search().Index(es.index).Query(query).
Aggregation("totalPageViews", totalPageViews).
Aggregation("maxPageViews", maxPageViews).
Size(0).Pretty(true).Do(ctx)
if err != nil {
log.Fatalf("querySumPage fail , err is %v", err)
}
searchRes, find := res.Aggregations.ValueCount("totalPageViews")
if find {
log.Printf("totalPageVies of %s is %f", author, *searchRes.Value)
}
searchRes, find = res.Aggregations.ValueCount("maxPageViews")
if find {
log.Printf("maxPageViews of %s is %f", author, *searchRes.Value)
} else {
log.Printf("maxPageViews of %s not found", author)
}
return nil
}
// BucketAggs 桶聚合
func (es *esObj) BucketAggs(ctx context.Context, print bool) {
maxPageView := elastic.NewMaxAggregation().Field("pageview")
minPageView := elastic.NewMinAggregation().Field("pageview")
everyOneMaxMinPageView := elastic.NewTermsAggregation().Field("author").
SubAggregation("maxPageView", maxPageView).
SubAggregation("minPageView", minPageView)
res, err := es.db.Search(es.index).Query(elastic.NewMatchAllQuery()).
Aggregation("everyOneMaxMinPageView", everyOneMaxMinPageView).
Size(0).
Do(ctx)
if err != nil {
log.Fatalf("buck search fail， err is %v", err)
}
buckRes, find := res.Aggregations.Terms("everyOneMaxMinPageView")
if !find {
log.Fatalf("buck search not find")
}
if !print {
return
}
for _, buck := range buckRes.Buckets {
log.Println(buck.Key)
maxPageViewRes, _ := buck.ValueCount("maxPageView")
log.Printf("maxPageView is %v", *maxPageViewRes.Value)
minPageViewRes, _ := buck.ValueCount("minPageView")
log.Printf("minPageView is %v", *minPageViewRes.Value)
}
}
// 获取demo数据
func getBlogDemo() (blogs []Blog) {
blogs = append(blogs, Blog{
Title: "将进酒",
Author: "李白",
Content: `君不见，黄河之水天上来，奔流到海不复回。
君不见，高堂明镜悲白发，朝如青丝暮成雪。
人生得意须尽欢，莫使金樽空对月。
天生我材必有用，千金散尽还复来。
烹羊宰牛且为乐，会须一饮三百杯。
岑夫子，丹丘生，将进酒，杯莫停。
与君歌一曲，请君为我倾耳听。
钟鼓馔玉不足贵，但愿长醉不复醒。
古来圣贤皆寂寞，惟有饮者留其名。
陈王昔时宴平乐，斗酒十千恣欢谑。
主人何为言少钱，径须沽取对君酌。
五花马，千金裘，呼儿将出换美酒，与尔同销万古愁。`,
PageView: 200000,
ReleaseTime: time.Date(168, 1, 20, 0, 0, 0, 0, time.Local),
IsDel: false,
})
blogs = append(blogs, Blog{
Title: "春望",
Author: "杜甫",
Content: `国破山河在，城春草木深。
感时花溅泪，恨别鸟惊心。
烽火连三月，家书抵万金。
白头搔更短，浑欲不胜簪。`,
PageView: 120000,
ReleaseTime: time.Date(170, 1, 20, 0, 0, 0, 0, time.Local),
IsDel: false,
})
blogs = append(blogs, Blog{
Title: "静夜思",
Author: "李白",
Content: `床前明月光，疑是地上霜。
举头望明月，低头思故乡。`,
PageView: 1200000,
ReleaseTime: time.Date(172, 1, 20, 0, 0, 0, 0, time.Local),
IsDel: false,
})
blogs = append(blogs, Blog{
Title: "黄鹤楼送孟浩然之广陵",
Author: "李白",
Content: `故人西辞黄鹤楼，烟花三月下扬州。
孤帆远影碧空尽，唯见长江天际流。`,
PageView: 180000,
ReleaseTime: time.Date(171, 1, 20, 0, 0, 0, 0, time.Local),
IsDel: false,
})
blogs = append(blogs, Blog{
Title: "望岳",
Author: "杜甫",
Content: `岱宗夫如何？齐鲁青未了。
造化钟神秀，阴阳割昏晓。
荡胸生曾云，决眦入归鸟。
会当凌绝顶，一览众山小。`,
PageView: 190000,
ReleaseTime: time.Date(183, 1, 20, 0, 0, 0, 0, time.Local),
IsDel: false,
})
return blogs
}
使用原始语句查询 es
package main
import (
"context"
"crypto/tls"
"encoding/json"
"fmt"
es7 "github.com/elastic/go-elasticsearch/v7"
"github.com/elastic/go-elasticsearch/v7/esapi"
"net/http"
"strings"
)
type H map[string]any
type LH []H
type esService struct {
client *es7.Client
index []string
}
func main() {
es := NewEsClient()
fmt.Println(es)
es.QueryDemo()
}
func NewEsClient() *esService {
tr := &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
}
c := es7.Config{
//如果没有安全校验，测试环境，建议使用 http ,不要使用 https
Addresses: []string{"http://vm.ser.cn:9200"},
//Username: "elastic",
//Password: "tophant_017",
Transport: tr,
}
client, err := es7.NewClient(c)
if err != nil {
panic(err)
}
fmt.Println("esService 连接成功")
fmt.Println(client.Info())
return &esService{
client: client,
index: []string{"kibana_sample_data_logs"},
}
}
func (es *esService) QueryDemo() {
countQuery := H{
"query": H{
"match_all": H{},
},
"size": 10,
}
jsonByte, err := json.Marshal(countQuery)
if err != nil {
panic(err)
}
// go 转成 string 类型
jsonString := string(jsonByte)
fmt.Println(jsonString)
req := esapi.SearchRequest{
Index: es.index,
Body: strings.NewReader(jsonString),
Pretty: true, //结果美化
}
if err != nil {
panic(err)
}
ctx := context.Background()
res, err := req.Do(ctx, es.client)
if err != nil {
panic(err)
}
if res.IsError() {
fmt.Printf("链接失败 %s", res.Status())
fmt.Printf("错误原因字段话%s", res.String())
return
}
defer res.Body.Close()
fmt.Println(res)
}
最新文章
- Android 四大组件之再论BroadCast
- 浏览器请求URL原理
- tomcat下的https项目创建与部署
- Java 10大精华文章收集001
- （单选后，显示相对应的div）点击免费没有变化，点击收费出现输入框
- poj 3237 Tree 树链剖分
- JAVA与图形界面开发（Applet应用程序、AWT库、Swing）
- zookeeper作为soa服务器集群的协调调度服务器
- OSAL多任务资源分配机制
- 浅谈异步IO各模型优缺点
- C#使用log4net
- js基础-需要注意的地方
- ASP.NET vNext (一）- 基本概念和环境配置
- Zookeeper与Kafka集群搭建
- python 字典操作方法详解
- XML之XPath
- 已操作文件的方式，新建一个用户alex
- @Select注解的情况下，重载的报错
- 学习记录----简单的原生js路由
- 解决 scapy “NameError: global name 'wrpcap' is not defined” 错误