Amazon S3 Transfer Managers シーク不能ストリーミング入力

Amazon S3 ユーティリティ

Amazon S3 Transfer Managers

Amazon S3 アップロードマネージャーとダウンロードマネージャーは、大きなオブジェクトを分割できるため、複数のパートに並行して転送できます。これにより、中断された転送を簡単に再開できます。

Amazon S3 アップロードマネージャー

Amazon S3 アップロードマネージャーは、ファイルを小さな部分に分割して並行してアップロードできるかどうかを決定します。並列アップロードの数とアップロードされたパートのサイズをカスタマイズできます。

次の例では、Amazon S3 を使用してファイルUploaderをアップロードします。の使用は Uploader s3.PutObject()オペレーションに似ています。


import "context"
import "github.com/aws/aws-sdk-go-v2/config"
import "github.com/aws/aws-sdk-go-v2/service/s3"
import "github.com/aws/aws-sdk-go-v2/feature/s3/manager"

// ...

cfg, err := config.LoadDefaultConfig(context.TODO())
if err != nil {
    log.Printf("error: %v", err)
    return
}

client := s3.NewFromConfig(cfg)

uploader := manager.NewUploader(client)
result, err := uploader.Upload(context.TODO(), &s3.PutObjectInput{
    Bucket: aws.String("amzn-s3-demo-bucket"),
    Key:    aws.String("my-object-key"),
    Body:   uploadFile,
})

設定オプション

NewUploader を使用してUploaderインスタンスをインスタンス化する場合、いくつかの設定オプションを指定して、オブジェクトのアップロード方法をカスタマイズできます。オプションは、に 1 つ以上の引数を指定することで上書きされますNewUploader。オプションには以下が含まれます：

PartSize – アップロードする各パートのバッファサイズをバイト単位で指定します。パートあたりの最小サイズは 5 MiB です。
Concurrency – 並列でアップロードするパートの数を指定します。
LeavePartsOnError – Amazon S3 に正常にアップロードされたパートを残すかどうかを示します。

Concurrency 値は、特定のUpload呼び出しで発生する可能性のあるパートアップロードの同時数を制限します。これはグローバルクライアントの同時実行数の制限ではありません。PartSize および Concurrency設定値を微調整して、最適な設定を見つけます。例えば、高帯域幅接続のシステムは、より大きなパートとより多くのアップロードを並列に送信できます。

たとえば、アプリケーションはを Concurrencyに設定Uploaderしてを設定します5。アプリケーションが 2 つの異なるゴルーチンUploadからを呼び出す場合、結果は10同時パートアップロード (2 つのゴルーチン * 5 ) ですConcurrency。

警告

アプリケーションリソースの枯渇を防ぐために、アプリケーションは同時呼び出しを Upload に制限することが期待されます。

以下は、Uploader作成時にデフォルトのパートサイズを設定する例です。


uploader := manager.NewUploader(client, func(u *Uploader) {
    u.PartSize = 10 * 1024 * 1024, // 10 MiB
})

Uploader とその設定の詳細については、 AWS SDK for Go API リファレンスの「アップローダー」を参照してください。

PutObjectInput 本文フィールド (io.ReadSeeker と io.Reader)

s3.PutObjectInput 構造体の Bodyフィールドは io.Reader型です。ただし、このフィールドには、ホスト環境のアプリケーションリソース使用率を向上させるために、 io.ReadSeekerと io.ReaderAtインターフェイスの両方を満たすタイプを入力できます。次の例では、両方のインターフェイスを満たす ReadSeekerAt タイプを作成します。


type ReadSeekerAt interface {
    io.ReadSeeker
    io.ReaderAt
}

io.Reader タイプの場合、パートをアップロードする前にリーダーのバイトをメモリにバッファする必要があります。PartSize または Concurrencyの値を増やすと、に必要なメモリ (RAM) が大幅にUploader増加します。必要なメモリは約 PartSize * ですConcurrency。たとえば、に 100 MB、に 10 MB を指定するPartSize場合Concurrency、には少なくとも 1 GB が必要です。

io.Reader タイプはバイトを読み取る前にサイズを判断できないため、はアップロードするパート数を計算Uploaderできません。したがって、を低PartSizeすぎる値に設定すると、は大きなファイルに対して Amazon S3 アップロード制限の 10,000 パートに達するUploader可能性があります。10,000 個を超えるパートをアップロードしようとすると、アップロードが停止し、エラーが返されます。

