golangでGoogle Cloud StorageへのUploadからBigQueryへLoadする

GCP認証

func Auth() error {
    // 環境変数にセットしたGoogleApplicationCredentialsを指定
    fn := os.Getenv("GOOGLE_APPLICATION_CREDENTIALS")
    b, err := ioutil.ReadFile(fn)
    if err != nil {
        return err
    }
    _, err = google.JWTConfigFromJSON(b)
    if err != nil {
        return err
    }
    return nil
}

CloudStorege Upload

func GcsPut(bktName string, objName string, path string) error {

    err := Auth()
    if err != nil {
        return err
    }

    ctx := context.Background()

    // Creates a client.
    client, err := storage.NewClient(ctx)
    if err != nil {
        return err
    }
    defer client.Close()

    w := client.Bucket(bktName).Object(objName).NewWriter(ctx)
    // option
    w.ObjectAttrs.ContentType = "application/octet-stream"
    w.ChunkSize = 1024

    defer w.Close()

    f, err := os.Open(path)
    if err != nil {
        return err
    }

    defer f.Close()
    if _, err = io.Copy(w, f); err != nil {
        log.Fatal(err)
    }

    return nil
}

ContentType	ChunkSize
型：String ContentTypeを指定します。	型：Int 指定したサイズを上限にオブジェクトを分割してアップロードする指定。 1回のリクエストでアップしたい場合はCunkSizeを指定。 上限を指定しない場合は０または無指定。

オプションの詳細はこちら

BigQuery Load

func BigQueryLoad(config Config, gcspath string, tableid string) error {

    err := Auth()
    if err != nil {
        return err
    }

    projectid := config.BigQuery.Projectid
    gcsuri := config.BigQuery.Gcsuri + gcspath
    datasetid := config.BigQuery.Datasetid

    ctx := context.Background()

    // Creates a client.
    client, err := bigquery.NewClient(ctx, projectid)
    if err != nil {
        return err
    }

    gcsRef := bigquery.NewGCSReference(gcsuri)

    // option
    gcsRef.AllowJaggedRows = true
    gcsRef.Compression = bigquery.Gzip
    gcsRef.MaxBadRecords = 1000
    gcsRef.FieldDelimiter = "n"

    schema := bigquery.Schema{
        &bigquery.FieldSchema{Name: "col1", Required: true, Type: bigquery.IntegerFieldType},
        &bigquery.FieldSchema{Name: "col2", Required: false, Type: bigquery.StringFieldType},
        &bigquery.FieldSchema{Name: "col3", Required: true, Type: bigquery.StringFieldType},
    }

    dataset := client.Dataset(datasetid)

    err = dataset.Table(tableid).Create(ctx, &bigquery.TableMetadata{
        Schema: schema,
    })
    if err != nil {
        return err
    }

    loader := dataset.Table(tableid).LoaderFrom(gcsRef)
    loader.CreateDisposition = bigquery.CreateNever

    job, err := loader.Run(ctx)
    if err != nil {
        return err
    }
    status, err := job.Wait(ctx)
    if err != nil {
        return err
    }
    if err := status.Err(); err != nil {
        return err
    }
    return nil
}

AllowJaggedRows	MaxBadRecords	Compression	FileDelimiter
型：bool 不正なレコードがあった場合、NULLとして扱うかエラーとして扱うかを選択します。	型：Int 不正なレコード数を許容するレコード数を指定します。	型：String Google Cloud StorageにuploadされたLoadするObjectの圧縮タイプを指定します。	型：String Loadされるファイルのdelimiterを指定します。

オプションの詳細はこちら
圧縮されたファイルをLoadをする場合はCompression、CSV形式ではないファイルをLoadする場合はFileDelimiterを指定しましょう。