csv - 如何从 S3 读取 CSV 文件

Question

我有以下代码:

package main

import (
    "encoding/csv"
    "fmt"
    "io/ioutil"
    "path"

    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/s3"
    "github.com/aws/aws-sdk-go/service/s3/s3iface"
)

var (
    // TOKEN = os.Getenv("TOKEN")
    svc s3iface.S3API
)

func main() {    
    // Panics if there is an error in creating session
    svc = s3iface.S3API(s3.New(session.Must(session.NewSession())))

    lambda.Start(Handler)
}

func Handler(evt events.S3Event) error {

    for _, rec := range evt.Records {

        key := rec.S3.Object.Key

        dir, file := path.Split(key)
        // Download the file from S3
        obj, err := svc.GetObject(&s3.GetObjectInput{
            Bucket: aws.String(rec.S3.Bucket.Name),
            Key:    aws.String(key),
        })
        if err != nil {
            return fmt.Errorf("error in downloading %s from S3: %s\n", key, err)
        }

        body, err := ioutil.ReadAll(obj.Body)
        if err != nil {
            return fmt.Errorf("error in reading file %s: %s\n", key, err)
        }

        reader := csv.NewReader(body)
        record, err := reader.ReadAll()
        if err != nil {
            fmt.Println("Error", err)
        }

        for value := range record { // for i:=0; i<len(record)
            fmt.Println("", record[value])
        }
    }
    return nil
}

我正在尝试从 S3 解析 CSV 文件，对每一行做一些事情，但我得到了

cannot use body (type []byte) as type io.Reader in argument to csv.NewReader:
    []byte does not implement io.Reader (missing Read method)

非常感谢任何建议

Answer 1

如错误所述:

cannot use body (type []byte) as type io.Reader in argument to csv.NewReader: []byte does not implement io.Reader (missing Read method)

因为您正在将响应返回的 []byte 传递给 csv.NewReader

需要通过 body 实现 io.Reader 以将其作为参数传递给 csv.NewReader。因为它需要 io.Reader 作为参数。尝试将您的代码更改为:

reader := csv.NewReader(bytes.NewBuffer(body))
record, err := reader.ReadAll()
if err != nil {
    fmt.Println("Error", err)
}

也因为 aws.GetObject返回指向 GetObjectOutput 结构的指针。

func (c *S3) GetObject(input *GetObjectInput) (*GetObjectOutput, error)

它本身实现了 Reader

type GetObjectOutput struct {
    ....
    // Object data.
    Body io.ReadCloser `type:"blob"`
    ....
}

因此您可以将返回的正文直接传递给 csv.NewReader。

还有一件事是您可以使用下载管理器

The s3manager package's Downloader provides concurrently downloading of Objects from S3. The Downloader will write S3 Object content with an io.WriterAt. Once the Downloader instance is created you can call Download concurrently from multiple goroutines safely.

func (d Downloader) Download(w io.WriterAt, input *s3.GetObjectInput, options ...func(*Downloader)) (n int64, err error)

下载在 S3 中下载一个对象，并使用并发 GET 请求将有效负载写入 w。

跨协程并发调用此方法是安全的。

// The session the S3 Downloader will use
sess := session.Must(session.NewSession())

// Create a downloader with the session and default options
downloader := s3manager.NewDownloader(sess)

// Create a file to write the S3 Object contents to.
f, err := os.Create(filename)
if err != nil {
    return fmt.Errorf("failed to create file %q, %v", filename, err)
}

// Write the contents of S3 Object to the file
n, err := downloader.Download(f, &s3.GetObjectInput{
    Bucket: aws.String(myBucket),
    Key:    aws.String(myString),
})
if err != nil {
    return fmt.Errorf("failed to download file, %v", err)
}