package module
Version: v0.8.0 Latest Latest

This package is not in the latest version of its module.

Go to latest
Published: Feb 13, 2020 License: Apache-2.0 Imports: 29 Imported by: 6


Datastore Connectivity for BigQuery (bgc)

Datastore Connectivity library for BigQuery in Go. GoDoc

This library is compatible with Go 1.5+

Please refer to if you encounter breaking changes.

This library uses SQL mode and streaming API to insert data as default. To use legacy SQL please use the following /* USE LEGACY SQL */ hint, in this case you will not be able to fetch repeated and nested fields.

Configuration parameters

To control insert method just provide config.parameters with the following value:

_table_name_.insertMethod = "load"

Note that if streaming is used, currently UPDATE and DELETE statements are not supported.


For streaming you can specify which column to use as insertId with the following config.params

_table_name_.insertMethod = "stream"
_table_name_.insertIdColumn = "sessionId"

streamBatchCount controls row count in batch (default 9999)


When inserting data data this library checks upto 60 sec if data has been added. To control this behaviour you can set insertWaitTimeoutInMs (default 60 sec)

To disable this mechanism set: insertWaitTimeoutInMs: -1


Retries insert when 503 internal error


Default dataset


Default 500

The maximum number of rows of data to return per page of results. In addition to this limit, responses are also limited to 10 MB.


  1. Google secrets for service account

a) set GOOGLE_APPLICATION_CREDENTIALS environment variable

b) credential can be a name with extension of the JSON secret file placed into ~/.secret/ folder


driverName: bigquery
credentials: bq # place your big query secret json to ~/.secret/bg.json
  datasetId: myDataset

c) full URL to secret file


driverName: bigquery
credentials: file://tmp/secret/mySecret.json
  datasetId: myDataset

Secret file has to specify the following attributes:

