问题标签 [fastparquet]

我有一个由 100,000 多行组成的数据框，每行有 100,000 列，总共有 10,000,000,000 个浮点值。

我之前已经设法在一个csv（制表符分隔的）文件中读取它们，并且我成功地将它们读取到具有 250GB RAM 的 50 核 Xeon 机器中，并尝试将其作为.parq目录写出：

浮点数huge.csv保存为字符串，大小为 125GB。

它已经写了huge.parq将近一个星期，目录是 14GB，看起来保存.to_parquet过程不会很快停止。

并且free -mh显示仍有可用内存，但保存.parq目录的时间非常慢：

问题是：

• 鉴于数据框和机器的大小，将 dask 数据框保存到镶木地板文件是否可行？

• 保存巨大的数据帧是否正常dask并且fastparquet需要这么长时间？

• 有什么方法可以估计保存镶木地板文件所需的时间吗？

我在已经安装了 python (3.6) 和 anaconda 的 EC2 服务器中安装了以下模块：

• 活泼的
• 皮箭
• s3fs
• 快速镶木地板

除了 fastparquet 其他一切都适用于导入。当我尝试导入 fastparquet 时，它会引发以下错误：

我该如何解决这个问题？

我正在寻找使用 python 从 s3 中读取多个分区目录的数据的方法。

data_folder/serial_number=1/cur_date=20-12-2012/abcdsd0324324.snappy.parquet data_folder/serial_number=2/cur_date=27-12-2012/asdsdfsd0324324.snappy.parquet

pyarrow 的 ParquetDataset 模块具有从分区读取的能力。所以我尝试了以下代码：

它抛出了以下错误：

根据 pyarrow 的文档，我尝试使用 s3fs 作为文件系统，即：

这会引发以下错误：

我仅限于使用 ECS 集群，因此spark/pyspark 不是一个选项。

有没有一种方法可以轻松地从 s3 中的此类分区目录中的 python 中轻松读取镶木地板文件？我觉得列出所有目录然后阅读并不是这个链接中建议的好习惯。我需要将读取的数据转换为 pandas 数据帧以进行进一步处理，因此更喜欢与 fastparquet 或 pyarrow 相关的选项。我也对 python 中的其他选项持开放态度。

我们在(Python) 和(R using the packet )中生成了一个parquet文件 。我们注意到了几个问题： DaskDrillSergeant

1. Dask(ie )的格式fastparquet有 a_metadata和 a_common_metadata文件，而parquetin 中的文件R \ Drill没有这些文件，parquet.crc而是有文件（可以删除）。这些实现之间有什么区别parquet？

我需要使用 dask 将具有相同架构的多个镶木地板文件加载到单个数据框中。这在它们都在同一个目录中时有效，但在它们位于不同目录中时无效。

例如：

工作得很好，但如果我复制data2.parq到不同的目录，以下内容不起作用：

我得到的回溯如下：

使用时我得到同样的错误dask.dataframe.read_parquet，我假设它使用相同的ParquetFile对象。

如何从不同目录加载多个文件？将我需要加载的所有文件放入同一个目录不是一种选择。

尝试将数据框保存为文件时出现错误。

该错误与“日期”有关，错误消息如下所示...

我已经检查过该列是“对象”，就像任何其他可以毫无问题地序列化的列一样。如果我从数据框中删除“dob”列，则此行将起作用。如果有日期+时间，这也将起作用。

快速镶木地板只接受日期？

我将新节点添加到 dask 集群并使用 conda 安装了所有依赖项。安装在所有 dask 节点上的包和版本方面都相似。

但是，新节点上的任务失败并出现以下错误：

[worker ....]：distributed.worker - 警告 - 无法反序列化任务

所有其他节点工作正常。仅供参考：所有节点都安装了快速镶木地板。

如果我遗漏了什么，请告诉我。

我有一个配置单元格式和快速压缩的镶木地板文件。它适合内存，pandas.info 提供以下数据。

parquet 文件中每组的行数仅为 100K

现在，用 dask 进行一些简单的计算，我得到以下时间

使用线程

使用分布式（本地集群）

没关系，每个大约 9 秒。

现在使用多处理，惊喜来了……

我希望多处理和分布式/本地集群处于同一数量级，可能与线程有一些差异（无论好坏）

但是，多处理需要 47 倍的时间才能在 in16 列上生成简单的平均值？

我的环境只是一个带有所需模块的全新 conda 安装。没有任何东西的精心挑选。

为什么会有这种差异？我无法管理 dask/distributed 以使其具有可预测的行为，以便能够根据问题的性质在不同的调度程序之间做出明智的选择。

这只是一个玩具示例，但我无法找到符合我期望的示例（至少我对阅读文档的理解）。

有什么我应该记住的吗？还是我完全错过了重点？

谢谢

JC

我发现在运行to_parquet函数时列对象的类型存在一些问题。我正在运行 dask 版本 15.1 。

我已经使用astype(object)显式转换了我的列，这有助于一些列但不是全部。
我收到以下错误 - TypeError: bad argument type for built-in operation。
添加object_encoding='utf-8参数后，我收到以下错误 - TypeError: expected list of bytes。

调试此问题的最佳方法是什么？
顺便说一句-我的数据集有希伯来语字符。

谢谢

我正在努力寻找如何在 Spark API 中指定 parquet 文件编写器的行组大小。

我找到了一种方法来做到这一点，即使用具有此选项的快速镶木地板 python 模块：

从 fastparquet 导入写入

write 有参数：row_group_offsets

另外， row_group size 的最佳数字是多少？

感谢快速镶木地板，我做了一些实验。例如，选择 100 万的 row_groupsize 比 10 000 快十倍。但是如果我选择超过 100 万个，它就会开始减慢我的简单查询速度。

预先感谢您的帮助