ReadSeekerAt タイプを実装するbody値の場合、 Uploaderは Amazon S3 に送信する前に本文の内容をメモリにバッファしません。は、ファイルを Amazon S3 にアップロードする前に予想されるパート数をUploader計算します。の現在の値でファイルのアップロードに 10,000 個を超えるパートPartSizeが必要な場合、はパートサイズの値Uploaderを増やして必要なパート数を減らします。

失敗したアップロードの処理

Amazon S3 へのアップロードが失敗した場合、デフォルトでは、 Uploaderは Amazon S3 AbortMultipartUploadオペレーションを使用してアップロードされたパートを削除します。この機能を使用すると、失敗したアップロードで Amazon S3 ストレージが消費されることがなくなります。

が正常にアップロードされたパートを Uploaderが削除しないように、を true LeavePartsOnErrorに設定できます。これは、部分的に完了したアップロードを再開する場合に便利です。アップロードされたパートを操作するには、失敗したアップロードUploadIDのを取得する必要があります。次の例は、manager.MultiUploadFailureエラーインターフェイスタイプを使用してを取得する方法を示していますUploadID。


result, err := uploader.Upload(context.TODO(), &s3.PutObjectInput{
    Bucket: aws.String("amzn-s3-demo-bucket"),
    Key:    aws.String("my-object-key"),
    Body:   uploadFile,
})
output, err := u.upload(input)
if err != nil {
    var mu manager.MultiUploadFailure
    if errors.As(err, &mu) {
        // Process error and its associated uploadID
        fmt.Println("Error:", mu)
        _ = mu.UploadID() // retrieve the associated UploadID
    } else {
        // Process error generically
        fmt.Println("Error:", err.Error())
    }
    return
}

アップロードごとのアップローダーオプションの上書き

メソッドに 1 つ以上の引数を指定Uploadすることで、を呼び出すときにUploaderオプションを上書きできます。これらのオーバーライドは同時実行に安全な変更であり、進行中のアップロードやマネージャーへの後続のUpload呼び出しには影響しません。たとえば、特定のアップロードリクエストPartSizeの設定を上書きするには、次のようにします。


params := &s3.PutObjectInput{
    Bucket: aws.String("amzn-s3-demo-bucket"),
    Key:    aws.String("my-key"),
    Body:   myBody,
}
resp, err := uploader.Upload(context.TODO(), params, func(u *manager.Uploader) {
    u.PartSize = 10 * 1024 * 1024, // 10 MiB
})

例

Amazon S3 にフォルダをアップロードする

次の例では、 path/filepathパッケージを使用してファイルのリストを再帰的に収集し、指定された Amazon S3 バケットにアップロードします。Amazon S3 オブジェクトのキーには、ファイルの相対パスがプレフィックスとして付けられます。


package main

