使用 Cloud Functions 将 Excel 文件从 FTP 提取到 BigQuery

Question

我正在创建一个自动脚本来从 FTP 下载文件并将它们存储到 BigQuery。

问题在于 BigQuery 仅接受 .csv 文件。出于这个原因，我正在寻找同时处理 .xlsx 和 .xls 文件的方法，条件是我计划将此批处理代码放在云端。

我提到后者是将.xlsx文件转换为.csv的一种方法是使用类似的东西：

import pandas as pd
data_xls = pd.read_excel('file_on_ftp.xlsx')
data_xls.to_csv('csvfile.csv', encoding='utf-8', index=False)

但是，这会在临时存储的某处创建本地文件。显然，使用 Cloud Functions 之后，我必须监控文件是否已被删除，这使得当一个云功能崩溃时它不可靠。

因此有没有更好的方法来处理 .xlsx 加载到 BigQuery 中？或者这是要走的路？

Answer 1

我的好方法是通过 Cloud Functions 将文件从 FTP 提取到 GCS。就像提到的@Dustin 一样，您可以将数据从 GCS 流式传输到 BigQuery

这是一种通过 Cloud Functions 从 FTP 提取文件的方法

#import libraries
from google.cloud import storage
import wget


def importFile(request):

 #set storage client
 client = storage.Client()

 # get bucket
 bucket = client.get_bucket('BUCKET-NAME') #without gs://
 blob = bucket.blob('file-name.csv')

 #See if file already exists
 if blob.exists() == False:

    #copy file to google storage
    try:
        link = 'ftp://account:password@ftp.domain.com/folder/file.csv' #for non-public ftp files
        ftpfile = wget.download(link, out='/tmp/destination-file-name.csv') #save downloaded file in /tmp folder of Cloud Functions
        blob.upload_from_filename(ftpfile)
        print('Copied file to Google Storage!')

    #print error if file doesn't exists
    except BaseException as error:
        print('An exception occurred: {}'.format(error))

 #print error if file already exists in Google Storage
 else:
    print('File already exists in Google Storage') 

Answer 2

您可能会对最近发布的本指南感兴趣："Streaming data from Cloud Storage into BigQuery using Cloud Functions"。

一般架构是：

1. 将有问题的文件从 FTP 上传到 Cloud Storage
2. 您的 Cloud Function 从 Cloud Storage 接收上传事件
3. 您的 Cloud Function 会将文件加载到内存中（磁盘上没有存储空间）
4. 您的 Cloud Function 函数将数据流式传输到 BigQuery

我不确定#1 是否适合您确保文件不会遗留在某处的需要，但我认为如果您在假设文件需要上传到 GCP 的情况下进行操作，这可能是最好的解决方案某处（替代方法是直接从您的本地计算机或您控制的实例流式传输到 BigQuery）。