python - 如何使用 Cloud composer 将大数据从 Postgres 导出到 S3？

Question

我一直在使用 Postgres 到 S3 运算符将数据从 Postgres 加载到 S3。但是最近，我不得不导出一个非常大的表，而我的 Airflow composer 失败了，没有任何日志，这可能是因为我们正在使用 Python 的 tempfile 模块的 NamedTemporaryFile 函数来创建一个临时文件，并且我们正在使用这个临时文件加载到 S3 .由于我们使用的是 Composer，这将被加载到 Composer 的本地内存中，并且由于文件的大小非常大，所以它会失败。

引用这里:https://cloud.google.com/composer/docs/how-to/using/troubleshooting-dags#task_fails_without_emitting_logs

我确实检查了 RedshiftToS3 运算符，因为它也使用了 Postgres Hook ，并且它有几个可以轻松加载大文件的卸载选项，但我意识到 Redshift 和 Postgres 之间没有一对一的对应关系。所以那是不可能的。有什么办法可以拆分我的 Postgres 查询吗？现在我正在执行 SELECT * FROM TABLENAME 另外，我没有关于该表的任何信息。

我也遇到过这个类似的运算符:https://airflow.apache.org/docs/stable/_modules/airflow/contrib/operators/sql_to_gcs.html

这里有一个参数approx_max_file_size_bytes:

This operator supports the ability to split large table dumps into multiple files (see notes in the filename param docs above). This param allows developers to specify the file size of the splits.

我从代码中了解到，当大小超过给定限制时，他们正在创建一个新的临时文件，那么他们是否将文件拆分为多个临时文件，然后分别上传？

编辑:我将再次准确解释我要做什么。目前，Postgres 到 S3 操作符创建一个临时文件并将游标返回的所有结果写入此文件，这会导致内存问题。所以我在想的是，我可以添加一个 max_file_size 限制，对于游标中的每一行，我会将结果写入我们的临时文件，如果我们的临时文件的大小超过我们设置的 max_file_size 限制，我们将写入我们的内容文件到 S3，然后刷新或删除该文件，然后创建一个新的临时文件，并将光标的下一行写入该文件，并将该文件也上传到 S3。我不确定如何像那样修改运算符？

Answer 1

正如您已经发现的那样，这是因为您正在为表中的每一行构建一个字典，当您的表中有很多行时，您的机器内存就会耗尽。

你已经真正回答了你自己的问题:只写 a 直到文件达到一定大小，然后将文件推送到 S3。或者，您可以将文件保存在磁盘上并每隔 x 行刷新一次字典对象，但在这种情况下，您的文件在磁盘上而不是在内存中可能会变得非常大。