import (
    "context"
    "log"
    "os"
    "path/filepath"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

var (
    localPath string
    bucket    string
    prefix    string
)

func init() {
    if len(os.Args) != 4 {
        log.Fatalln("Usage:", os.Args[0], "<local path> <bucket> <prefix>")
    }
    localPath = os.Args[1]
    bucket = os.Args[2]
    prefix = os.Args[3]
}

func main() {
    walker := make(fileWalk)
    go func() {
        // Gather the files to upload by walking the path recursively 
        if err := filepath.Walk(localPath, walker.Walk); err != nil {
            log.Fatalln("Walk failed:", err)
        }
        close(walker)
    }()

    cfg, err := config.LoadDefaultConfig(context.TODO())
    if err != nil {
        log.Fatalln("error:", err)
    }
    
    // For each file found walking, upload it to Amazon S3
    uploader := manager.NewUploader(s3.NewFromConfig(cfg))
    for path := range walker {
        rel, err := filepath.Rel(localPath, path)
        if err != nil {
            log.Fatalln("Unable to get relative path:", path, err)
        }
        file, err := os.Open(path)
        if err != nil {
            log.Println("Failed opening file", path, err)
            continue
        }
        defer file.Close()
        result, err := uploader.Upload(context.TODO(), &s3.PutObjectInput{
            Bucket: &bucket,
            Key:    aws.String(filepath.Join(prefix, rel)),
            Body:   file,
        })
        if err != nil {
            log.Fatalln("Failed to upload", path, err)
        }
        log.Println("Uploaded", path, result.Location)
    }
}

type fileWalk chan string

func (f fileWalk) Walk(path string, info os.FileInfo, err error) error {
    if err != nil {
        return err
    }
    if !info.IsDir() {
        f <- path
    }
    return nil
}

ダウンロードマネージャー

Amazon S3 ダウンローダーマネージャーは、ファイルを小さな部分に分割して並列ダウンロードできるかどうかを決定します。並列ダウンロードの数とダウンロードしたパートのサイズをカスタマイズできます。

例: ファイルのダウンロード

次の例では、Amazon S3 を使用してファイルDownloaderをダウンロードします。の使用はDownloader、s3.GetObject オペレーションに似ています。


import "context"
import "github.com/aws/aws-sdk-go-v2/aws"
import "github.com/aws/aws-sdk-go-v2/config"
import "github.com/aws/aws-sdk-go-v2/service/s3"
import "github.com/aws/aws-sdk-go-v2/feature/s3/manager"

// ...

cfg, err := config.LoadDefaultConfig(context.TODO())
if err != nil {
    log.Println("error:", err)
    return
}

client := s3.NewFromConfig(cfg)

downloader := manager.NewDownloader(client)
numBytes, err := downloader.Download(context.TODO(), downloadFile, &s3.GetObjectInput{
    Bucket: aws.String("amzn-s3-demo-bucket"), 
    Key:    aws.String("my-key"),
})

downloadFile パラメータは io.WriterAt型です。WriterAt インターフェイスを使用すると、 Downloaderはファイルの複数の部分を並列に書き込むことができます。

設定オプション

Downloader インスタンスをインスタンス化するときに、設定オプションを指定して、オブジェクトのダウンロード方法をカスタマイズできます。

PartSize – ダウンロードする各パートのバッファサイズをバイト単位で指定します。パートあたりの最小サイズは 5 MB です。
Concurrency – 並列ダウンロードするパートの数を指定します。

Concurrency 値は、特定のDownload呼び出しに対して発生する可能性のあるパートダウンロードの同時数を制限します。これはグローバルクライアントの同時実行数の制限ではありません。PartSize および Concurrency設定値を微調整して、最適な設定を見つけます。例えば、高帯域幅接続のシステムは、より大きなパートとより多くのダウンロードを並行して受け取ることができます。

たとえば、アプリケーションは Concurrencyの Downloaderを使用してを設定します5。次に、アプリケーションは 2 つの異なるゴルーチンDownloadからを呼び出します。結果は10同時パートダウンロードになります (2 つのゴルーチン * 5 Concurrency）。

警告

アプリケーションリソースの枯渇を防ぐために、アプリケーションは同時呼び出しDownloadをに制限することが期待されます。

Downloader とそのその他の設定オプションの詳細については、 AWS SDK for Go API リファレンスの「manage.Downloader」を参照してください。

ダウンロードあたりのダウンローダーオプションの上書き

メソッドに 1 つ以上の関数引数を指定Downloadすることで、を呼び出すときにDownloaderオプションを上書きできます。これらの上書きは同時実行が安全な変更であり、進行中のアップロードやマネージャーへの後続のDownload呼び出しには影響しません。たとえば、特定のアップロードリクエストPartSizeの設定を上書きするには、次のようにします。


params := &s3.GetObjectInput{
    Bucket: aws.String("amzn-s3-demo-bucket"),
    Key:    aws.String("my-key"),
}
resp, err := downloader.Download(context.TODO(), targetWriter, params, func(u *manager.Downloader) {
    u.PartSize = 10 * 1024 * 1024, // 10 MiB
})

例

バケット内のすべてのオブジェクトをダウンロードする

次の例では、ページ分割を使用して Amazon S3 バケットからオブジェクトのリストを収集します。次に、各オブジェクトをローカルファイルにダウンロードします。


package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "path/filepath"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

var (
    Bucket         = "amzn-s3-demo-bucket" // Download from this bucket
    Prefix         = "logs/"    // Using this key prefix
    LocalDirectory = "s3logs"   // Into this directory
)

func main() {
    cfg, err := config.LoadDefaultConfig(context.TODO())
    if err != nil {
        log.Fatalln("error:", err)
    }

    client := s3.NewFromConfig(cfg)
    manager := manager.NewDownloader(client)

    paginator := s3.NewListObjectsV2Paginator(client, &s3.ListObjectsV2Input{
        Bucket: &Bucket,
        Prefix: &Prefix,
    })

    for paginator.HasMorePages() {
        page, err := paginator.NextPage(context.TODO())
        if err != nil {
            log.Fatalln("error:", err)
        }
        for _, obj := range page.Contents {
            if err := downloadToFile(manager, LocalDirectory, Bucket, aws.ToString(obj.Key)); err != nil {
                log.Fatalln("error:", err)
            }
        }
    }
}

func downloadToFile(downloader *manager.Downloader, targetDirectory, bucket, key string) error {
    // Create the directories in the path
    file := filepath.Join(targetDirectory, key)
    if err := os.MkdirAll(filepath.Dir(file), 0775); err != nil {
        return err
    }

    // Set up the local file
    fd, err := os.Create(file)
    if err != nil {
        return err
    }
    defer fd.Close()

    // Download the file using the AWS SDK for Go
    fmt.Printf("Downloading s3://%s/%s to %s...\n", bucket, key, file)
    _, err = downloader.Download(context.TODO(), fd, &s3.GetObjectInput{Bucket: &bucket, Key: &key})

    return err
}

GetBucketRegion

GetBucketRegion は、Amazon S3 バケットの AWS リージョンの場所を決定するためのユーティリティ関数です。 GetBucketRegionは Amazon S3 クライアントを取得し、それを使用して、クライアントの設定されたリージョンに関連付けられた AWS パーティション内のリクエストされたバケットの場所を決定します。

例えば、バケットのリージョンを見つけるにはamzn-s3-demo-bucket：


cfg, err := config.LoadDefaultConfig(context.TODO())
if err != nil {
    log.Println("error:", err)
    return
}

bucket := "amzn-s3-demo-bucket"
region, err := manager.GetBucketRegion(ctx, s3.NewFromConfig(cfg), bucket)
if err != nil {
    var bnf manager.BucketNotFound
    if errors.As(err, &bnf) {
        log.Printf("unable to find bucket %s's Region\n", bucket)
    } else {
        log.Println("error:", err)
    }
    return
}
fmt.Printf("Bucket %s is in %s region\n", bucket, region)

GetBucketRegion がバケットの場所を解決できない場合、関数は例に示すように BucketNotFound エラータイプを返します。

シーク不能ストリーミング入力

PutObject やなどの API オペレーションの場合UploadPart、Amazon S3 クライアントは、デフォルトで io.Seeker インターフェイスを実装するBody入力パラメータの値を想定しています。io.Seeker インターフェイスは、クライアントがアップロードする値の長さを決定し、リクエスト署名のペイロードハッシュを計算するために使用されます。Body 入力パラメータ値がを実装していない場合io.Seeker、アプリケーションはエラーを受け取ります。


operation error S3: PutObject, failed to compute payload hash: failed to seek
body to start, request stream is not seekable

この動作を変更するには、機能オプションミドルウェアを使用してオペレーションメソッドのを変更します。WithAPIOptions ヘルパーは、ゼロ以上のミドルウェアミューテーターの機能オプションを返します。ペイロードハッシュの計算を無効にし、署名なしペイロードリクエスト署名を使用するには、v4.SwapComputePayloadSHA256ForUnsignedPayloadMiddleware を追加します。


resp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
    Bucket: &bucketName,
    Key: &objectName,
    Body: bytes.NewBuffer([]byte(`example object!`)),
    ContentLength: 15, // length of body
}, s3.WithAPIOptions(
    v4.SwapComputePayloadSHA256ForUnsignedPayloadMiddleware,
))

警告

Amazon S3 では、バケットにアップロードされたすべてのオブジェクトに対してコンテンツの長さを指定する必要があります。Body 入力パラメータはio.Seekerインターフェイスを実装していないため、クライアントはリクエストの ContentLengthパラメータを計算できません。パラメータはアプリケーションによって指定する必要があります。ContentLength パラメータが指定されていない場合、リクエストは失敗します。

シーク可能でなく、長さが不明なアップロードAmazon S3 アップロードマネージャーには SDK を使用します。

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

Amazon EC2 インスタンスメタデータサービス

ミドルウェア