CODEWITHSUNDEEP
pythonstreaminggoogle-bigquery
Sema196
Sema196

Reputation: 263

How to stream Csv file into BigQuery?

Examples I found so far is streaming json to BQ, e.g. https://cloud.google.com/bigquery/streaming-data-into-bigquery

How do I stream Csv or any file type into BQ? Below is a block of code for streaming and seems "issue" is in insert_all_data where 'row' defined as json.. thanks

# [START stream_row_to_bigquery]
def stream_row_to_bigquery(bigquery, project_id, dataset_id, table_name, row,
                           num_retries=5):
    insert_all_data = {
        'rows': [{
            'json': row,
            # Generate a unique id for each row so retries don't accidentally
            # duplicate insert
            'insertId': str(uuid.uuid4()),
        }]
    }
    return bigquery.tabledata().insertAll(
        projectId=project_id,
        datasetId=dataset_id,
        tableId=table_name,
        body=insert_all_data).execute(num_retries=num_retries)
    # [END stream_row_to_bigquery]

Upvotes: 3

Views: 1193

Answers (1)

Marlon Abeykoon
Marlon Abeykoon

Reputation: 12505

This is how I wrote using bigquery-python library very easily.

def insert_data(datasetname,table_name,DataObject):
          client = get_client(project_id, service_account=service_account,
                            private_key_file=key, readonly=False, swallow_results=False)

          insertObject = DataObject
          try:
              result  = client.push_rows(datasetname,table_name,insertObject)
          except Exception, err:
              print err
              raise
          return result

Here insertObject is a list of dictionaries where one dictionary contains one row.

eg: [{field1:value1, field2:value2},{field1:value3, field2:value4}]

csv can be read as follows,

import pandas as pd
fileCsv = pd.read_csv(file_path+'/'+filename, parse_dates=C, infer_datetime_format=True)
data = []
for row_x in range(len(fileCsv.index)):
    i = 0
    row = {}
    for col_y in schema:
        row[col_y['name']] = _sorted_list[i]['col_data'][row_x]
        i += 1
    data.append(row)
insert_data(datasetname,table_name,data)

data list can be sent to the insert_data

This will do that but still there's a limitation that I already raised here.

Upvotes: 2

Related Questions