未能在elasticsearch Olivere包中插入大容量数据

Question

我正在尝试插入批量数据。我没有得到任何错误，除了响应属性错误：true。这是我的代码片段。

func BulkInsert(ctx context.Context, products []model.Pt) error {
Counter++
// create new bulk processor from client
bulkProcessor, err := ElasticVar.BulkProcessor().
Name("Bulk Worker 1").
Workers(5).
BulkActions(1000).
BulkSize(2 << 20).
FlushInterval(1 * time.Second).
Stats(true).
After(after).
Do(context.Background())
if err != nil {
log.Println("NewBulkProcessorService err", err)
return err
}
defer bulkProcessor.Close()
storeIds := map[string]string{}
// Enqueue the document
for _, product := range products {
if _, ok := storeIds[product.Id]; ok {
continue
} else {
storeIds[product.StoreId] = product.StoreId
}
dataJSON, err := json.Marshal(product)
if err != nil {
// Look up the failed documents with res.Failed(), and e.g. recommit
log.Println("elastic marshal failed", err)
return errors.New("elastic marshal failed")
}
bulkProcessor.Add(elastic.NewBulkIndexRequest().
OpType("index").
Index(Index).
Type(Type).
Id(product.Id).
Doc(string(dataJSON)))
}
stats := bulkProcessor.Stats()
fmt.Printf("Number of times flush has been invoked: %dn", stats.Flushed)
fmt.Printf("Number of times workers committed reqs: %dn", stats.Committed)
fmt.Printf("Number of requests indexed            : %dn", stats.Indexed)
fmt.Printf("Number of requests reported as created: %dn", stats.Created)
fmt.Printf("Number of requests reported as updated: %dn", stats.Updated)
fmt.Printf("Number of requests reported as success: %dn", stats.Succeeded)
fmt.Printf("Number of requests reported as failed : %dn", stats.Failed)
fmt.Printf("nn")
for i, w := range stats.Workers {
fmt.Printf("Worker %d: Number of requests queued: %dn", i, w.Queued)
fmt.Printf("           Last response time       : %vn", w.LastDuration)
}
fmt.Printf(`Inserted Count: %d`, Counter)
return nil
}
func after(executionID int64, requests []elastic.BulkableRequest, response *elastic.BulkResponse, err error) {
if err != nil {
log.Printf("bulk commit failed, err: %vn", err)
}
// do what ever you want in case bulk commit success
log.Printf("commit response=%vn", response.Errors)
log.Printf("commit successfully, len(requests)=%dn", len(requests))
}

我正在批量添加所有数据。

在回调之后的响应。错误是真的。当我查询数据时。没有插入任何记录。请帮忙。

Answer 1

我能够通过创建访问错误字符串来调试它

for _, info := range response.Created() {
fmt.Println("nBulk response created error:", info.Error)
fmt.Println("nBulk response created: error reason4", info.Error.Reason)
fmt.Println("nBulk response IndexProduct:", info.IndexProduct)
}

问题是我通过了Type="；产品"；而不是Type＝"0"；doc"；