Query data from a BQ table
func mainReturn() interface{} {
bigqueryService, err := getBigqueryService()
if(err != nil) {
return "ERROR"
}
var useQueryCache bool = true
billingProjectId := "billingProjectId"
projectId := "projectId"
datasetId := "datasetId"
tableId := "tableId"
var query = "select * from [" + projectId + ":" + datasetId + "." + tableId + "];"
var request bigquery.QueryRequest = bigquery.QueryRequest{
Kind: "bigquery#queryRequest",
Query: query,
UseQueryCache: &useQueryCache,
//To turn of legacySql, do this:
//UseLegacySql: false,
//ForceSendFields: []string{"UseLegacySql"},
}
//Run the query, using JobsService
//https://godoc.org/google.golang.org/api/bigquery/v2#JobsService.Query
call := bigqueryService.Jobs.Query(billingProjectId, &request)
//executes the query
//https://godoc.org/google.golang.org/api/bigquery/v2#JobsQueryCall
resp, err := call.Do()
if err != nil {
return "ERROR"
}
var results string = ""
//loop through the rows, it is a slice of TableRow pointers, []*TableRow
rows := resp.Rows
for i :=0 ; i < len(rows); i++ {
row := rows[i]
//https://godoc.org/google.golang.org/api/bigquery/v2#TableRow
//A tableRow is a slice of TableCell pointers []*TableCell
fields := row.F
for fi := 0; fi < len(fields); fi++ {
results = results + xSprintf("Field %v = %v, ", fi, fields[fi].V)
}
results = results + "\n"
}
return results
}
|
This example access BigQuery service, reference a table, constructs a select statement, and then displays the result of the statement |
Copy table
func copyTable(billingProjectId string, srcProjectId string, srcDatasetId string, srcTableId string, destProjectId string, destDatasetId string, destTableId string, bigqueryService *bigquery.Service) error {
//==========Set copy configurations==========
//https://godoc.org/google.golang.org/api/bigquery/v2#JobConfigurationTableCopy
var createDisposition string = "CREATE_IF_NEEDED" //Other options: "CREATE_NEVER"
var destinationTable bigquery.TableReference = bigquery.TableReference {
ProjectId: destProjectId,
DatasetId: destDatasetId,
TableId: destTableId,
}
var sourceTable bigquery.TableReference = bigquery.TableReference {
ProjectId: srcProjectId,
DatasetId: srcDatasetId,
TableId: srcTableId,
}
var writeDisposition string = "WRITE_TRUNCATE" //Other options: "WRITE_APPEND", "WRITE_EMPTY"
//==========Set job configurations==============
//https://godoc.org/google.golang.org/api/bigquery/v2#JobReference
var jobId string = "copyTableJob_" + strconv.FormatInt(time.Now().Unix(), 10)
//========Prepare a BQ job for submission=======
job := &bigquery.Job {
Configuration: &bigquery.JobConfiguration {
Copy: &bigquery.JobConfigurationTableCopy {
CreateDisposition: createDisposition,
DestinationTable: &destinationTable,
SourceTable: &sourceTable,
WriteDisposition: writeDisposition,
},
},
JobReference: &bigquery.JobReference {
JobId: jobId,
ProjectId: billingProjectId,
},
}
//==============Submit the BQ job===================
call := bigqueryService.Jobs.Insert(billingProjectId, job)
job, err := call.Do()
if err != nil {
return err
}
jobId = job.JobReference.JobId
xPrintf("BQ job has been submitted with jobId=%v\n", jobId)
//==============Poll job status=====================
//poll job status until the job is done or we reach a time out
var timeoutSec int64 = 60 * 2
var pollStartSec = time.Now().Unix()
xPrintf("pollStartSec=%v\n", pollStartSec)
for {
var bqjob *bigquery.Job
var geterr error
bqjob, geterr = bigqueryService.Jobs.Get(billingProjectId, jobId).Do()
if(geterr != nil) {
return geterr
}
xPrintf("Polled job state=%v\n", bqjob.Status.State)
if(bqjob.Status.State == "DONE") {
//Display job info
xPrintf("\n======Job is done======\n")
xPrintf("State=%v\n", bqjob.Status.State)
xPrintf("UserEmail=%v\n", bqjob.UserEmail)
xPrintf("Etag=%v\n", bqjob.Etag)
if(bqjob.Status.ErrorResult != nil) {
xPrintf("Error Location=%v\n", bqjob.Status.ErrorResult.Location)
xPrintf("Error Message=%v\n", bqjob.Status.ErrorResult.Message)
xPrintf("Error Reason=%v\n", bqjob.Status.ErrorResult.Reason)
} else if(bqjob.Statistics != nil) {
xPrintf("CreationTime=%v\n", bqjob.Statistics.CreationTime)
xPrintf("StartTime=%v\n", bqjob.Statistics.StartTime)
xPrintf("EndTime=%v\n", bqjob.Statistics.EndTime)
xPrintf("Duration ms=%v\n", bqjob.Statistics.EndTime-bqjob.Statistics.CreationTime)
}
break
}
xPrintln("Sleeping...")
time.Sleep(10 * time.Second)
var elapsedSec int64 = time.Now().Unix() - pollStartSec
if(elapsedSec > timeoutSec) {
xPrintln("Reached timed out")
timeoutErr := errors.New("Polling job status has reached timeout")
return timeoutErr
}
}
return nil
}
func mainReturn() interface{} {
bigqueryService, err := getBigqueryService()
if(err != nil) {
return "Error obtaining bigquery service=" + err.Error()
}
var billingProjectId string = "billingProjectId"
var srcProjectId string = "srcProjectId"
var srcDatasetId string = "srcDatasetId"
var srcTableId string = "srcTableId"
var destProjectId string = "destProjectId"
var destDatasetId string = "destDatasetId"
var destTableId string = "destTableId"
copyError := copyTable(billingProjectId, srcProjectId, srcDatasetId, srcTableId, destProjectId, destDatasetId, destTableId, bigqueryService)
if(copyError != nil) {
return "Copy error=" + copyError.Error()
}
return "OK"
}
|
Constructs source and destination table to copy |
Load data into BQ from CS
func mainReturn() interface{} {
bigqueryService, err := getBigqueryService()
if(err != nil) {
return "ERROR"
}
var billingProjectId = "billingProjectId"
var projectId = "projectId"
var datasetId = "datasetId"
var tableId = "tableId"
var schema bigquery.TableSchema
var schemaStr = "{\"fields\":[{\"name\": \"color\", \"type\": \"STRING\"},{\"name\": \"price\", \"type\": \"INTEGER\"}]}"
var sourceUri = "gs://dwtest/destinationfile.txt"
//json unmarshall: https://golang.org/pkg/encoding/json/#example_Unmarshal
err = json.Unmarshal([]byte(schemaStr), &schema)
if(err != nil) {
return "ERROR"
}
job := &bigquery.Job{
Configuration: &bigquery.JobConfiguration{
Load: &bigquery.JobConfigurationLoad{
DestinationTable: &bigquery.TableReference{
DatasetId: datasetId,
ProjectId: projectId,
TableId: tableId,
},
MaxBadRecords: 0,
Schema: &schema,
SourceUris: []string{sourceUri},
SourceFormat: "CSV",
FieldDelimiter: "\t",
},
},
}
call := bigqueryService.Jobs.Insert(billingProjectId, job)
job, err2 := call.Do()
if err2 != nil {
return "ERROR"
}
//check job status
var jobid = job.JobReference.JobId
bqjob, err3 := bigqueryService.Jobs.Get(billingProjectId, jobid).Do()
if(err3 != nil) {
return "ERROR"
}
return xSprintf("%v", bqjob.Status.State)
}
|
Loads data from cloud storage file into big query destination table |
Query into a BQ table
func loadToBQTableFromQuery(billingProjectId string, destProjectId string, destDatasetId string, destTableId string, query string, udfs []string, bigqueryService *bigquery.Service) error {
//==========Set query configurations==========
//https://godoc.org/google.golang.org/api/bigquery/v2#JobConfigurationQuery
var allowLargeResults bool = true
var createDisposition string = "CREATE_IF_NEEDED" //Other options: "CREATE_NEVER"
//var defaultDataset bigquery.DatasetReference = bigquery.DatasetReference { ProjectId: "defaultProjectId", DatasetId: "defaultDatasetId" }
var destinationTable bigquery.TableReference = bigquery.TableReference {
ProjectId: destProjectId,
DatasetId: destDatasetId,
TableId: destTableId,
}
var flattenResults bool = false
var maximumBillingTier int64 = 100
var priority string = "INTERACTIVE" //Other options: "BATCH"
var useLegacySql bool = false
var useQueryCache bool = false
var writeDisposition string = "WRITE_TRUNCATE" //Other options: "WRITE_EMPTY"
var userDefinedFunctionResources []*bigquery.UserDefinedFunctionResource = nil
if(udfs != nil && len(udfs) > 0) {
userDefinedFunctionResources = make([]*bigquery.UserDefinedFunctionResource, len(udfs))
for i := range udfs {
userDefinedFunctionResources[i] = &bigquery.UserDefinedFunctionResource { ResourceUri: udfs[i]}
}
}
//==========Set job configurations==============
//https://godoc.org/google.golang.org/api/bigquery/v2#JobReference
var jobId string = "loadFromQueryJob_" + strconv.FormatInt(time.Now().Unix(), 10)
//========Prepare a BQ job for submission=======
job := &bigquery.Job {
Configuration: &bigquery.JobConfiguration {
Query: &bigquery.JobConfigurationQuery {
AllowLargeResults: allowLargeResults,
CreateDisposition: createDisposition,
//DefaultDataset: &defaultDataset,
DestinationTable: &destinationTable,
FlattenResults: &flattenResults,
MaximumBillingTier: &maximumBillingTier,
Priority: priority,
Query: query,
UseLegacySql: useLegacySql,
UseQueryCache: &useQueryCache,
UserDefinedFunctionResources: userDefinedFunctionResources,
WriteDisposition: writeDisposition,
},
},
JobReference: &bigquery.JobReference {
JobId: jobId,
ProjectId: billingProjectId,
},
}
//==============Submit the BQ job===================
call := bigqueryService.Jobs.Insert(billingProjectId, job)
job, err := call.Do()
if err != nil {
return err
}
jobId = job.JobReference.JobId
xPrintf("BQ job has been submitted with jobId=%v\n", jobId)
//==============Poll job status=====================
//poll job status until the job is done or we reach a time out
var timeoutSec int64 = 60 * 2
var pollStartSec = time.Now().Unix()
xPrintf("pollStartSec=%v\n", pollStartSec)
for {
var bqjob *bigquery.Job
var geterr error
bqjob, geterr = bigqueryService.Jobs.Get(billingProjectId, jobId).Do()
if(geterr != nil) {
return geterr
}
xPrintf("Polled job state=%v\n", bqjob.Status.State)
if(bqjob.Status.State == "DONE") {
//Display job info
xPrintf("\n======Job is done======\n")
xPrintf("State=%v\n", bqjob.Status.State)
xPrintf("UserEmail=%v\n", bqjob.UserEmail)
xPrintf("Etag=%v\n", bqjob.Etag)
if(bqjob.Status.ErrorResult != nil) {
xPrintf("Error Location=%v\n", bqjob.Status.ErrorResult.Location)
xPrintf("Error Message=%v\n", bqjob.Status.ErrorResult.Message)
xPrintf("Error Reason=%v\n", bqjob.Status.ErrorResult.Reason)
} else if(bqjob.Statistics != nil) {
xPrintf("CreationTime=%v\n", bqjob.Statistics.CreationTime)
xPrintf("StartTime=%v\n", bqjob.Statistics.StartTime)
xPrintf("EndTime=%v\n", bqjob.Statistics.EndTime)
xPrintf("Duration ms=%v\n", bqjob.Statistics.EndTime-bqjob.Statistics.CreationTime)
xPrintf("BillingTier=%v\n", bqjob.Statistics.Query.BillingTier)
xPrintf("CacheHit=%v\n", bqjob.Statistics.Query.CacheHit)
xPrintf("TotalBytesBilled=%v\n", bqjob.Statistics.Query.TotalBytesBilled)
xPrintf("TotalBytesProcessed=%v\n", bqjob.Statistics.Query.TotalBytesProcessed)
if(bqjob.Statistics.Query.ReferencedTables != nil) {
for i := range bqjob.Statistics.Query.ReferencedTables {
xPrintf("Referenced Table=%v:%v.%v\n", bqjob.Statistics.Query.ReferencedTables[i].ProjectId, bqjob.Statistics.Query.ReferencedTables[i].DatasetId, bqjob.Statistics.Query.ReferencedTables[i].TableId)
}
}
}
break
}
xPrintln("Sleeping...")
time.Sleep(10 * time.Second)
var elapsedSec int64 = time.Now().Unix() - pollStartSec
if(elapsedSec > timeoutSec) {
xPrintln("Reached timed out")
timeoutErr := errors.New("Polling job status has reached timeout")
return timeoutErr
}
}
return nil
}
func mainReturn() interface{} {
bigqueryService, err := getBigqueryService()
if(err != nil) {
return "Error obtaining bigquery service=" + err.Error()
}
var billingProjectId string = "yourBillingProject"
var destProjectId string = "destProjectId"
var destDatasetId string = "destDatasetId"
var destTableId string = "destTableId"
var query string = "select 'apple' as fruit, 'red' as color"
var udfs []string = nil
queryError := loadToBQTableFromQuery(billingProjectId, destProjectId, destDatasetId, destTableId, query, udfs, bigqueryService)
if(queryError != nil) {
return "Query error=" + queryError.Error()
}
return "OK"
}
|
Runs a query and loads the data results into a destination table |
List tables from a dataset
func getMetaFromBQTable(projectId string, datasetId string, tableId string, bigqueryService *bigquery.Service)(*bigquery.Table, error) {
//https://godoc.org/google.golang.org/api/bigquery/v2#TablesService
call := bigqueryService.Tables.Get(projectId, datasetId, tableId)
table, err := call.Do()
if err != nil {
return nil, err
}
return table, nil
}
func parseUnixTime(sec int64)(string) {
t := time.Unix(sec, 0)
return t.Format("2006-01-02 15:04:05")
}
func printMetaFromBQTable(projectId string, datasetId string, tableId string, bigqueryService *bigquery.Service, buf *bytes.Buffer) {
table, err := getMetaFromBQTable(projectId, datasetId, tableId, bigqueryService)
if err != nil {
xFprintf(buf, "Error getting table info:", err.Error())
return
}
//https://godoc.org/google.golang.org/api/bigquery/v2#Table
xFprintf(buf, "Id: %v\n", table.Id)
xFprintf(buf, "CreationTime: %v\n", table.CreationTime)
xFprintf(buf, "CreationTime: %v\n", parseUnixTime(int64(table.CreationTime)/1000))
xFprintf(buf, "LastModifiedTime: %v\n", table.LastModifiedTime)
xFprintf(buf, "LastModifiedTime: %v\n", parseUnixTime(int64(table.LastModifiedTime)/1000))
xFprintf(buf, "ExpirationTime: %v\n", table.ExpirationTime)
xFprintf(buf, "Etag: %v\n", table.Etag)
xFprintf(buf, "FriendlyName: %v\n", table.FriendlyName)
xFprintf(buf, "Kind: %v\n", table.Kind)
xFprintf(buf, "NumBytes: %v\n", table.NumBytes)
xFprintf(buf, "NumRows: %v\n", table.NumRows)
}
func getTablesFromDataset(projectId string, datasetId string, bigqueryService *bigquery.Service)(error) {
//first list all tables from the dataset
//template of looping pages: https://cloud.google.com/storage/docs/json_api/v1/json-api-go-samples
var buf bytes.Buffer
var pagetoken string = ""
ind := 0
for {
call := bigqueryService.Tables.List(projectId, datasetId)
//PageToken sets the optional parameter "pageToken": Page token, returned by a previous call, to request the next page of results
if len(pagetoken) > 0 {
call = call.PageToken(pagetoken)
}
list, err := call.Do()
if err != nil {
return err
}
//https://godoc.org/google.golang.org/api/bigquery/v2#TableList
if len(pagetoken) == 0 {
//just print it once at the beginning
xFprintf(&buf, "TotalItems: %v\n", list.TotalItems)
}
for _, table := range list.Tables {
//https://godoc.org/google.golang.org/api/bigquery/v2#TableListTables
xFprintf(&buf, "========= Table %v ==============\n", ind)
xFprintf(&buf, "Id: %v\n", table.Id)
xFprintf(&buf, "Kind: %v\n", table.Kind)
//parse out the tableId
tableid := table.Id
del := strings.LastIndex(tableid, ".")
if del != -1 {
tableid = tableid[del+1:]
}
printMetaFromBQTable(projectId, datasetId, tableid, bigqueryService, &buf)
ind++;
}
pagetoken = list.NextPageToken
if len(pagetoken) == 0 {
break
}
}
xPrintln(buf.String())
return nil
}
func mainReturn() interface{} {
projectId := "projectId"
datasetId := "datasetId"
bigqueryService, err := getBigqueryService()
if(err != nil) {
return "ERROR"
}
err2 := getTablesFromDataset(projectId, datasetId, bigqueryService)
if(err2 != nil) {
return "ERROR"
}
return "OK"
}
|
Runs API for selected projectId and datasetId to obtain and list table metadata |
Delete tables older than N days from table name _yyyymmdd
func deleteTable(projectId string, datasetId string, tableId string, bigqueryService *bigquery.Service)(error) {
call := bigqueryService.Tables.Delete(projectId, datasetId, tableId)
err := call.Do()
if err != nil {
return err
}
return nil
}
func deleteTablesOlderThan(ndays int, projectId string, datasetId string, tablePrefix string, bigqueryService *bigquery.Service)(error) {
//figure out the cutoff date, any tables older than this cutoff date will be deleted
now := time.Now()
//go back n days and format to string yyyymmdd
cutOffDate := now.Add(time.Duration(-1 * ndays * 24) * time.Hour).Format("20060102")
xPrintf("cutOffDate: %v\n", cutOffDate)
//go through all tables from the dataset
var pagetoken string = ""
//define the table pattern to look for: tablePrefix_yyyymmdd
reg := regexp.MustCompile(`^` + tablePrefix + `_(\d\d\d\d\d\d\d\d)$`)
//collect ids of the tables to be deleted
toDelete := make([]string, 0, 250)
for {
call := bigqueryService.Tables.List(projectId, datasetId)
if len(pagetoken) > 0 {
call = call.PageToken(pagetoken)
}
list, err := call.Do()
if err != nil {
return err
}
for _, table := range list.Tables {
//parse out the tableId, strip out projectId and datasetId from it
tableid := table.Id
del := strings.LastIndex(tableid, ".")
if del != -1 {
tableid = tableid[del+1:]
}
//check for the table pattern
matches := reg.FindStringSubmatch(tableid)
if(len(matches) > 0) {
//get date yyyymmdd from the table name
tdate := matches[1]
//if it is older than the cutoff date, delete the table
if(tdate < cutOffDate) {
toDelete = append(toDelete, tableid)
}
}
}
pagetoken = list.NextPageToken
if len(pagetoken) == 0 {
break
}
}
if(len(toDelete) > 0) {
xPrintf("Number of tables to be deleted: %v\n", len(toDelete))
for i, tableid := range toDelete {
xPrintf("========= Table %v ==============\n", i)
xPrintf("Id: %v:%v.%v\n", projectId, datasetId, tableid)
errd := deleteTable(projectId, datasetId, tableid, bigqueryService)
if errd != nil {
return errd
}
xPrintln("deleted")
}
}
return nil
}
func mainReturn() interface{} {
projectId := "projectId"
datasetId := "datasetId"
tablePrefix := "tablePrefix"
olderThanDays := 40
bigqueryService, err := getBigqueryService()
if(err != nil) {
return "ERROR:" + err.Error()
}
errd := deleteTablesOlderThan(olderThanDays, projectId, datasetId, tablePrefix, bigqueryService)
if(errd != nil) {
return "Delete ERROR:" + errd.Error()
}
return "OK"
}
|
Specify table with prefix _YYYYMMDD and days where tables older than that day is deleted |
Expire tables by date in table Id
func getMetaFromBQTable(projectId string, datasetId string, tableId string, bigqueryService *bigquery.Service)(*bigquery.Table, error) {
//https://godoc.org/google.golang.org/api/bigquery/v2#TablesService
call := bigqueryService.Tables.Get(projectId, datasetId, tableId)
table, err := call.Do()
if err != nil {
return nil, err
}
return table, nil
}
func setTableExpiration(projectId string, datasetId string, tableId string, expireMillisecond int64, bigqueryService *bigquery.Service)(error) {
//first get table metadata
table, err := getMetaFromBQTable(projectId, datasetId, tableId, bigqueryService)
if err != nil {
return err
}
table.ExpirationTime = expireMillisecond
call := bigqueryService.Tables.Patch(projectId, datasetId, tableId, table)
_, errp := call.Do()
if errp != nil {
return errp
}
return nil
}
func setTableExpirationBasedOnDateInTableId(ndays int, expirationTime string, projectId string, datasetId string, tablePrefix string, bigqueryService *bigquery.Service)(error) {
localLocation := time.Now().Location()
//loop through all tables from the dataset
var pagetoken string = ""
reg := regexp.MustCompile(`^` + tablePrefix + `_(\d\d\d\d\d\d\d\d)$`)
i := 0
for {
call := bigqueryService.Tables.List(projectId, datasetId)
if len(pagetoken) > 0 {
call = call.PageToken(pagetoken)
}
list, err := call.Do()
if err != nil {
return err
}
for _, table := range list.Tables {
//parse out the tableId, strip out projectId and datasetId
tableid := table.Id
del := strings.LastIndex(tableid, ".")
if del != -1 {
tableid = tableid[del+1:]
}
//check for table pattern: tablePrefix_yyyymmdd
matches := reg.FindStringSubmatch(tableid)
if(len(matches) > 0) {
xPrintf("========= Table %v ==============\n", i)
xPrintf("Id: %v:%v.%v\n", projectId, datasetId, tableid)
//convert string yyyymmdd to type time in local timezone
tdate, err := time.ParseInLocation("20060102 15:04:05", matches[1] + " " + expirationTime, localLocation)
if(err != nil) {
return err
}
//calculate the expiration date for this table
//dial it forward by n days
expire := tdate.Add(time.Duration(ndays * 24) * time.Hour)
xPrintln("tdate: ", tdate)
xPrintln("expire: ", expire)
//set expiration
errx := setTableExpiration(projectId, datasetId, tableid, expire.Unix()*1000, bigqueryService)
if(errx != nil) {
return errx
}
xPrintf("Set to expire:%v\n", expire)
}
}
pagetoken = list.NextPageToken
if len(pagetoken) == 0 {
break
}
}
return nil
}
func mainReturn() interface{} {
projectId := "projectId"
datasetId := "datasetId"
tablePrefix := "tablePrefix"
expirationTime := "04:00:00"
ndays := 60
bigqueryService, err := getBigqueryService()
if(err != nil) {
return "ERROR:" + err.Error()
}
errx := setTableExpirationBasedOnDateInTableId(ndays, expirationTime, projectId, datasetId, tablePrefix, bigqueryService)
if(errx != nil) {
return "setTableExpirationBasedOnDateInTableId Error:" + errx.Error()
}
return "OK"
}
|
For tables with prefix and older than specified day set expiration time |
Insert object into CS
func mainReturn() interface{} {
storageService, err := getStorageService()
if(err != nil) {
return "ERROR"
}
var bucket = "cs_bucket"
var objectId = "filename.txt"
var content = "apple is blue"
storageObject, err := storageService.Objects.Insert(bucket, &storage.Object{Name: objectId}).Media(strings.NewReader(content)).Do()
if err != nil {
return "ERROR"
}
return storageObject.Name
}
|
Create an object (file) and save it to cloud storage |
List objects from CS
func mainReturn() interface{} {
storageService, err := getStorageService()
if(err != nil) {
return "ERROR"
}
var bucket = "sc_bucket"
var objPrefix = "objectname"
// Get the list of objects in the bucket matching the specified prefix.
list := storageService.Objects.List(bucket)
list.Prefix(objPrefix)
objects, err := list.Do()
if err != nil {
return "ERROR"
}
//https://godoc.org/google.golang.org/api/storage/v1#Objects
//loop just 20 objects
var results string = ""
for i := 0; i < 20; i++ {
results = results + xSprintf("object %v == %v,", i, objects.Items[i].Name)
}
return results
}
|
List objects in cloud storage bucket |
