在上传到 BigQuery 之前对 GCS 文件进行非规范化处理

Question

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

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

如果性能(速度(和成本(货币(是我的目标，那么在这种情况下写入 BigQuery 的最佳方式是什么？在非规范化之前，这些文件每个大约为 10KB。每行大约为 1000 个字符。非规范化后，大约是其三倍。在 BigQuery 中成功加载非规范化文件后，我不需要保留这些文件。我担心性能，以及有关插入/写入的任何特定 BigQuery 每日配额。我认为没有，除非你正在做DML语句，但如果我错了，请纠正我。

Answer 1

我会使用当您将文件上传到存储桶时触发的云函数。

这是很常见的，谷歌有一个回购教程，专门针对JSON文件使用云函数将数据从云存储流式传输到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。