es_index_data/createEsIndex/bidding_del.go

package main

import (
	"esindex/config"
	"go.mongodb.org/mongo-driver/bson"
	"go.uber.org/zap"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
)

//根据bidding表extracttype=-1，删除es中重复数据
func biddingDelByExtracttype(data []byte, mapInfo map[string]interface{}) {
	defer util.Catch()
	q, _ := mapInfo["query"].(map[string]interface{})
	if q == nil {
		q = map[string]interface{}{
			"_id": bson.M{
				"$gt":  mongodb.StringTOBsonId(mapInfo["gtid"].(string)),
				"$lte": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
			},
			"extracttype": -1,
		}
	}

	//bidding
	conn := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(conn)
	//查询数据
	biddingquery := conn.DB(MgoB.DbName).C(config.Conf.DB.MongoB.Coll).Find(q).Select(
		bson.M{"_id": 1},
	).Sort("_id").Iter()
	log.Info("开始迭代...")
	i := 0
	var n int
	var dnum int
	for tmp := make(map[string]interface{}); biddingquery.Next(tmp); i = i + 1 {
		n++
		_id := mongodb.BsonIdToSId(tmp["_id"])
		if Es.DelById(config.Conf.DB.Es.IndexB, _id) { //删除
			dnum++
		}
		if n%200 == 0 {
			log.Info("current", zap.Int("当前:", n))
		}
		tmp = make(map[string]interface{})
	}
	log.Info("over", zap.Int("共删除：", dnum))
}