bulkprocessor.go

package main

import (
        "fmt"
        "github.com/olivere/elastic"
        "strconv"
        "sync"
        "time"
)

var (
        client    *elastic.Client
        processor *elastic.BulkProcessor
)

type User struct {
        FirstName string
        LastName  string
        Bio       string
}

func init() {
        client, err := elastic.NewClient()
        if err != nil {
                panic(err)
        }

        processor, err = client.BulkProcessor().
                Name("UserWorker").
                Workers(8).
                BulkActions(10000).
                Do()
        if err != nil {
                panic(err)
        }
}

func main() {
        fmt.Println("")
        startTime := time.Now()
        var wg sync.WaitGroup

        wc := 8

        wg.Add(wc)

        for x := 0; x < wc; x++ {
                go func(x int) {
                        for i := 0; i < (1000000 / wc); i++ {
                                r := elastic.NewBulkIndexRequest().Index("users").Type("user").Doc(&User{
                                        FirstName: "User",
                                        LastName:  strconv.Itoa(i),
                                        Bio:       "Noodles and company",
                                })

                                processor.Add(r)
                        }

                        wg.Done()
                        fmt.Println(x, "is done building request")
                }(x)
        }

        wg.Wait()
        fmt.Println("Done building all requests.")

        processor.Close()
        endTime := time.Now()

        fmt.Printf("Done. %s\n", endTime.Sub(startTime))

        time.Sleep(time.Second)
        fmt.Println("")
}