使用 Cloud Function 将 sqlite .db 文件数据从云存储写入/流式传输到 BigQuery

Question

我将 sqlite .db 文件存储在 Google Cloud Storage Bucket 中。这些文件每天上传。我想将 .db 文件中的特定表数据流式传输/写入 BigQuery 表。我试图编写一个云函数来连接 sqlite db 并获取 pandas 数据框中的记录并加载到 BQ 表。我的 python 程序无法运行，因为它无法在运行时连接到 sqlite db 并从中获取数据。

我的代码：

import sqlite3
import pandas as pd
from google.cloud import bigquery
from google.cloud import storage

def bqDataLoad(event, context):
    storage_client = storage.Client() 
    bucketName = 'my-bucket'
    blobName = 'mysqlite.db'
    bucket = storage_client.get_bucket(bucketName) 
    blob = bucket.blob(blobName)
    fileName = "gs://" + bucketName + "/" + blobName
        
    bigqueryClient = bigquery.Client()
    tableRef = bigqueryClient.dataset("ds").table("bqtable")
    cnx = sqlite3.connect(fileName) 
    dataFrame = pd.read_sql_query("SELECT * FROM sqlitetable", cnx)

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

我尝试将 sqlite3.connect(fileName) 传递给“cnx”，然后云函数出现无法连接的错误。 也试过 sqlite3.connect(blob) 但它给出了错误，因为字符串是预期的。

任何帮助将不胜感激。

谢谢

Answer 1

我相信您无法连接到数据库，因为您需要将其下载到运行云函数的实例。请注意，在 Cloud Functions 环境中，/tmp folder will be the only writeable directory 并且由于 Cloud Functions 是内存系统，因此在部署时分配给 Cloud Functions 的 RAM 将用于托管 .db 文件。

下面的代码 sn-p 应该可以工作（它基于著名的chinook SQLite sample database）。

假设您已经拥有created the table within the BigQuery dataset，其架构如下：

ArtistId    INTEGER   NULLABLE  
Name        STRING    NULLABLE  

基于 Cloud SQL 数据库的架构（因为要运行的特定查询取决于该架构）并且存储在 Cloud Storage 存储分区中的 .db 对象未嵌套在任何文件夹下。

1. 在本地开发设置中创建一个包含以下文件的目录：

一个。 requirements.txt

google-cloud-storage
google-cloud-bigquery
pandas
pyarrow

b. main.py

from google.cloud import storage
from google.cloud import bigquery
import sqlite3
import pandas as pd

BUCKET_NAME = "[YOUR-BUCKET]" #Change as per your setup
OBJECT_NAME = "chinook.db" #Change as per your setup
DATABASE_NAME_IN_RUNTIME = "/tmp/chinook.db" #Remember that only the /tmp folder is writable within the directory
QUERY = "SELECT * FROM artists;" #Change as per your query
TABLE_ID = "[PROJECT-ID].[DATASET-ID].[TABLE-ID]" # Change with the format your-project.your_dataset.your_table_name

storage_client = storage.Client()
bigquery_client = bigquery.Client()

# Fetch the .db file from Cloud Storage
def get_db_file_from_gcs(bucket_name, object_name, filename):
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(object_name)
    return blob.download_to_filename(filename)

#Makes the connections to the DB
def connect_to_sqlite_db(complete_filepath_to_db):
    connection = sqlite3.connect(complete_filepath_to_db)
    return connection

#Run the query and return a cursor to iterate over the results
def run_query_to_db(connection, query):
    with connection:
        cursor = connection.cursor()
        cursor.execute(query)
        return cursor.fetchall()

#Run the query and saves it to a dataframe
def run_query_to_db_with_pandas(connection, query):
    with connection:
        df = pd.read_sql_query(query, connection)
        return df

def gcssqlite_to_bq(request):
    print("Getting .db file from Storage")
    get_db_file_from_gcs(BUCKET_NAME, OBJECT_NAME, DATABASE_NAME_IN_RUNTIME)
    print("Downloaded .db file in CF instance RAM")
    print("Trying to connect to database using sqlite")
    cnx = connect_to_sqlite_db(DATABASE_NAME_IN_RUNTIME)
    print("Connected to database")
    print("Attempting to perform a query")
    results = run_query_to_db_with_pandas(cnx, QUERY)
    print("Writing data to BigQuery")
    bigqueryJob = bigquery_client.load_table_from_dataframe(results, TABLE_ID)
    bigqueryJob.result()
    print("The Job to write to Big Query is finished")
    return "Executed Function"

1. 根据您的要求更新 main.py 文件。

2. 使用以下命令通过根据需要更改参数来部署可公开调用的云功能：

gcloud functions deploy [CLOUD_FUNCTION_NAME] --region [REGION] --entry-point gcssqlite_to_bq --timeout 540 --memory 1024MB --runtime python38 --trigger-http --allow-unauthenticated

免责声明：这种方法容易出现 OOM 错误，具体取决于 .db 文件的大小和向数据库发出的查询的复杂性。请注意，这是memory limits offered by Cloud Functions（当前为 4GB），如果您的 .db 文件中有大量数据（按 GB 或 TB 的顺序），则可能更好的方法是将数据迁移到 Cloud SQL（SQLite但不支持）并使用federated queries directly from BigQuery。