Golang - gzip mongodb 查找查询的游标数据、写入文件并解压缩时出错

Question

我正在迭代mongodb游标，并对数据进行gzip封装并发送到S3对象。当尝试使用gzip -d解压缩上传的文件时，出现以下错误，

gzip: 9.log.gz: invalid compressed data--crc error
gzip: 9.log.gz: invalid compressed data--length error

下面给出了我用于迭代、压缩和上传的代码，

// CursorReader struct acts as reader wrapper on top of mongodb cursor
type CursorReader struct {
Csr *mongo.Cursor
}
// Read func reads the data from cursor and puts it into byte array
func (cr *CursorReader) Read(p []byte) (n int, err error) {
dataAvail := cr.Csr.Next(context.TODO())
if !dataAvail {
n = 0
err = io.EOF
if cr.Csr.Close(context.TODO()) != nil {
fmt.Fprintf(os.Stderr, "Error: MongoDB: getting logs: close cursor: %s", err)
}
return
}
var b bytes.Buffer
w := gzip.NewWriter(&b)
w.Write([]byte(cr.Csr.Current.String() + "n"))
w.Close()
n = copy(p, []byte(b.String()))
err = nil
return
}
cursor, err := coll.Find(ctx, filter) // runs the find query and returns cursor
csrRdr := new(CursorReader) // creates a new cursorreader instance
csrRdr.Csr = cursor // assigning the find cursor to cursorreader instance
_, err = s3Uploader.Upload(&s3manager.UploadInput{  // Uploading the data to s3 in parts
Bucket: aws.String("bucket"),
Key:    aws.String("key")),
Body:   csrRdr, 
})

如果数据很低，那么我就没有问题。但如果数据很大，那么我就错了。到目前为止，我调试过的东西，试图压缩1500个文档，每个文档的大小都是15MB，但出现了错误。甚至我也尝试过将gzipped字节直接写入本地文件，但我也遇到了同样的错误。

Answer 1

问题似乎是在func(*CursorReader) Read([]byte) (int, error)中重复调用gzip.NewWriter()

您正在为每次对Read的调用分配一个新的gzip.Writer。gzip压缩是有状态的，因此所有操作只能使用一个Writer实例。

---

解决方案#1

对于您的问题，一个相当简单的解决方案是读取光标中的所有行，并将其传递给gzip.Writer，然后将gzipped内容存储到内存缓冲区中。

var cursor, _ = collection.Find(context.TODO(), filter)
defer cursor.Close(context.TODO())
// prepare a buffer to hold gzipped data
var buffer bytes.Buffer
var gz = gzip.NewWriter(&buffer)
defer gz.Close()
for cursor.Next(context.TODO()) {
if _, err = io.WriteString(gz, cursor.Current.String()); err != nil {
// handle error somehow  ¯_(ツ)_/¯
}
}
// you can now use buffer as io.Reader
// and it'll contain gzipped data for your serialized rows
_, err = s3.Upload(&s3.UploadInput{
Bucket: aws.String("..."),
Key:    aws.String("...")),
Body:   &buffer, 
})

---

解决方案#2

另一种解决方案是使用io.Pipe()和goroutines创建一个流，该流根据需要读取和压缩数据，而不是在内存缓冲区中。如果您正在读取的数据非常大，并且您无法将其全部保存在内存中，那么这将非常有用。

var cursor, _ = collection.Find(context.TODO(), filter)
defer cursor.Close(context.TODO())
// create pipe endpoints
reader, writer := io.Pipe()
// note: io.Pipe() returns a synchronous in-memory pipe
// reads and writes block on one another
// make sure to go through docs once.
// now, since reads and writes on a pipe blocks
// we must move to a background goroutine else
// all our writes would block forever
go func() {
// order of defer here is important
// see: https://stackoverflow.com/a/24720120/6611700
// make sure gzip stream is closed before the pipe
// to ensure data is flushed properly
defer writer.Close()
var gz = gzip.NewWriter(writer)
defer gz.Close()
for cursor.Next(context.Background()) {
if _, err = io.WriteString(gz, cursor.Current.String()); err != nil {
// handle error somehow  ¯_(ツ)_/¯
}
}
}()
// you can now use reader as io.Reader
// and it'll contain gzipped data for your serialized rows
_, err = s3.Upload(&s3.UploadInput{
Bucket: aws.String("..."),
Key:    aws.String("...")),
Body:   reader, 
})