Оптимизируйте запись в CSV в Go

Question

Следующий фрагмент проверяет номер телефона и записывает данные в CSV.

func Parse(phone Input, output *PhoneNumber) error {
    var n PhoneNumber
    num, _ := phonenumbers.Parse(phone.Number, phone.Prefix)
    n.PhoneNumber = phonenumbers.Format(num, phonenumbers.E164)
    n.CountryCode = num.GetCountryCode()
    n.PhoneType = phonenumbers.GetNumberType(num)
    n.NetworkName, _ = phonenumbers.GetCarrierForNumber(num, "EN")
    n.Region = phonenumbers.GetRegionCodeForNumber(num)
    *output = n
    return nil
}

func createFile(path string) {
    // detect if file exists
    var _, err = os.Stat(path)

    // create file if not exists
    if os.IsNotExist(err) {
        var file, err = os.Create(path)
        if err != nil {
            return
        }
        defer file.Close()
    }
}

func worker(ctx context.Context, dst chan string, src chan []string) {
    for {
        select {
        case dataArray, ok := <-src: // you must check for readable state of the channel.
            if !ok {
                return
            }
            go processNumber(dataArray[0])
        case <-ctx.Done(): // if the context is cancelled, quit.
            return
        }
    }
}

func processNumber(number string) {
    num, e := phonenumbers.Parse(number, "")
    if e != nil {
        return
    }
    region := phonenumbers.GetRegionCodeForNumber(num)
    carrier, _ := phonenumbers.GetCarrierForNumber(num, "EN")
    path := "sample_all.csv"
    createFile(path)
    var csvFile, _ = os.OpenFile(path, os.O_APPEND|os.O_WRONLY, os.ModeAppend)
    csvwriter := csv.NewWriter(csvFile)
    _ = csvwriter.Write([]string{phonenumbers.Format(num, phonenumbers.E164), fmt.Sprintf("%v", num.GetCountryCode()), fmt.Sprintf("%v", phonenumbers.GetNumberType(num)), carrier, region})

    defer csvFile.Close()

    csvwriter.Flush()
}

func ParseFile(phone Input, output *PhoneNumber) error {

    // create a context
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    // that cancels at ctrl+C
    go onSignal(os.Interrupt, cancel)

    numberOfWorkers := 2

    start := time.Now()

    csvfile, err := os.Open(phone.File)
    if err != nil {
        log.Fatal(err)
    }
    defer csvfile.Close()

    reader := csv.NewReader(csvfile)

    // create the pair of input/output channels for the controller=>workers com.
    src := make(chan []string)
    out := make(chan string)

    // use a waitgroup to manage synchronization
    var wg sync.WaitGroup

    // declare the workers
    for i := 0; i < numberOfWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            worker(ctx, out, src)
        }()
    }

    // read the csv and write it to src
    go func() {
        for {
            record, err := reader.Read()
            if err == io.EOF {
                break
            } else if err != nil {
                log.Fatal(err)
            }
            src <- record // you might select on ctx.Done().
        }
        close(src) // close src to signal workers that no more job are incoming.
    }()

    // wait for worker group to finish and close out
    go func() {
        wg.Wait()  // wait for writers to quit.
        close(out) // when you close(out) it breaks the below loop.
    }()

    // drain the output
    for res := range out {
        fmt.Println(res)
    }

    fmt.Printf("\n%2fs", time.Since(start).Seconds())
    return nil
}

В функции processNumber, если я пропущу запись в CSV, процесс проверки номера завершится за 6 секунд, но запись одной записи за раз в CSV увеличит потребление времени до 15 с.

Как я могу оптимизировать код? Могу ли я разделить записи на части и записать их порциями вместо записи по одной строке за раз?

Answer 1

Работать непосредственно в рабочей рутине вместо того, чтобы запускать рутину для каждой задачи.

Открыть файл вывода файла один раз. Flu sh выходной файл один раз.

func worker(ctx context.Context, dst chan []string, src chan []string) {
    for {
        select {
        case dataArray, ok := <-src: // you must check for readable state of the channel.
            if !ok {
                return
            }
            dst <- processNumber(dataArray[0])
        case <-ctx.Done(): // if the context is cancelled, quit.
            return
        }
    }
}

func processNumber(number string) []string {
    num, e := phonenumbers.Parse(number, "")
    if e != nil {
        return
    }
    region := phonenumbers.GetRegionCodeForNumber(num)
    carrier, _ := phonenumbers.GetCarrierForNumber(num, "EN")
    return []string{phonenumbers.Format(num, phonenumbers.E164), fmt.Sprintf("%v", num.GetCountryCode()), fmt.Sprintf("%v", phonenumbers.GetNumberType(num)), carrier, region}
}

func ParseFile(phone Input, output *PhoneNumber) error {

    // create a context
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    // that cancels at ctrl+C
    go onSignal(os.Interrupt, cancel)

    numberOfWorkers := 2

    start := time.Now()

    csvfile, err := os.Open(phone.File)
    if err != nil {
        log.Fatal(err)
    }
    defer csvfile.Close()

    reader := csv.NewReader(csvfile)

    // create the pair of input/output channels for the controller=>workers com.
    src := make(chan []string)
    out := make(chan string)

    // use a waitgroup to manage synchronization
    var wg sync.WaitGroup

    // declare the workers
    for i := 0; i < numberOfWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            worker(ctx, out, src)
        }()
    }

    // read the csv and write it to src
    go func() {
        for {
            record, err := reader.Read()
            if err == io.EOF {
                break
            } else if err != nil {
                log.Fatal(err)
            }
            src <- record // you might select on ctx.Done().
        }
        close(src) // close src to signal workers that no more job are incoming.
    }()

    // wait for worker group to finish and close out
    go func() {
        wg.Wait()  // wait for writers to quit.
        close(out) // when you close(out) it breaks the below loop.
    }()

    path := "sample_all.csv"

    file, err := os.Create(path)
    if err != nil {
        return err
    }
    defer file.Close()
    csvwriter := csv.NewWriter(csvFile)

    // drain the output
    for res := range out {
        csvwriter.Write(res)
    }
    csvwriter.Flush()

    fmt.Printf("\n%2fs", time.Since(start).Seconds())
    return nil
}