将 csv 从 gcs 迁移到 postgresql

Question

我正在尝试使用 python 脚本将 csv 文件从已从 BigQuery 导出的 Google Cloud Storage (GCS) 迁移到 PostgreSQL Google cloud sql 实例。

我希望使用 Google API，但在 the documentation 中找到了这个：

PostgreSQL 实例不支持使用 Cloud SQL Admin API 导入 CSV 数据。

作为替代方案，我可以使用 psycopg2 库并将 csv 文件的行流式传输到 SQL 实例中。我可以通过三种方式做到这一点

• 逐行：读取每一行然后提交插入命令再提交
• 批处理流：读取每一行，然后提交插入命令，然后在 10 行或 100 行等之后提交。
• 整个 csv：读取每一行并提交插入命令，然后仅在文档末尾提交。

我担心这些 csv 文件可能包含数百万行，并且为上述三个选项中的任何一个运行此过程对我来说似乎是个坏主意。

我有什么选择？ 本质上，我在 BigQuery 中有一些原始数据，我们在导出到 GCS 之前对其进行了一些预处理，以准备导入到 PostgreSQL 实例。 我需要将这些预处理数据从 BigQuery 导出到 PostgreSQL 实例。

这不是 question 的副本，因为我最好寻找将数据从 BigQuery 导出到 PostgreSQL 实例的解决方案，无论是通过 GCS 还是直接。

Answer 1

您可以按照@GrahamPolley 的建议使用Cloud Dataflow 进行导入过程。确实，此解决方案涉及一些额外的工作（熟悉 Dataflow、设置所有内容等）。即使有额外的工作，这将是您的情况的首选解决方案。但是，还有其他解决方案可用，我将在下面解释其中之一。

要使用 Dataflow 设置迁移过程，这个关于 exporting BigQuery to Google Datastore 的教程就是一个很好的例子

---

Cloud Dataflow 的替代解决方案

Cloud SQL for PostgreSQL 不支持从 .CSV 导入，但支持 .SQL 文件。

指定 uri 的文件类型。
SQL：该文件包含 SQL 语句。
CSV：文件包含 CSV 数据。 PostgreSQL 实例不支持使用 Cloud SQL Admin API 导入 CSV 数据。

直接的解决方案是使用一些工具将.CSV 文件转换为.SQL（Google 没有提供我知道的工具，但网上有很多），然后导入到 PostgreSQL。

如果您想以更“程序化”的方式实施此解决方案，我建议使用Cloud Functions，这是我将如何尝试的示例：

1. 设置triggers when a file is uploaded to a Cloud Storage bucket的云函数
2. 对函数进行编码以获取上传的文件并检查它是否为.CSV。如果是，请使用 csv-to-sql API (example of API here) 将文件转换为 .SQL
3. 将新文件存储在 Cloud Storage 中
4. 导入到 PostgreSQL

Answer 2

我发现pyscopg2 模块有 copy_from() 允许加载整个 csv 文件而不是单独流式传输行。 使用这种方法的缺点是csv文件仍然需要从GCS下载并存储在本地。

这里是使用pyscopg2'copy_from()'的细节。 （来自here）

import psycopg2

conn = psycopg2.connect("host=localhost dbname=postgres user=postgres")
cur = conn.cursor()
with open('user_accounts.csv', 'r') as f:
    # Notice that we don't need the `csv` module.
    next(f)  # Skip the header row.
    cur.copy_from(f, 'users', sep=',')

conn.commit()

Answer 3

您可以只使用一个类来使您从互联网上提取的文本表现得像一个文件。我已经使用过几次了。

import io
import sys


class IteratorFile(io.TextIOBase):
    """ given an iterator which yields strings,
    return a file like object for reading those strings """

    def __init__(self, obj):
        elements = "{}|" * len(obj[0])
        elements = (unicode(elements[:-1]).format(*x) for x in obj)
        self._it = elements
        self._f = io.cStringIO()

    def read(self, length=sys.maxsize):

        try:
            while self._f.tell() < length:
                self._f.write(next(self._it) + "\n")

        except StopIteration as e:
            # soak up StopIteration. this block is not necessary because
            # of finally, but just to be explicit
            pass

        except Exception as e:
            print("uncaught exception: {}".format(e))

        finally:
            self._f.seek(0)
            data = self._f.read(length)

            # save the remainder for next read
            remainder = self._f.read()
            self._f.seek(0)
            self._f.truncate(0)
            self._f.write(remainder)
            return data

    def readline(self):
        return next(self._it)

Answer 4

在开始之前，您应该确保：

您要导入的数据库和表必须 您的 Cloud SQL 实例上已存在。

CSV 文件格式要求 CSV 文件每行必须有一行 数据并具有逗号分隔的字段。

然后，您可以在接下来的步骤中import data to a Cloud SQL instance using a CSV file 出现在 GCS 存储桶中 [GCLOUD]

1. 描述您要从中导出的实例：

gcloud sql instances describe [INSTANCE_NAME]

1. 复制 serviceAccountEmailAddress 字段。

2. 将服务帐号作为写入者添加到存储桶 ACL：

gsutil acl ch -u [SERVICE_ACCOUNT_ADDRESS]:W gs://[BUCKET_NAME]

4. 将服务帐号作为阅读器添加到导入文件中：

gsutil acl ch -u [SERVICE_ACCOUNT_ADDRESS]:R gs://[BUCKET_NAME]/[IMPORT_FILE_NAME]

5. 导入文件

gcloud sql import csv [INSTANCE_NAME] gs://[BUCKET_NAME]/[FILE_NAME] \ --database=[DATABASE_NAME] --table=[TABLE_NAME]

6. 如果您不需要保留之前设置的 ACL 提供的权限，请移除该 ACL：

gsutil acl ch -d [SERVICE_ACCOUNT_ADDRESS] gs://[BUCKET_NAME]