|
|
@@ -5,11 +5,14 @@ import (
|
|
|
"encoding/json"
|
|
|
"fmt"
|
|
|
"github.com/olivere/elastic/v7"
|
|
|
+ "github.com/robfig/cron/v3"
|
|
|
+ "github.com/spf13/viper"
|
|
|
"io"
|
|
|
util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
|
|
|
es "jygit.jydev.jianyu360.cn/data_processing/common_utils/elastic"
|
|
|
"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
|
|
|
"log"
|
|
|
+ "sync"
|
|
|
"time"
|
|
|
)
|
|
|
|
|
|
@@ -18,8 +21,13 @@ var (
|
|
|
saveEsSp = make(chan bool, 5)
|
|
|
EsBulkSize = 50
|
|
|
Es *es.Elastic
|
|
|
- portraitIndex = "project_portrait" // 画像索引
|
|
|
- portraitMgo = "wcc_project_portrait" // MongoDB 的表名
|
|
|
+ Mgo *mongodb.MongodbSim //bidding 地址
|
|
|
+ MgoP *mongodb.MongodbSim // 项目地址
|
|
|
+ portraitIndex = "" // 画像索引
|
|
|
+ portraitMgo = "" // MongoDB 的表名
|
|
|
+ GF GlobalConf
|
|
|
+ // 情报分类一级标签
|
|
|
+ topInfos = []string{"情报_物业", "情报_环境采购", "情报_印务商机", "情报_家具招投标", "情报_车辆租赁"}
|
|
|
)
|
|
|
|
|
|
type PortraitData struct {
|
|
|
@@ -31,50 +39,159 @@ type PortraitData struct {
|
|
|
Lasttime int64 `json:"lasttime"`
|
|
|
ProjectCount int64 `json:"project_count"`
|
|
|
ProjectMoney float64 `json:"project_money"`
|
|
|
+ UpdateTime int64 `json:"update_time"`
|
|
|
+}
|
|
|
+
|
|
|
+func InitConfig() (err error) {
|
|
|
+ viper.SetConfigFile("config.toml") // 指定配置文件路径
|
|
|
+ viper.SetConfigName("config") // 配置文件名称(无扩展名)
|
|
|
+ viper.SetConfigType("toml") // 如果配置文件的名称中没有扩展名，则需要配置此项
|
|
|
+
|
|
|
+ viper.AddConfigPath("./")
|
|
|
+ viper.AddConfigPath("./conf/") // 还可以在工作目录中查找配置
|
|
|
+ viper.AddConfigPath("../conf/") // 还可以在工作目录中查找配置
|
|
|
+ err = viper.ReadInConfig() // 查找并读取配置文件
|
|
|
+ if err != nil { // 处理读取配置文件的错误
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ err = viper.Unmarshal(&GF)
|
|
|
+
|
|
|
+ return err
|
|
|
+
|
|
|
}
|
|
|
|
|
|
func Init() {
|
|
|
+ InitConfig()
|
|
|
+ // 正式环境
|
|
|
+ Es = &es.Elastic{
|
|
|
+ //S_esurl: "http://127.0.0.1:19908",
|
|
|
+ S_esurl: GF.Es.URL,
|
|
|
+ I_size: 5,
|
|
|
+ Username: GF.Es.Username,
|
|
|
+ Password: GF.Es.Password,
|
|
|
+ }
|
|
|
+ Es.InitElasticSize()
|
|
|
+ //测试环境
|
|
|
//Es = &es.Elastic{
|
|
|
- // //S_esurl: "http://127.0.0.1:19908",
|
|
|
- // S_esurl: "http://172.17.4.184:19908",
|
|
|
+ // S_esurl: "http://192.168.3.149:9201",
|
|
|
// I_size: 5,
|
|
|
- // Username: "jybid",
|
|
|
- // Password: "Top2023_JEB01i@31",
|
|
|
+ // Username: "",
|
|
|
+ // Password: "",
|
|
|
//}
|
|
|
//Es.InitElasticSize()
|
|
|
- //测试环境
|
|
|
- Es = &es.Elastic{
|
|
|
- S_esurl: "http://192.168.3.149:9201",
|
|
|
- I_size: 5,
|
|
|
- Username: "",
|
|
|
- Password: "",
|
|
|
+
|
|
|
+ // bidding 地址
|
|
|
+ Mgo = &mongodb.MongodbSim{
|
|
|
+ MongodbAddr: GF.Mongo.Host,
|
|
|
+ //MongodbAddr: "127.0.0.1:27083",
|
|
|
+ Size: 10,
|
|
|
+ DbName: GF.Mongo.DB,
|
|
|
+ UserName: GF.Mongo.Username,
|
|
|
+ Password: GF.Mongo.Password,
|
|
|
+ Direct: GF.Mongo.Direct,
|
|
|
}
|
|
|
- Es.InitElasticSize()
|
|
|
+ Mgo.InitPool()
|
|
|
+
|
|
|
+ // 抽取库项目表
|
|
|
+ MgoP = &mongodb.MongodbSim{
|
|
|
+ MongodbAddr: GF.Mongop.Host,
|
|
|
+ Size: 10,
|
|
|
+ DbName: GF.Mongop.DB,
|
|
|
+ UserName: GF.Mongop.Username,
|
|
|
+ Password: GF.Mongop.Password,
|
|
|
+ Direct: GF.Mongop.Direct,
|
|
|
+ }
|
|
|
+
|
|
|
+ MgoP.InitPool()
|
|
|
+
|
|
|
+ portraitIndex = GF.Env.PortraitIndex
|
|
|
+ portraitMgo = GF.Env.PortraitMgo
|
|
|
+ if portraitIndex == "" || portraitMgo == "" {
|
|
|
+ log.Fatalln("画像索引或者MongoDB数据表为空")
|
|
|
+ }
|
|
|
+
|
|
|
}
|
|
|
func main() {
|
|
|
Init()
|
|
|
- go SaveEsMethod()
|
|
|
- //dealAllData()
|
|
|
- allDataEs() //处理存量数据到es
|
|
|
+ go SaveEsMethod() // 生索引
|
|
|
+
|
|
|
+ //定时任务
|
|
|
+ local, _ := time.LoadLocation("Asia/Shanghai")
|
|
|
+ c := cron.New(cron.WithLocation(local), cron.WithSeconds())
|
|
|
+ _, err := c.AddFunc(GF.Env.Spec, dealIncData)
|
|
|
+ if err != nil {
|
|
|
+ log.Println("AddFunc err", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ c.Start()
|
|
|
+ defer c.Stop()
|
|
|
|
|
|
- ch := make(chan bool, 1)
|
|
|
- <-ch
|
|
|
+ select {}
|
|
|
}
|
|
|
|
|
|
// dealIncData 处理增量数据
|
|
|
func dealIncData() {
|
|
|
-
|
|
|
now := time.Now()
|
|
|
- yesterday := time.Date(now.Year(), now.Month(), now.Day()-1, 0, 0, 0, 0, now.Location())
|
|
|
+ yesterday := time.Date(now.Year(), now.Month(), now.Day()+GF.Env.Days, 0, 0, 0, 0, now.Location())
|
|
|
today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
|
|
|
- q := map[string]interface{}{
|
|
|
+ where := map[string]interface{}{
|
|
|
"pici": map[string]interface{}{
|
|
|
"$gt": yesterday.Unix(),
|
|
|
"$lte": today.Unix(),
|
|
|
},
|
|
|
}
|
|
|
- log.Println(q)
|
|
|
|
|
|
+ url := GF.Es.URL
|
|
|
+ username := GF.Es.Username
|
|
|
+ password := GF.Es.Password
|
|
|
+
|
|
|
+ // 创建 Elasticsearch 客户端
|
|
|
+ client, err := elastic.NewClient(
|
|
|
+ elastic.SetURL(url),
|
|
|
+ elastic.SetBasicAuth(username, password),
|
|
|
+ elastic.SetSniff(false),
|
|
|
+ )
|
|
|
+ if err != nil {
|
|
|
+ log.Fatalf("创建 Elasticsearch 客户端失败：%s", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ defer util.Catch()
|
|
|
+ sess := MgoP.GetMgoConn()
|
|
|
+ defer MgoP.DestoryMongoConn(sess)
|
|
|
+
|
|
|
+ var wg sync.WaitGroup
|
|
|
+ ch := make(chan map[string]interface{}, 10000)
|
|
|
+
|
|
|
+ // 并行处理结果
|
|
|
+ for i := 0; i < 2; i++ {
|
|
|
+ wg.Add(1)
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ for hit := range ch {
|
|
|
+ processProjectData(hit, client, Mgo)
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ }
|
|
|
+
|
|
|
+ count := 0
|
|
|
+ it := sess.DB(GF.Mongop.DB).C(GF.Mongop.Coll).Find(where).Select(nil).Iter()
|
|
|
+ for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
|
|
|
+ if count%5000 == 0 {
|
|
|
+ log.Println("current:", count, tmp["buyer"])
|
|
|
+ }
|
|
|
+ // 没有采购单位
|
|
|
+ if util.ObjToString(tmp["buyer"]) == "" {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ ch <- tmp
|
|
|
+ }
|
|
|
+
|
|
|
+ close(ch) // 关闭通道
|
|
|
+ wg.Wait()
|
|
|
+ //处理完毕生索引
|
|
|
+ incDataEs()
|
|
|
+ log.Println("增量数据处理完毕")
|
|
|
}
|
|
|
|
|
|
// dealAllData 处理存量数据，
|
|
|
@@ -109,10 +226,6 @@ func dealAllData() {
|
|
|
}
|
|
|
MgoB.InitPool()
|
|
|
|
|
|
- //query := elastic.NewBoolQuery().
|
|
|
- // Must(elastic.NewTermQuery("toptype", "结果")).
|
|
|
- // Must(elastic.NewTermQuery("subtype", "招标"))
|
|
|
-
|
|
|
ctx := context.Background()
|
|
|
//开始滚动搜索
|
|
|
scrollID := ""
|
|
|
@@ -229,34 +342,283 @@ func dealAllData() {
|
|
|
fmt.Println("结束~~~~~~~~~~~~~~~")
|
|
|
}
|
|
|
|
|
|
+// dealAllDataB 处理存量数据，协程处理全部类型
|
|
|
+func dealAllDataB() {
|
|
|
+ url := "http://172.17.4.184:19908"
|
|
|
+ username := "jybid"
|
|
|
+ password := "Top2023_JEB01i@31"
|
|
|
+ index := "buyer"
|
|
|
+
|
|
|
+ // 创建 Elasticsearch 客户端
|
|
|
+ client, err := elastic.NewClient(
|
|
|
+ elastic.SetURL(url),
|
|
|
+ elastic.SetBasicAuth(username, password),
|
|
|
+ elastic.SetSniff(false),
|
|
|
+ )
|
|
|
+ if err != nil {
|
|
|
+ log.Fatalf("创建 Elasticsearch 客户端失败：%s", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ MgoB := &mongodb.MongodbSim{
|
|
|
+ MongodbAddr: "172.17.189.140:27080",
|
|
|
+ Size: 10,
|
|
|
+ DbName: "qfw",
|
|
|
+ UserName: "SJZY_RWbid_ES",
|
|
|
+ Password: "SJZY@B4i4D5e6S",
|
|
|
+ }
|
|
|
+ MgoB.InitPool()
|
|
|
+
|
|
|
+ ctx := context.Background()
|
|
|
+ scroll := "10m"
|
|
|
+ searchSource := elastic.NewSearchSource().
|
|
|
+ Size(10000).
|
|
|
+ Sort("_doc", true)
|
|
|
+
|
|
|
+ searchService := client.Scroll(index).
|
|
|
+ Size(10000).
|
|
|
+ Scroll(scroll).
|
|
|
+ SearchSource(searchSource)
|
|
|
+
|
|
|
+ res, err := searchService.Do(ctx)
|
|
|
+ if err != nil {
|
|
|
+ if err == io.EOF {
|
|
|
+ fmt.Println("没有数据")
|
|
|
+ return
|
|
|
+ } else {
|
|
|
+ panic(err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ fmt.Println("总数是：", res.TotalHits())
|
|
|
+ var wg sync.WaitGroup
|
|
|
+ ch := make(chan *elastic.SearchHit, 10000)
|
|
|
+
|
|
|
+ // 并行处理结果
|
|
|
+ for i := 0; i < 10; i++ {
|
|
|
+ wg.Add(1)
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ for hit := range ch {
|
|
|
+ processHit(hit, client, MgoB)
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ }
|
|
|
+
|
|
|
+ num := 0
|
|
|
+ for len(res.Hits.Hits) > 0 {
|
|
|
+ for _, hit := range res.Hits.Hits {
|
|
|
+ num++
|
|
|
+ ch <- hit
|
|
|
+ if num%1000 == 0 {
|
|
|
+ log.Println("current num:", num)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ scrollID := res.ScrollId
|
|
|
+ res, err = client.Scroll().ScrollId(scrollID).Scroll(scroll).Do(ctx)
|
|
|
+ if err != nil {
|
|
|
+ if err == io.EOF {
|
|
|
+ close(ch)
|
|
|
+ break
|
|
|
+ }
|
|
|
+ log.Println("滚动搜索失败：", err, res)
|
|
|
+ close(ch)
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ wg.Wait()
|
|
|
+
|
|
|
+ // 在循环外调用 ClearScroll
|
|
|
+ _, err = client.ClearScroll().ScrollId(res.ScrollId).Do(ctx)
|
|
|
+ if err != nil {
|
|
|
+ log.Printf("清理滚动搜索失败：%s", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ fmt.Println("结束~~~~~~~~~~~~~~~")
|
|
|
+}
|
|
|
+
|
|
|
+func processHit(hit *elastic.SearchHit, client *elastic.Client, MgoB *mongodb.MongodbSim) {
|
|
|
+ var doc map[string]interface{}
|
|
|
+ err := json.Unmarshal(hit.Source, &doc)
|
|
|
+ if err != nil {
|
|
|
+ log.Printf("解析文档失败：%s", err)
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ for _, v := range topInfos {
|
|
|
+ portrait := PortraitData{
|
|
|
+ Buyer: util.ObjToString(doc["name"]),
|
|
|
+ BusinessType: getStr(util.ObjToString(doc["buyerclass"])),
|
|
|
+ Class: v,
|
|
|
+ }
|
|
|
+
|
|
|
+ query := elastic.NewBoolQuery().
|
|
|
+ Must(
|
|
|
+ elastic.NewTermQuery("buyer", util.ObjToString(doc["name"])),
|
|
|
+ elastic.NewTermQuery("tag_topinformation", v),
|
|
|
+ )
|
|
|
+ searchService2 := client.Search().
|
|
|
+ Index("projectset").
|
|
|
+ Query(query).
|
|
|
+ Sort("lasttime", false).
|
|
|
+ Size(1).
|
|
|
+ Aggregation("total_price", elastic.NewSumAggregation().Field("sortprice"))
|
|
|
+
|
|
|
+ searchResult, err := searchService2.Do(context.Background())
|
|
|
+ if err != nil {
|
|
|
+ log.Fatalf("Error getting response: %s", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ if searchResult.Hits.TotalHits.Value > 0 {
|
|
|
+ portrait.ProjectCount = searchResult.TotalHits()
|
|
|
+ for _, hit := range searchResult.Hits.Hits {
|
|
|
+ var doc2 map[string]interface{}
|
|
|
+ err := json.Unmarshal(hit.Source, &doc2)
|
|
|
+ if err != nil {
|
|
|
+ log.Printf("解析文档失败：%s", err)
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ portrait.Lasttime = util.Int64All(doc2["lasttime"])
|
|
|
+ portrait.Area = util.ObjToString(doc2["area"])
|
|
|
+ portrait.City = util.ObjToString(doc2["city"])
|
|
|
+ }
|
|
|
+ if agg, found := searchResult.Aggregations.Sum("total_price"); found {
|
|
|
+ portrait.ProjectMoney = *agg.Value
|
|
|
+ } else {
|
|
|
+ log.Println("Aggregation not found")
|
|
|
+ }
|
|
|
+ MgoB.Save(portraitMgo, structToMap(portrait))
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// processProjectData 处理项目表增量数据
|
|
|
+func processProjectData(tmp map[string]interface{}, client *elastic.Client, MgoB *mongodb.MongodbSim) {
|
|
|
+ for _, v := range topInfos {
|
|
|
+ existsWhere := map[string]interface{}{
|
|
|
+ "buyer": tmp["buyer"],
|
|
|
+ "class": v,
|
|
|
+ }
|
|
|
+ portrait := PortraitData{
|
|
|
+ Buyer: util.ObjToString(tmp["buyer"]),
|
|
|
+ BusinessType: getStr(util.ObjToString(tmp["buyerclass"])),
|
|
|
+ Class: v,
|
|
|
+ }
|
|
|
+
|
|
|
+ query := elastic.NewBoolQuery().
|
|
|
+ Must(
|
|
|
+ elastic.NewTermQuery("buyer", util.ObjToString(tmp["buyer"])),
|
|
|
+ elastic.NewTermQuery("tag_topinformation", v),
|
|
|
+ )
|
|
|
+ searchService2 := client.Search().
|
|
|
+ Index("projectset").
|
|
|
+ Query(query).
|
|
|
+ Sort("lasttime", false).
|
|
|
+ Size(1).
|
|
|
+ Aggregation("total_price", elastic.NewSumAggregation().Field("sortprice"))
|
|
|
+
|
|
|
+ searchResult, err := searchService2.Do(context.Background())
|
|
|
+ if err != nil {
|
|
|
+ log.Fatalf("Error getting response: %s", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ if searchResult.Hits.TotalHits.Value > 0 {
|
|
|
+ portrait.ProjectCount = searchResult.TotalHits()
|
|
|
+ for _, hit := range searchResult.Hits.Hits {
|
|
|
+ var doc2 map[string]interface{}
|
|
|
+ err = json.Unmarshal(hit.Source, &doc2)
|
|
|
+ if err != nil {
|
|
|
+ log.Printf("解析文档失败：%s", err)
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ portrait.Lasttime = util.Int64All(doc2["lasttime"])
|
|
|
+ portrait.Area = util.ObjToString(doc2["area"])
|
|
|
+ portrait.City = util.ObjToString(doc2["city"])
|
|
|
+ }
|
|
|
+ if agg, found := searchResult.Aggregations.Sum("total_price"); found {
|
|
|
+ portrait.ProjectMoney = *agg.Value
|
|
|
+ } else {
|
|
|
+ log.Println("Aggregation not found")
|
|
|
+ }
|
|
|
+
|
|
|
+ portrait.UpdateTime = time.Now().Unix()
|
|
|
+ //
|
|
|
+ exist, _ := MgoB.FindOne(portraitMgo, existsWhere)
|
|
|
+ if exist != nil && len(*exist) > 0 {
|
|
|
+ // 存在只更新部分内容
|
|
|
+ update := map[string]interface{}{
|
|
|
+ "lasttime": portrait.Lasttime,
|
|
|
+ "project_money": portrait.ProjectMoney,
|
|
|
+ "project_count": portrait.ProjectCount,
|
|
|
+ "update_time": portrait.UpdateTime,
|
|
|
+ }
|
|
|
+ if util.Int64All((*exist)["project_count"]) != portrait.ProjectCount || util.Int64All((*exist)["lasttime"]) != portrait.Lasttime || util.Float64All((*exist)["project_money"]) != portrait.ProjectMoney {
|
|
|
+ id := mongodb.BsonIdToSId((*exist)["_id"])
|
|
|
+ MgoB.UpdateById(portraitMgo, id, map[string]interface{}{"$set": update})
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ //不存在直接保存
|
|
|
+ MgoB.Save(portraitMgo, structToMap(portrait))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// incDataEs 增量数据处理生索引
|
|
|
+func incDataEs() {
|
|
|
+ now := time.Now()
|
|
|
+ today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
|
|
|
+ where := map[string]interface{}{
|
|
|
+ "update_time": map[string]interface{}{
|
|
|
+ "$gte": today.Unix(),
|
|
|
+ },
|
|
|
+ }
|
|
|
+
|
|
|
+ defer util.Catch()
|
|
|
+ sess := Mgo.GetMgoConn()
|
|
|
+ defer Mgo.DestoryMongoConn(sess)
|
|
|
+ count := 0
|
|
|
+ it := sess.DB("qfw").C(portraitMgo).Find(where).Select(nil).Iter()
|
|
|
+ for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
|
|
|
+ if count%1000 == 0 {
|
|
|
+ log.Println("current:", count)
|
|
|
+ }
|
|
|
+ id := mongodb.BsonIdToSId(tmp["_id"])
|
|
|
+ tmp["id"] = id
|
|
|
+ tmp["_id"] = id
|
|
|
+ delete(tmp, "update_time")
|
|
|
+ saveEsPool <- tmp
|
|
|
+ }
|
|
|
+ log.Println("数据处理完毕")
|
|
|
+}
|
|
|
+
|
|
|
// allDataEs 处理存量数据到es
|
|
|
func allDataEs() {
|
|
|
- //MgoB := &mongodb.MongodbSim{
|
|
|
- // MongodbAddr: "172.17.189.140:27080",
|
|
|
- // //MongodbAddr: "127.0.0.1:27083",
|
|
|
- // Size: 10,
|
|
|
- // DbName: "qfw",
|
|
|
- // UserName: "SJZY_RWbid_ES",
|
|
|
- // Password: "SJZY@B4i4D5e6S",
|
|
|
- // //Direct: true,
|
|
|
- //}
|
|
|
- //MgoB.InitPool()
|
|
|
- //
|
|
|
MgoB := &mongodb.MongodbSim{
|
|
|
- MongodbAddr: "192.168.3.206:27002",
|
|
|
- Size: 10,
|
|
|
- DbName: "qfw_data",
|
|
|
- UserName: "root",
|
|
|
- Password: "root",
|
|
|
+ MongodbAddr: "172.17.189.140:27080",
|
|
|
+ //MongodbAddr: "127.0.0.1:27083",
|
|
|
+ Size: 10,
|
|
|
+ DbName: "qfw",
|
|
|
+ UserName: "SJZY_RWbid_ES",
|
|
|
+ Password: "SJZY@B4i4D5e6S",
|
|
|
//Direct: true,
|
|
|
}
|
|
|
MgoB.InitPool()
|
|
|
+ // 测试环境
|
|
|
+ //MgoB := &mongodb.MongodbSim{
|
|
|
+ // MongodbAddr: "192.168.3.206:27002",
|
|
|
+ // Size: 10,
|
|
|
+ // DbName: "qfw_data",
|
|
|
+ // UserName: "root",
|
|
|
+ // Password: "root",
|
|
|
+ // //Direct: true,
|
|
|
+ //}
|
|
|
+ //MgoB.InitPool()
|
|
|
|
|
|
defer util.Catch()
|
|
|
sess := MgoB.GetMgoConn()
|
|
|
defer MgoB.DestoryMongoConn(sess)
|
|
|
count := 0
|
|
|
- it := sess.DB("qfw_data").C(portraitMgo).Find(nil).Select(nil).Iter()
|
|
|
+ it := sess.DB(MgoB.DbName).C(portraitMgo).Find(nil).Select(nil).Iter()
|
|
|
for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
|
|
|
if count%5000 == 0 {
|
|
|
log.Println("current:", count, tmp["_id"])
|
|
|
@@ -266,7 +628,6 @@ func allDataEs() {
|
|
|
tmp["_id"] = id
|
|
|
|
|
|
saveEsPool <- tmp
|
|
|
- //Es.Save("project_portrait", tmp)
|
|
|
}
|
|
|
log.Println("数据处理完毕")
|
|
|
}
|
wcc