是否可以从 Python 中的不同进程编写相同的 Parquet 文件夹？

我用fastparquet.

它似乎有效，但我想知道_metadata如果两个进程同时写入文件，文件如何不会发生冲突。

另外，为了使其正常工作，我必须使用ignore_divisions=True这对于稍后在阅读 Parquet 文件时获得快速性能来说并不理想，对吗？

我从位于同一文件夹中的多个 Parquet 文件创建了一个 Parquet 文件。每个文件对应一个分区。

Parquet 文件是在不同的进程中创建的（使用 Python concurrent.futures）。这是我在一个进程中运行的代码示例：

df最多包含100e3行（和 22 列）并在整数索引（称为cid）上编制索引。

然后我使用以下方法创建了两个元数据文件：

确实_metadata并且_common_metadata在包含所有 Parquet 文件的文件夹中正确创建。

我天真地认为，因为数据被索引和/或它具有元数据文件，所以获取数据大小等基本信息应该很快。例如，以下内容需要永远：

那除外吗？

另请注意，大多数列是int64，float64其中少数是object(string不同大小的。

如何有效地仅读取托管在云 blob 存储（例如 S3 / Azure Blob 存储）中的 parquet 文件的某些列？

列式结构是 parquet 文件格式的主要优势之一，因此选择性地读取列可以减少 I/O 负载。将数据存储在 blob 存储中以在云上运行大规模工作负载也是很自然的。但是，一旦将 parquet 文件存储为 blob，大多数库（dask、fastparquet、pyarrow）就无法真正利用这一点，因为底层fseek实际上不可能直接在 blob 上 - 这意味着无论选择哪一列在读取之前必须将整个文件下载到本地文件系统。

因此，如果我的用例是单独的应用程序需要不同的列，并且只为几列下载整个文件的性能成本是不可接受的，那么最佳实践是什么？我是否应该为每列存储不同的镶木地板文件以及一个公共索引，然后使用 pandas/dask 等在应用程序级别合并？apache parquet 格式是否具有按列拆分数据集的一些内置支持 - 类似于 hive 格式按分区然后按分区拆分的方式？

感谢任何使用 dask 或 fastparquet 的帮助/具体示例。

我正在尝试指定文件夹权限，然后dask.dataframe.to_parquet()与partition_on参数一起使用，因为此参数为提供的 DataFrame 列中的每个组创建一个文件夹。

我已经尝试了以下代码，但没有任何运气。在这里，我设置了传递给的“主”文件夹的模式，to_parquet使得文件夹权限为drwxrws---（由 ls -la 显示），但创建的partition_on文件夹具有文件夹权限drwxr-sr-x：

注意：任务是 dask.delayed，因为我需要将镶木地板文件保存在与 Dask 调度程序/工作程序所在的同一台远程计算机上。

我想设置文件夹权限，因为我的 Dask 调度程序/工作人员正在使用与我的用户在同一组中的 Unix 用户，从而使我的用户能够删除 Dask 工作人员创建的文件夹，如果模式和组权限（27xx）已配置。

详细说明：文件夹的模式和权限设置为“2770”。SGID“2”确保文件夹中的新文件将使用与文件夹本身相同的组权限创建。umask "770" 确保用户和组成员都具有对该文件夹及其文件的读取、写入和执行权限，而所有其他用户都没有权限。

解决方案

正如@mdurant 指出的那样，首先正常保存文件，然后再更改权限。下面的新代码解决了我的文件夹权限问题：

我正在尝试将数据帧保存到镶木地板，然后将它们加载到红移中。为此，我执行以下操作：

然后我使用“COPY”命令将保存的文件直接加载到redshift中：

它导致以下错误： write path: has an incompatible Parquet schema for column ...

如果我对 .csv 应用相同的程序，它工作得很好。切换到镶木地板时出现问题的原因是什么？

与此类似的问题。但是我的错误报告
ModuleNotFoundError: No module named 'fastparquet'
当我conda list在同一个虚拟环境下运行时，我得到了 但是，当我在“cmd”提示符下的交互式对话中时，我可以直接调用。（见下文）
import fastparquetpython

我在数据框中有一些时间序列数据，以时间为索引。对索引进行排序，数据存储在多个 parquet 文件中，每个文件中有一天的数据。我使用 dask 2.9.1

当我从一个镶木地板文件加载数据时，分区设置正确。

当我从多个文件加载数据时，我没有在生成的 dask 数据框中得到这些定义。

下面的例子说明了这个问题：

在这里，我得到 2 个分区和(None, None, None)分区

我可以让 dd.read_parquet 将分区设置为实际值吗？

更新

在我的实际数据中，我有一个镶木地板文件 pr day。

这些文件是通过保存数据帧中的数据来创建的，其中时间戳用作索引。索引已排序。每个文件的大小为 100-150MB，当加载到内存时，它使用应用程序 2.5GB 的 RAM，激活索引很重要，因为重新创建索引非常繁重。

我没有设法在 read_parquet 上找到使其在加载时创建除法的参数或引擎组合。

数据文件被命名为“yyyy-mm-dd.parquet”，所以我绑定到从该信息创建部门：

这没有启用索引的使用，并且在某些情况下它失败了“TypeError：只能将元组（而不是“列表”）连接到元组”

然后我尝试将部门设置为元组 ddf.divisions = tuple(divisions)，然后它起作用了。当索引设置正确时，dask 速度非常快

更新 2

更好的方法是单独读取 dask 数据帧，然后将它们连接起来：

以这种方式设置分区，它还解决了随着时间的推移处理添加列的另一个问题。

使用以下代码pyarrow将pandas.DataFrame包含Player对象转换为pyarrow.Table

我们得到错误：

使用遇到同样的错误

是否可以使用pyarrow回退到序列化这些 Python 对象pickle？还是有更好的解决方案？pyarrow.Table最终将使用Parquet.write_table().

• 使用 Python 3.8.0、pandas 0.25.3、pyarrow 0.13.0。
• pandas.DataFrame.to_parquet()不支持多索引，因此pq.write_table(pa.Table.from_dataframe(pandas.DataFrame))首选使用的解决方案。

谢谢！

根据fastparquet的项目页面，fastparquet支持多种压缩方式

可选（压缩算法；gzip 始终可用）：

尤其zstandard是现代算法，它提供了高压缩比以及令人印象深刻的快速压缩/解压缩速度。这就是我在 fastparquet 中想要的。

但是在fastparquet.write的文档中

压缩应用于每一列，例如 GZIP 或 SNAPPY 或像 {"col1": "SNAPPY", "col2": None} 这样的字典来指定每列的压缩类型。在这两种情况下，压缩器设置都是底层压缩器默认值。要将参数传递给底层压缩器，每个 dict 条目本身应该是一个字典：

没有提到zstandard。更糟糕的是，如果我写

它弹出错误说

压缩“LZ4”不可用。选项：['GZIP'，'未压缩']

所以fastparquest只支持'GZIP'？这与项目页面有很大的差异！我错过了一些包裹吗？如何将 fastparquest 与所有项目页面声明的压缩算法一起使用？

我正在将大量（100 到 1000）的镶木地板文件读入单个 dask 数据帧（单台机器，所有本地）。我意识到

效率远低于

对于我的特定应用程序，第二种方法 ( from_delayed) 需要 6 秒才能完成，第一种方法需要 39 秒。在这种dd.read_parquet情况下，在工人开始做某事之前似乎有很多开销，并且有相当多的transfer-...操作分散在任务流图中。我想了解这里发生了什么。read_parquet这种方法慢得多的原因可能是什么？它与仅读取文件并将它们分成块有什么不同？