Простой ETL для выгрузки не семплированных данных из Google Analytics прямо в DWH BigQuery на python

c = GAClient(json_keyfile="etl1.json.json")
 
date = '2020-06-06'
 
time.sleep(240)
 
request_body = {
    'view_id': '77255247',
    'start_date': date,
    'end_date': date,
    'dimensions': {
        'ga:dimension4',
        'ga:dimension1',
        'ga:dimension2',
        'ga:eventCategory',
        'ga:eventAction',
        'ga:eventLabel'
         
 
    },
    'metrics': {
        'ga:users',
    },
    'filter': 'ga:nthHour==000000,ga:nthHour==000001,ga:nthHour==000002,ga:nthHour==000003,ga:nthHour==000004,ga:nthHour==000005,ga:nthHour==000006
         
}
 
response = c.get_all_data(request_body)
response['info']
ga_data = response['data']
ga_data['dimension1'] = ga_data['dimension1'].str.replace('.','-')
ga_data.to_csv("events_info{}-ind00-06.csv".format(date), index = False)
now = datetime.now()
current_time = now.strftime("%H:%M:%S")
print('Job finished time {} - GA-get and transform events_info{}-ind00-06.csv'.format(current_time, date))

##### backUp

bucket_uri = 'gs://analytics_1'
bucket_name = 'analytics_1'
#адрес куда записываем данные
bucket_target = "datasets/Trans{}.csv".format(date)
#файл подкачки
local_dataset = "Trans{}.csv".format(date)
#полный адрес куда записываем данные
bucket_target_uri = bucket_uri + '/' + bucket_target
#dataset
bigquery_dataset = 'uploadtest'
#таблица
bigquery_table = 'Trans'

def upload_blob(bucket_name, source_file_name, destination_blob_name):
    """Upload a CSV to Google Cloud Storage.
 
    1. Retrieve the target bucket.
    2. Set destination of data to be uploaded.
    3. Upload local CSV.
    """
    storage_client = storage.Client.from_service_account_json(
        'tactical-gate-storage-credit-account-app-super.json')
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    # Commence Upload
    blob.upload_from_filename(source_file_name)
    print('File {} uploaded to {}.'.format(
        source_file_name,
        destination_blob_name))
 
### Big Query Insert
 
def insert_bigquery(target_uri, dataset_id, table_id):
    """Insert CSV from Google Storage to BigQuery Table.
 
    1. Specify target dataset within BigQuery.
    2. Create a Job configuration.
    3. Specify that we are autodetecting datatypes.
    4. Reserve row #1 for headers.
    5. Specify the source format of the file (defaults to CSV).
    6. Pass the URI of the data storage on Google Cloud Storage from.
    7. Load BigQuery Job.
    8. Execute BigQuery Job.
    """
    bigquery_client = bigquery.Client.from_service_account_json(
        'tactical-gate-storage-credit-account-app-super.json')
    dataset_ref = bigquery_client.dataset(dataset_id)
    job_config = bigquery.LoadJobConfig()
    job_config.autodetect = True
    job_config.skip_leading_rows = 1
    job_config.source_format = bigquery.SourceFormat.CSV
    uri = bucket_target_uri
    load_job = bigquery_client.load_table_from_uri(
        uri,
        dataset_ref.table(table_id),
        job_config=job_config) # API request
    print('Starting job {}'.format(load_job.job_id))
    # Waits for table load to complete.
    load_job.result()
    print('Job finished Trans{} data loaded into BQ'.format(date))
 
 
upload_blob(bucket_name, local_dataset, bucket_target)
insert_bigquery(bucket_target_uri, bigquery_dataset, bigquery_table)