将云存储中的文本文件 (.txt) 加载到大查询表中

Posted 2023-03-24

【中文标题】将云存储中的文本文件 (.txt) 加载到大查询表中

【英文标题】：loading a text files (.txt) in cloud storage into big query table

【发布时间】：2020-07-31 06:28:37

【问题描述】：

我有一组文本文件，每 5 分钟上传一次到谷歌云存储中。我想每 5 分钟将它们放入 BigQuery（因为文本文件每 5 分钟上传到 Cloud Storage）。我知道无法将文本文件上传到 BigQuery。最好的方法是什么？

文本文件示例

提前致谢。

【参考方案1】：

他是另一种方法，它将使用基于事件的Cloud Function 将数据加载到 BigQuery 中。创建一个以"Trigger Type" 为云存储的云函数。一旦文件/文件加载到云存储桶中，它将调用/触发云函数事件，并将云存储中的数据加载到 BigQuery 中。

import pandas as pd
from google.cloud import bigquery

def bqDataLoad(event, context):
    bucketName = event['bucket']
    blobName = event['name']
    fileName = "gs://" + bucketName + "/" + blobName
    
    bigqueryClient = bigquery.Client()
    tableRef = bigqueryClient.dataset("bq-dataset-name").table("bq-table-name")

    dataFrame = pd.read_csv(fileName)

    bigqueryJob = bigqueryClient.load_table_from_dataframe(dataFrame, tableRef)
    bigqueryJob.result()

【讨论】：

谢谢。这是迄今为止最好的方法。但我有几个问题......我的文件是文本。我添加了要提问的文件样本。那么如何将其转换为 CSV？

在文本文件中，有几个主题。所以首先我必须选择一个主题并将其写回 CSV 然后 bigquery（不同的主题有不同的数据字段。Ex 主题 cat 有两列，Bat 有三列）

默认情况下，它需要一个逗号分隔的文件，但您可以使用您选择的任何分隔符。从文件读取时请使用分隔符选项。示例：dataFrame = pd.read_csv(fileName, sep="|")

是的。但是我们不能使用 pandas.read_csv() 读取文本文件吗？你有什么想法来转换这个吗？

您可以通过提供分隔符按原样使用分隔文本文件。示例：dataFrame = pd.read_csv("file.txt", sep="|")

【参考方案2】：

您可以利用BigQuery transfers。

Create an empty BigQuery table 带有架构（编辑为文本）文本：STRING 将 .txt 文件转换为 .csv 文件 从 Google Cloud Storage 创建 BigQuery 传输 将您的 .csv 文件上传到 GCS 存储桶中 检查您的转移是否成功

目前，此服务每小时传输新添加的文件，1h minimum file age limitation 即将发送到be removed。

该服务检查是否存在自上传到存储桶中超过 1 小时的新文件，例如：

text1.csv 于 4:46 上传 text2.csv 于 5:01 上传 text3.csv 于 5:06 上传 结果： 5:00 的传输运行不会传输任何文件 6:00 的传输运行将传输 text1.csv 7:00 的传输运行将传输 text2.csv 和 text3.csv

对于第 2 步，您需要将文本文件处理为 be accepted by BigQuery。我认为最简单的方法是使用 .csv 文件。 按如下方式编辑您的 .txt 文件：

在文本的开头和结尾添加字符“，例如“我要去市场买菜。” “另存为”文件为 text1.csv 将文件命名为具有相同的开头字符，例如text[...].csv 以便能够使用wildcards 为您的下一个文件重复此操作（text2.csv、text3.csv、text4.csv ...）

您还需要确保以下几点：

您的文本中不包含 " 字符 - 将它们替换为 ' 字符 确保您的整个文本内联为换行符 (EOF) are not supported

对于第 3 步，请在下面找到建议的 transfer configurations：

计划选项：

自定义 --> 每 1 小时

云存储 URI：

你的存储桶/你的文件夹/文本*

传输将提取所有以名称文本开头的文件

写作偏好：

追加

文件格式：

CSV

对于第 5 步，如果传输成功，则每小时verify in the Transfer details page。如果出现错误，则不会传输整批文件。使用CLI（请参阅下面的命令）获取有关哪个文件有问题以及错误性质的信息。您需要从存储桶中删除相应的文件，更正并重新上传。

bq --format=prettyjson show -j [bqts_...]

也可以preview your BigQuery table查看您传输的文本。

【讨论】：

谢谢。您是否有任何想法将位于云存储中的一组 .txt 文件批量转换为 .csv 文件。它们都采用相同的格式。

我认为您可以运行此脚本***.com/a/41323026/9398998 来添加字符，并根据此帖子unix.stackexchange.com/questions/32100/… 以新名称将它们保存在存储桶的新文件夹中。您可能希望使用“bash”和“shell”标签开始一个新问题以获得自定义回复。

【参考方案3】：

这是迄今为止尚未建议的替代选项：可以在 pandas 中打开 txt 文件（例如在 GCP AI Notebooks 中），然后使用 pandas 函数pandas.DataFrame.to_gbq 上传它。这非常简单，并在此处详细描述。如果要追加，我发现这是最佳解决方案，因为没有必要描述架构。