|
|
@@ -1,6 +1,7 @@
|
|
|
package elastic
|
|
|
|
|
|
import (
|
|
|
+ util "app.yhyue.com/data_processing/common_utils"
|
|
|
"context"
|
|
|
"errors"
|
|
|
"fmt"
|
|
|
@@ -157,16 +158,18 @@ func (e *Elastic) Close() {
|
|
|
// }
|
|
|
//}
|
|
|
|
|
|
-func (e *Elastic) BulkSave(index string, obj map[string]interface{}) {
|
|
|
+func (e *Elastic) BulkSave(index string, obj []map[string]interface{}) {
|
|
|
client := e.GetEsConn()
|
|
|
defer e.DestoryEsConn(client)
|
|
|
if client != nil {
|
|
|
req := client.Bulk()
|
|
|
- for k, v := range obj {
|
|
|
+ for _, v := range obj {
|
|
|
//if isDelBefore {
|
|
|
// req = req.Add(es.NewBulkDeleteRequest().Index(index).Id(fmt.Sprintf("%v", v["_id"])))
|
|
|
//}
|
|
|
- req = req.Add(es.NewBulkIndexRequest().Index(index).Id(k).Doc(v))
|
|
|
+ id := util.ObjToString(v["_id"])
|
|
|
+ delete(v, "_id")
|
|
|
+ req = req.Add(es.NewBulkIndexRequest().Index(index).Id(id).Doc(v))
|
|
|
}
|
|
|
_, err := req.Do(context.Background())
|
|
|
if err != nil {
|
