python - 如何使用 GCS compose 方法正确连接 CSV 文件？

Question

我正在尝试使用 GCS compose() 方法将多个 CSV 组合成一个 CSV。
目标是使用 Python 的多处理模块来处理大型 CSV 文件，而无需将所有数据存储在一个地方，直到它到达 GCS。目的是在对这些大文件进行操作时使用尽可能少的内存。例如:文件被分成 4 个部分，每个子进程一个，对数据完成工作，每个子进程将文件的 1/4 上传到 GCS，其中 compose 方法将组合四个文件中的每一个以创建一个 CSV。
Pandas 不起作用，因为我试图避免将数据全部放在一个地方(消耗了太多内存)。
以下是我在最后尝试合并 CSV 时遇到的问题。
CSV 1:

col1,col2,col3
1,2,3
1,2,3
1,2,3

CSV 2:

col4,col5,col6
4,5,6
4,5,6
4,5,6

当我使用 GCS API 中的 compose() 方法时，我在目标文件中得到了这个结果:

col1,col2,col3
1,2,3
1,2,3
1,2,3
col4,col5,col6
4,5,6
4,5,6
4,5,6

但我正在寻找的是:

col1,col2,col3,col4,col5,col6
1,2,3,4,5,6
1,2,3,4,5,6
1,2,3,4,5,6

产生的代码:

bucket = STORAGE_CLIENT.bucket(bucket_name)
destination = bucket.blob(destination_blob_name)
destination.content_type = "text/csv"
destination.compose(sources)

有没有人对我如何以我想要的方式合并 CSV 有任何建议？

Answer 1

因为 Cloud Storage compose 无法实现你想要的(compose 很笨，它需要一个文件并将以下内容附加到前一个文件中!)，如果你有大量数据，我建议你使用大查询!

在您拥有的每种文件类型上创建一个联合表

执行 extract data 查询

EXPORT DATA OPTIONS(
  uri='gs://my_bucket/*.csv',
  format='CSV',
  overwrite=true,
  header=true,
  field_delimiter=',') AS
SELECT * FROM `my_federated_table` JOIN/UNION ....

您将在输出中有多个文件，但格式正确。如果您只想输出一个文件，请将 header 设置为 false 并使用 compose 合并文件。