type Config struct {
	//google cloud credential
	ClientEmail  string `json:"client_email,omitempty"`
	TokenURL     string `json:"token_uri,omitempty"`
	PrivateKey   string `json:"private_key,omitempty"`
	PrivateKeyID string `json:"private_key_id,omitempty"`
	ProjectID  string `json:"project_id,omitempty"`
  1. Private key (pem)


driverName: bigquery
credentials: bq # place your big query secret json to ~/.secret/bg.json
  serviceAccountId: "***"
  datasetId: MyDataset
  projectId: spheric-arcadia-98015
  privateKeyPath: /tmp/secret/bq.pem


The following is a very simple example of Reading and Inserting data

package main

import (

type MostLikedCity struct {
	City      string
	Visits    int
	Souvenirs []string

type  Traveler struct {
	Id            int
	Name          string
	LastVisitTime time.Time
	Achievements  []string
	MostLikedCity MostLikedCity
	VisitedCities []struct {
		City   string
		Visits int

func main() {

    config, err := dsc.NewConfigWithParameters("bigquery", "",
    	    "bq", // google cloud secret placed in ~/.secret/bg.json

    if err != nil {

    factory := dsc.NewManagerFactory()
    manager, err := factory.Create(config)
    if err != nil {
        log.Fatalf("Failed to create manager %v", err)

    traveler := Traveler{}
    success, err := manager.ReadSingle(&traveler, " SELECT id, name, lastVisitTime, visitedCities, achievements, mostLikedCity FROM travelers WHERE id = ?", []interface{}{4}, nil)
    if err != nil {

    travelers :=  make([]Traveler, 0)
    err:= manager.ReadAll(&interest, "SELECT iid, name, lastVisitTime, visitedCities, achievements, mostLikedCity",nil, nil)
    if err != nil {

   // ...

    inserted, updated, err := manager.PersistAll(&travelers, "travelers", nil)
    if err != nil {
    // ...

   //Custom reading handler with reading query info type to get CacheHit, TotalRows, TotalBytesProcessed

   var resultInfo = &bgc.QueryResultInfo{}
   var perf = make(map[string]int)  
   	err = manager.ReadAllWithHandler(`SELECT DATE(date), COUNT(*) FROM performance_agg WHERE DATE(date) = ?  GROUP BY 1`, []interface{}{
   	}, func(scanner dsc.Scanner) (toContinue bool, err error) {
   	        var date string
   	        var count int
   	        err = scanner.Scan(&date, &count)
   	        if err != nil {
   	        	return false, err
   	        perf[date] = count
   		return true, nil
   	log.Printf("cache: %v,  rows: %v, bytes: %v", resultInfo.CacheHit, resultInfo.TotalRows, resultInfo.TotalBytesProcessed)

    dialect := dsc.GetDatastoreDialect(config.DriverName)
    DDL, err := dialect.ShowCreateTable(manager, "performance_agg")
    fmt.Printf("%v %v\n", DDL, err)




The source code is made available under the terms of the Apache License, Version 2, as stated in the file LICENSE.

Individual files may be made available under their own specific license, all compatible with Apache License, Version 2. Please see individual files for details.

Credits and Acknowledgements

Library Author: Adrian Witas

Contributors: Mikhail Berlyant




View Source
const (
	ServiceAccountIdKey = "serviceAccountId"
	PrivateKey          = "privateKey"
	PrivateKeyPathKey   = "privateKeyPath"
	ProjectIDKey        = "projectId"
	DataSetIDKey        = "datasetId"
	DateFormatKey       = "dateFormat"
	MaxResultsKey       = "maxResults"
View Source
const (
	InsertMethodStream       = "stream"
	InsertMethodLoad         = "load"
	InsertWaitTimeoutInMsKey = "insertWaitTimeoutInMs"
	InsertIdColumn           = "insertIdColumn"
	StreamBatchCount         = "streamBatchCount" //can not be more than 10000

	InsertMaxRetires = "insertMaxRetires"


This section is empty.


func GetServiceAndContextForManager

func GetServiceAndContextForManager(manager dsc.Manager) (*bigquery.Service, context.Context, error)

GetServiceAndContextForManager returns big query service and context for passed in datastore manager.


type ColumnInfo

type ColumnInfo struct {
	Name            string
	DataType        string
	IsNullable      bool
	IsPartitioned   bool
	ClusterPosition int

type Compressed

type Compressed struct {
	// contains filtered or unexported fields

Compressed represent compressed encoded payload

func NewCompressed

func NewCompressed(encoderFactory toolbox.EncoderFactory) *Compressed

NewCompressed return new compressed struct

func (*Compressed) Append

func (c *Compressed) Append(data map[string]interface{}) error

Append append data to compressing stream

func (*Compressed) GetAndClose

func (c *Compressed) GetAndClose() (io.Reader, error)

GetAndClose returns reader and cloases the stream

type InsertTask

type InsertTask struct {
	// contains filtered or unexported fields

InsertTask represents insert streaming task.

func NewInsertTask

func NewInsertTask(manager dsc.Manager, table *dsc.TableDescriptor, waitForCompletion bool) (*InsertTask, error)

NewInsertTask creates a new streaming insert task, it takes manager, table descript with schema, waitForCompletion flag with time duration.

func (*InsertTask) Insert

func (it *InsertTask) Insert(reader io.Reader) error

InsertAll streams all records into big query, returns number records streamed or error.

func (*InsertTask) InsertAll

func (it *InsertTask) InsertAll(data interface{}) (int, error)

InsertAll streams or load all records into big query, returns number records streamed or error.

func (*InsertTask) InsertSingle

func (it *InsertTask) InsertSingle(record map[string]interface{}) error

InsertSingle streams single records into big query.

func (*InsertTask) LoadAll

func (it *InsertTask) LoadAll(data interface{}) (int, error)

InsertAll streams all records into big query, returns number records streamed or error.

func (*InsertTask) StreamAll

func (it *InsertTask) StreamAll(data interface{}) (int, error)

InsertAll streams all records into big query, returns number records streamed or error.

type QueryIterator

type QueryIterator struct {
	Rows []*bigquery.TableRow
	// contains filtered or unexported fields

QueryIterator represetns a QueryIterator.

func NewQueryIterator

func NewQueryIterator(manager dsc.Manager, query string) (*QueryIterator, error)

NewQueryIterator creates a new query iterator for passed in datastore manager and query.

func (*QueryIterator) GetColumnTypes

func (qi *QueryIterator) GetColumnTypes() ([]dsc.ColumnType, error)

GetColumns returns query columns, after query executed.

func (*QueryIterator) GetColumns

func (qi *QueryIterator) GetColumns() ([]string, error)

GetColumns returns query columns, after query executed.

func (*QueryIterator) HasNext

func (qi *QueryIterator) HasNext() bool

HasNext returns true if there is next row to fetch.

func (*QueryIterator) Next

func (qi *QueryIterator) Next() ([]interface{}, error)

Next returns next row.

type QueryResultInfo

type QueryResultInfo struct {
	CacheHit            bool
	TotalRows           int
	TotalBytesProcessed int

Represents query result into

func (*QueryResultInfo) Set

func (i *QueryResultInfo) Set(info *QueryResultInfo)

Set sets info values

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
t or T : Toggle theme light dark auto