Arrow CSV 读取器在同一文件上返回多个 null 值

在IT行业这个发展更新速度很快的行业，只有不停止的学习，才不会被行业所淘汰。如果你是Golang学习者，那么本文《Arrow CSV 读取器在同一文件上返回多个 null 值》就很适合你！本篇内容主要包括##content_title##，希望对大家的知识积累有所帮助，助力实战开发！

我正在尝试使用多个 goroutine 读取同一个文件，其中每个 goroutine 都分配了一个字节来开始读取，并分配了一些行来读取 linelimit。

通过将 csv.chunksize 选项设置为 chunksize 变量，当文件适合内存时，我成功地做到了这一点。但是，当文件大于内存时，我需要减少 csv.chunksize 选项。我正在尝试这样的事情

package main

import (
    "io"
    "log"
    "os"
    "sync"

    "github.com/apache/arrow/go/v11/arrow"
    "github.com/apache/arrow/go/v11/arrow/csv"
)

// A reader to read lines from the file starting from the byteOffset. The number
// of lines is specified by linesLimit.
func produce(
    id int,
    ch chan<- arrow.Record,
    byteOffset int64,
    linesLimit int64,
    filename string,
    wg *sync.WaitGroup,
) {
    defer wg.Done()

    fd, _ := os.Open(filename)
    fd.Seek(byteOffset, io.SeekStart)

    var remainder int64 = linesLimit % 10
    limit := linesLimit - remainder
    chunkSize := limit / 10

    reader := csv.NewInferringReader(fd,
        csv.WithChunk(int(chunkSize)),
        csv.WithNullReader(true, ""),
        csv.WithComma(','),
        csv.WithHeader(true),
        csv.WithColumnTypes(map[string]arrow.DataType{
            "Start_Time":        arrow.FixedWidthTypes.Timestamp_ns,
            "End_Time":          arrow.FixedWidthTypes.Timestamp_ns,
            "Weather_Timestamp": arrow.FixedWidthTypes.Timestamp_ns,
        }))
    reader.Retain()
    defer reader.Release()

    var count int64
    for reader.Next() {
        rec := reader.Record()
        rec.Retain() // released at the other end of the channel
        ch <- rec
        count += rec.NumRows()
        if count == limit {
            if remainder != 0 {
                flush(id, ch, fd, remainder)
            }
            break
        } else if count > limit {
            log.Panicf("Reader %d read more than it should, expected=%d, read=%d", id, linesLimit, count)
        }
    }

    if reader.Err() != nil {
        log.Panicf("error: %s in line %d,%d", reader.Err().Error(), count, id)
    }
}

func flush(id int,
    ch chan<- arrow.Record,
    fd *os.File,
    limit int64,
) {
    reader := csv.NewInferringReader(fd,
        csv.WithChunk(int(limit)),
        csv.WithNullReader(true, ""),
        csv.WithComma(','),
        csv.WithHeader(false),
    )

    reader.Retain()
    defer reader.Release()

    record := reader.Record()
    record.Retain() // nil pointer dereference error here
    ch <- record
}

我尝试了之前代码的多个版本，包括：

1. 复制文件描述符
2. 复制文件描述符的偏移量，打开同一个文件 并寻求这种抵消。
3. 在调用 flush 或关闭第一个 fd 之前关闭第一个读取器。

无论我如何更改代码，错误似乎都是相同的。请注意，对 flush 的阅读器的任何调用都会引发错误。包括reader.next和reader.err()。

我使用 csv 阅读器是否错误？这是重复使用同一文件的问题吗？

编辑：我不知道这是否有帮助，但是在没有任何 seek 的 flush 中打开一个新的 fd 可以避免该错误（不知何故，任何 seek 都会导致出现原始错误）。但是，如果没有 seek，代码就不正确（即删除 seek 会导致任何 goroutine 根本无法读取文件的一部分）。

正确答案

主要问题是，csv阅读器在下面使用了bufio.reader，其默认缓冲区大小为4096。这意味着reader.next()将读取比需要更多的字节，并缓存额外的字节。如果在 reader.next() 之后直接从文件中读取，将会错过缓存的字节。

下面的演示展示了这种行为：

package main

import (
    "bytes"
    "fmt"
    "io"
    "os"

    "github.com/apache/arrow/go/v11/arrow"
    "github.com/apache/arrow/go/v11/arrow/csv"
)

func main() {
    // Create a two-column csv file with this content (the second column has 1024 bytes):
    // 0,000000....
    // 1,111111....
    // 2,222222....
    // 3,333333....
    temp := createTempFile()

    schema := arrow.NewSchema(
        []arrow.Field{
            {Name: "i64", Type: arrow.PrimitiveTypes.Int64},
            {Name: "str", Type: arrow.BinaryTypes.String},
        },
        nil,
    )
    r := csv.NewReader(
        temp, schema,
        csv.WithComma(','),
        csv.WithChunk(3),
    )
    defer r.Release()

    r.Next()

    // To check what's left after the first chunk is read.
    // If the reader stop at the end of the chunk, the content left will be:
    // 3,333333....
    // But in fact, the content left is:
    // 33333333333
    buf, err := io.ReadAll(temp)
    if err != nil {
        panic(err)
    }

    fmt.Printf("%s\n", buf)
}

func createTempFile() *os.File {
    temp, err := os.CreateTemp("", "test*.csv")
    if err != nil {
        panic(err)
    }
    for i := 0; i < 4; i++ {
        fmt.Fprintf(temp, "%d,", i)
        if _, err := temp.Write(bytes.Repeat([]byte{byte('0' + i)}, 1024)); err != nil {
            panic(err)
        }
        if _, err := temp.Write([]byte("\n")); err != nil {
            panic(err)
        }
    }

    if _, err := temp.Seek(0, io.SeekStart); err != nil {
        panic(err)
    }

    return temp
}

第二个读取器的目的似乎是防止它读取另一个 csv 数据块。如果您提前知道下一个 csv 数据块的偏移量，则可以将文件包装在 io.sectionreader 中，使其仅读取当前的 csv 数据块。当前的问题没有提供有关这部分的足够信息，也许我们应该将其留给另一个问题。

注释：

1. fd, _ := os.open(filename): 永远不要忽略错误。至少记录它们。
2. fd 大多数时候表示文件描述符。不要将其用于 *os.file 类型的变量，尤其是当 *os.file 有一个方法 fd 时。

终于介绍完啦！小伙伴们，这篇关于《Arrow CSV 读取器在同一文件上返回多个 null 值》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！