bigquery

Documentation

Index

• func WithInRangeForDateShardingTable(tableID string, start string, end string) (bool, error)
• type APIOptions
  • func WithDryRun() APIOptions
  • func WithStreamLogFn(f func(msg string)) APIOptions
  • func WithWait() APIOptions
• type BQColumn
• type BigQueryService
  • func NewBigQueryService(bq *bigquery.Client) (*BigQueryService, error)
  • func (s *BigQueryService) Close() error
  • func (s *BigQueryService) Insert(ctx context.Context, dataset *bigquery.Dataset, table string, ...) error
• type DateShardingTableTarget
  • func (t *DateShardingTableTarget) Match(tableID string) (bool, error)
• type JobResult
• type Service
  • func NewService(ctx context.Context, bq *bigquery.Client) (*Service, error)
  • func (s *Service) Close(ctx context.Context) error
  • func (s *Service) DeleteTablesByTablePrefix(ctx context.Context, projectID string, datasetID string, tablePrefix string, ...) ([]string, error)
• type StreamingInsertError
  • func (e *StreamingInsertError) Error() string
• type StreamingInsertErrors
  • func (e *StreamingInsertErrors) Append(err *StreamingInsertError)
  • func (e *StreamingInsertErrors) Error() string
  • func (e *StreamingInsertErrors) ErrorOrNil() error
• type TableService
  • func NewTableService(ctx context.Context, bq *bigquery.Client) (*TableService, error)
  • func (s *TableService) Close(ctx context.Context) error
  • func (s *TableService) DeleteByPrefix(ctx context.Context, projectID string, datasetID string, tablePrefix string, ...) ([]string, error)
  • func (s *TableService) ExistColumn(ctx context.Context, projectID string, datasetID string, tableID string, ...) (bool, error)
  • func (s *TableService) RunDMLToShardingTables(ctx context.Context, projectID string, datasetID string, ...) ([]string, error)

Functions

func WithInRangeForDateShardingTable

func WithInRangeForDateShardingTable(tableID string, start string, end string) (bool, error)

WithInRangeForDateShardingTable is tableで指定したテーブル名がstart, endで指定したYYYYMMDDの範囲にあるかを返す Ex: table=hoge20190101 start=20180101 end=20190102

Types

type APIOptions

type APIOptions func(options *apiOptions)

func WithDryRun

func WithDryRun() APIOptions

WithDryRun is 変更が発生するbigqueryのAPIは実行しない

func WithStreamLogFn

func WithStreamLogFn(f func(msg string)) APIOptions

WithStreamLogFn is Query結果を元にAPIを実行した時にログを処理できる関数を指定できる

func WithWait

func WithWait() APIOptions

WithWait is Jobが一つずつ完了するのを待ってから、次のJobを投入する

type BQColumn

type BQColumn struct {
	Name     string
	Children map[string]*BQColumn
	Target   bool
}

type BigQueryService

type BigQueryService struct {
	BQ *bigquery.Client
}

func NewBigQueryService

func NewBigQueryService(bq *bigquery.Client) (*BigQueryService, error)

func (*BigQueryService) Close

func (s *BigQueryService) Close() error

func (*BigQueryService) Insert

func (s *BigQueryService) Insert(ctx context.Context, dataset *bigquery.Dataset, table string, sss []*bigquery.StructSaver) error

InsertStructSaverToBigQuery is StructSaverをBigQueryにStreamingInsertでInsertする InsertはAtomicには行われない。 Error時は StreamingInsertErrors を返す。

type DateShardingTableTarget

type DateShardingTableTarget struct {
	// Prefix is TableID Prefix
	Prefix string

	// Start is Start YYYYMMDD
	Start string

	// End is End YYYYMMDD
	End string
}

func (*DateShardingTableTarget) Match

func (t *DateShardingTableTarget) Match(tableID string) (bool, error)

type JobResult

type JobResult struct {
	TableID string
	Job     *bigquery.Job
}

type Service

type Service struct {
	BQ *bigquery.Client
}

func NewService

func NewService(ctx context.Context, bq *bigquery.Client) (*Service, error)

func (*Service) Close

func (s *Service) Close(ctx context.Context) error

func (*Service) DeleteTablesByTablePrefix

func (s *Service) DeleteTablesByTablePrefix(ctx context.Context, projectID string, datasetID string, tablePrefix string, ops ...APIOptions) ([]string, error)

DeleteTablesByTablePrefix is 指定したPrefixに合致するTableを削除する

削除したTableIDの一覧を返す 途中で削除に失敗した場合もそれまで削除したTableIDの一覧は返す

type StreamingInsertError

type StreamingInsertError struct {
	InsertID string
	Err      error
}

func (*StreamingInsertError) Error

func (e *StreamingInsertError) Error() string

type StreamingInsertErrors

type StreamingInsertErrors struct {
	Errors []*StreamingInsertError
	// contains filtered or unexported fields
}

func (*StreamingInsertErrors) Append

func (e *StreamingInsertErrors) Append(err *StreamingInsertError)

func (*StreamingInsertErrors) Error

func (e *StreamingInsertErrors) Error() string

func (*StreamingInsertErrors) ErrorOrNil

func (e *StreamingInsertErrors) ErrorOrNil() error

type TableService

type TableService struct {
	BQ *bigquery.Client
}

func NewTableService

func NewTableService(ctx context.Context, bq *bigquery.Client) (*TableService, error)

func (*TableService) Close

func (s *TableService) Close(ctx context.Context) error

func (*TableService) DeleteByPrefix

func (s *TableService) DeleteByPrefix(ctx context.Context, projectID string, datasetID string, tablePrefix string, ops ...APIOptions) ([]string, error)

DeleteByPrefix is 指定したPrefixに合致するTableを削除する

削除したTableIDの一覧を返す 途中で削除に失敗した場合もそれまで削除したTableIDの一覧は返す

func (*TableService) ExistColumn

func (s *TableService) ExistColumn(ctx context.Context, projectID string, datasetID string, tableID string, target string) (bool, error)

ExistColumn is 対象のTableに対象のColumnがある場合は、trueを返す ARRAY(STRUCT)には対応していない

func (*TableService) RunDMLToShardingTables

func (s *TableService) RunDMLToShardingTables(ctx context.Context, projectID string, datasetID string, target *DateShardingTableTarget, dml string, ops ...APIOptions) ([]string, error)

RunDMLToShardingTables is ShardingTableに対してDMLを実行する