温馨提示:本文翻译自stackoverflow.com,查看原文请点击:google cloud storage - Denormalize a GCS file before uploading to BigQuery
google-bigquery google-cloud-run google-cloud-storage

google cloud storage - 在上传到BigQuery之前对GCS文件进行规范化

发布于 2020-04-04 00:03:02

我已经在.Net Core中编写了一个Cloud Run API,该API从GCS位置读取文件,然后进行非规范化(即,为每行添加更多信息以包括文本描述),然后将其写入BigQuery表。我有两个选择:

  1. 我的云运行API可以创建非标准化的CSV文件,并将它们写入另一个GCS位置。然后,另一个云运行API可以提取那些非规范化的CSV文件,并将其直接写入BigQuery。
  2. 我的云运行API可以读取原始CSV文件,对它们在内存(文件流)中进行规范化,然后以某种方式从内存中的文件流直接写入BigQuery表。

如果性能(速度)和成本(货币)是我的目标,那么在这种情况下写BigQuery的最佳方法是什么。在反规范化之前,这些文件每个大约为10KB。每行大约1000个字符。非正规化后,它大约是原来的三倍。在BigQuery中成功加载非规范化文件后,我不需要保留它们。我担心性能以及围绕插入/写入的任何特定BigQuery每日配额。除非您正在执行DML语句,否则我认为不会有任何问题,但是如果我做错了,请更正我。

查看更多

提问者
AIK DO
被浏览
84
Juancki 2020-01-31 20:05

我将使用将文件上传到存储桶时触发的云功能。

如此普遍,以至于Google 针对JSON文件为此提供了一个回购教程,该教程使用Cloud Functions将数据从Cloud Storage流式传输到BigQuery

然后,我将从以下示例main.py文件中进行修改

def streaming(data, context):
    '''This function is executed whenever a file is added to Cloud Storage'''
    bucket_name = data['bucket']
    file_name = data['name']
    db_ref = DB.document(u'streaming_files/%s' % file_name)
    if _was_already_ingested(db_ref):
        _handle_duplication(db_ref)
    else:
        try:
            _insert_into_bigquery(bucket_name, file_name)
            _handle_success(db_ref)
        except Exception:
            _handle_error(db_ref)

为此,可以接受CSV文件

import json
import csv
import logging
import os
import traceback
from datetime import datetime

from google.api_core import retry
from google.cloud import bigquery
from google.cloud import storage
import pytz



PROJECT_ID = os.getenv('GCP_PROJECT')
BQ_DATASET = 'fromCloudFunction'
BQ_TABLE = 'mytable'

CS = storage.Client()
BQ = bigquery.Client()


def streaming(data, context):
    '''This function is executed whenever a file is added to Cloud Storage'''
    bucket_name = data['bucket']
    file_name = data['name']

    newRows = postProcessing(bucket_name, file_name)

    # It is recommended that you save 
    # what you process for debugging reasons.
    destination_bucket = 'post-processed' # gs://post-processed/
    destination_name = file_name
    # saveRowsToBucket(newRows,destination_bucket,destination_name)
    rowsInsertIntoBigquery(newRows)



class BigQueryError(Exception):
    '''Exception raised whenever a BigQuery error happened''' 

    def __init__(self, errors):
        super().__init__(self._format(errors))
        self.errors = errors

    def _format(self, errors):
        err = []
        for error in errors:
            err.extend(error['errors'])
        return json.dumps(err)

def postProcessing(bucket_name, file_name):
    blob = CS.get_bucket(bucket_name).blob(file_name)
    my_str = blob.download_as_string().decode('utf-8')
    csv_reader = csv.DictReader(my_str.split('\n'))                                                                   
    newRows = []
    for row in csv_reader:
        modified_row = row # Add your logic
        newRows.append(modified_row)
    return newRows

def rowsInsertIntoBigquery(rows):
    table = BQ.dataset(BQ_DATASET).table(BQ_TABLE)
    errors = BQ.insert_rows_json(table,rows)
    if errors != []:
        raise BigQueryError(errors)

如果需要的话,仍然需要定义map(row-> newRow)和函数saveRowsToBucket

最新帖子

最新文章