从云存储重写具有云功能的 csv 文件并将其发送到 BigQuery

Question

我正在编写一个小型云函数 python 脚本来重写来自存储的 csv 文件（跳过一些列）并将其发送到 BigQuery。

我的脚本的 BigQuery 部分是这样的：

def bq_import(request):
    job_config.skip_leading_rows = 1
# The source format defaults to CSV, so the line below is optional.
    job_config.source_format = bigquery.SourceFormat.CSV
    uri = "gs://url.appspot.com/fil.csv"

    load_job = bq_client.load_table_from_uri(
        uri, dataset_ref.table('table'), job_config=job_config
    )  # API request

    load_job.result()  # Waits for table load to complete.

    destination_table = bq_client.get_table(dataset_ref.table('table'))

我发现这个脚本允许我通过跳过一些列来重写 csv：

def remove_csv_columns(input_csv, output_csv, exclude_column_indices):
    with open(input_csv) as file_in, open(output_csv, 'w') as file_out:
        reader = csv.reader(file_in)
        writer = csv.writer(file_out)
        writer.writerows(
            [col for idx, col in enumerate(row)
             if idx not in exclude_column_indices]
            for row in reader)

remove_csv_columns('in.csv', 'out.csv', (3, 4))

所以我基本上需要让这两个脚本在我的云功能中协同工作。但是我不确定我应该如何处理remove_csv_columns 函数，尤其是output_csv 变量。我应该创建一个空的虚拟 csv 文件吗？还是一个数组或类似的东西？如何即时重写此 csv 文件？

我认为我的最终脚本应该是这样的，但缺少一些东西......

uri = "gs://url.appspot.com/fil.csv"

def remove_csv_columns(uri, output_csv, exclude_column_indices):
    with open(input_csv) as file_in, open(output_csv, 'w') as file_out:
    reader = csv.reader(file_in)
    writer = csv.writer(file_out)
    writer.writerows(
            [col for idx, col in enumerate(row)
            if idx not in exclude_column_indices]
            for row in reader)

def bq_import(request):
    job_config.skip_leading_rows = 1
# The source format defaults to CSV, so the line below is optional.
    job_config.source_format = bigquery.SourceFormat.CSV
    csv_file = remove_csv_columns('in.csv', 'out.csv', (3, 4))

    load_job = bq_client.load_table_from_uri(
        csv_file, dataset_ref.table('table'), job_config=job_config
    )  # API request

    load_job.result()  # Waits for table load to complete.

    destination_table = bq_client.get_table(dataset_ref.table('table'))

基本上我认为我需要通过remove_csv_columns 在 bq_import 函数中定义我的 cvs 文件，但我不确定如何。

顺便说一句，我正在学习 python，但我不是开发专家。谢谢。

Answer 1

您的代码有很多错误，我会在更正中尽量清楚

uri = "gs://url.appspot.com/fil.csv"

我不知道你的函数是如何触发的，但通常要处理的文件包含在request 对象中，如this for event from GCS。使用存储桶和名称动态构建您的uri

def remove_csv_columns(uri, output_csv, exclude_column_indices):
    with open(input_csv) as file_in, open(output_csv, 'w') as file_out:

小心：你使用uri 作为函数参数名，你使用input_csv 以读取模式打开你的输入文件。因为input_csv 不存在，您的代码在这里崩溃了！

这里再说一句。 uri 是函数参数名称，仅在函数内部知道，并且与外部关系，除了填充此值的调用者。它与您在uri = "gs://url.appspot.com/fil.csv" 之前定义的全局变量绝对没有链接

    reader = csv.reader(file_in)
    writer = csv.writer(file_out)
    writer.writerows(
            [col for idx, col in enumerate(row)
            if idx not in exclude_column_indices]
            for row in reader)

def bq_import(request):
    job_config.skip_leading_rows = 1
# The source format defaults to CSV, so the line below is optional.
    job_config.source_format = bigquery.SourceFormat.CSV
    csv_file = remove_csv_columns('in.csv', 'out.csv', (3, 4))

您的输入文件是静态的。阅读我关于动态uri 建设的评论。

看remove_csv_columns函数：它什么都不返回，它只是在out.csv中写入一个新文件。因此，您的 csv_file 在这里不代表任何内容。另外，这个函数有什么作用？读取in.csv 文件并写入out.csv 文件（通过删除列）。您必须将文件传递给此函数

顺便说一句，您必须从 Cloud Storage 下载文件并将其存储在本地。在 Cloud Function 中，只有 /tmp 是可写的。因此你的代码应该是这样的

# create storage client
storage_client = storage.Client()
# get bucket with name
bucket = storage_client.get_bucket('<your bucket>')
# get bucket data as blob
blob = bucket.get_blob('<your full file name, path included')
# convert to string
data = blob.download_as_string()
# write the file
with open('/tmp/input.csv', 'w') as file_out:
  file_out.write(data )
remove_csv_columns('/tmp/input.csv', '/tmp/out.csv', (3, 4))

---

继续你的代码

    load_job = bq_client.load_table_from_uri(
        csv_file, dataset_ref.table('table'), job_config=job_config
    )  # API request

函数load_table_from_uri 将文件从存在于 Cloud Storage 中的文件加载到 BigQuery。在这里，它不是你的目标，你想将本地创建的文件out.csv 加载到函数中。正确的电话是load_job = bq_client.load_table_from_file(open('/tmp/out.csv', 'rb'), job_config=job_config)

    load_job.result()  # Waits for table load to complete.

    destination_table = bq_client.get_table(dataset_ref.table('table'))

然后，考虑清理 /tmp 目录以释放内存，注意 Cloud Function 超时，在您的 requirements.txt 文件中导入正确的库（至少 Cloud Storage 和 BigQuery 依赖项），最后获得蛋糕roles of your cloud function

---

但是，这只是为了提高您的 Python 代码和技能。无论如何，这个功能是没用的。

确实，有云功能，如前所述，你只能写在/tmp目录上。它是一个内存文件系统，Cloud Function 仅限于 2Gb 内存（包括文件和执行代码占用空间）。顺便说一句，您的输入文件的大小不能超过 800Mb 左右，对于小文件，更容易。

• 将原始文件加载到 BigQuery 的临时表中
• 像这样对您希望的列执行INSERT SELECT BigQuery 查询

INSERT INTO `<dataset>.<table>`
SELECT * except (<column to ignore>) from `<dataset>.<temporary table>`

• 删除您的临时表

由于您的文件很小（小于 1GB）并且因为 BigQuery 免费套餐是 5TB 的扫描数据（您只需为扫描的数据而不是处理付费，免费进行所有您想要的 SQL 转换），因此更容易与 Python 相比，将数据处理到 BigQuery 中。

函数处理时间会更长，您可以在函数而不是 BigQuery 上支付处理时间。