问题标签 [dask-dataframe]

我浏览了 Dask 教程，它们总是从客户端的初始化开始：

我最感兴趣的是使用 Dask 的 read_csv 函数在我的笔记本电脑上并行读取 DataFrame。

尽管设置了 n_workers=4，但 Dask 在读取 csv 时会使用所有内核。是否初始化Client都是一样的。当我在本地使用 Dask 并且仅用于读取文件时，我什至需要初始化客户端吗？它是用 Dask 隐式初始化的吗？

我正在尝试使用我设置的 dask 集群运行和并行化这个 sqlalchemy 查询，因为我没有足够的内存从本地计算机执行它。

我的代码如下 - 我不确定这是否是实现此目的的最佳方法：

我收到以下错误：

我正在尝试在大型数据集上计算过去 n 天（n = 30）的滚动平均值。在 Pandas 中，我会使用以下命令：

基本上，函数按“id 代码”分组，并且在“entry_time_flat”（日期时间对象）的最后 n 天，计算特征“持续时间”的平均值。

但是，为了保持代码高效，最好在 Dask 数据帧上重现此函数，而不将其转换为 Pandas DF。

如果我在 Dask DF 上运行上述代码，则会引发以下错误：

最终，在 Dask 数据帧的最后 n 天内，我如何计算“持续时间”列的平均值？

我正在尝试过滤 Dask DataFrame，然后使用map_partitions将函数应用于每个分区。DataFrame该函数需要一个至少有 1 行的 pandas 。

这是为 MCVE 生成一些虚拟数据作为pandas DataFrame（然后转换为 Dask DataFrame）的代码

数据结构：对于每个store_id，正好有 6 行。

现在我创建了一些store_id我想用来过滤上述数据的 s列表

然后我将上述数据（熊猫DataFrame）转换为dask.dataframe

现在我打印的分区ddf

这是意料之中的。每个分区有 6 行和一个 (unique) store_id。因此，每个分区都包含单个store_id.

store_id我现在使用上面的 s列表过滤 Dask 数据框

我再次打印过滤后的分区ddf

这是意料之中的，因为每个分区都有一个store_id，并且通过过滤，一些分区将被完全过滤掉，因此它们将包含零行。

Dataframe所以，现在我将根据Dask DataFrame 最佳实践重新分区过滤

我预计这种重新分区操作只会产生大小均匀的非空分区。但是，现在当我重新打印分区时，我得到了与前一个类似的输出（不均匀的分区大小和一些空分区），就好像重新分区没有发生一样

我的下一步是在过滤后对每个分区应用一个函数，但这不起作用，因为有一些分区（pandas DataFrame）由于缺少行而函数无法处理。

重新分区的 Dask 文档解释得很好（与我上面链接的最佳实践相同），它看起来很简单，但是在重新分区之后，我仍然得到一些零行的分区，并且map_partitions会在这里失败。我确定我在这里遗漏了一些东西。

有几篇关于重新分区（1、2）的 SO 帖子，但它们不处理空分区。

问题

有没有办法确保在重新分区后，所有分区将再次有 6 行并且没有空分区？即是否有可能有一个重新分区的 DaskDataFrame具有相同大小（非空）的分区？

编辑

目前，Dask 似乎无法处理空分区：问题1、2。这些可能与我在这里遇到的问题有关。

我有 1024 个镶木地板文件，每个 1mbin 大小。我正在使用 pythondask将这些 1024 个文件合并到一个文件中，并且我有很多磁盘空间，但是 ram 是有限的。

有没有使用 python 解决这个问题的有效方法dask？

这是我的代码，它会抛出内存错误。如果我在引擎盖下错了，请纠正我。该代码正在逐个文件加载并创建每个 DF，然后进行连接。在这种情况下，它不需要很多内存吗？

有没有其他方法可以解决？

如何转换这种类型的两个日期和时间列：

使用 dask map_partitions 进入单个日期时间列（例如 format='%H:%M:%S %Y-%m-%d'）？

在那里：

为 dask 创建提交函数。在这里，我想按名称访问已发布的数据集：

最后提交：

这会产生 - 但应该是0.5：

那么我怎样才能published dataset从dask工作中访问呢？

我有一个包含小镶木地板文件（600）的目录，我想在这些镶木地板上进行 ETL 并将这些镶木地板合并到每个文件 128mb。什么是处理数据的最佳方式。

我应该读取 parquet 目录中的每个文件并将 concat 作为单个数据框并执行 groupBY 吗？或者向 dd.read_parquet 提供 parquet 目录名称并进行处理？

我觉得，当我逐个文件读取文件时，它会创建一个非常大的 dask 图，无法作为图像拟合。我想它也适用于这么多线程？这会导致内存错误。

哪种方式最好读取镶木地板文件以作为 dask 数据帧处理？逐个文件或提供整个目录？？

我正在使用 dask read_parquet 读取文件列表并将这些数据帧连接起来并写入某个文件。在连接期间，dask 是否在连接时将所有数据读取到内存中，或者它只加载模式的连接（我正在与轴 0 连接）？

提前致谢

我正在尝试使用 map_partitions 在 dask 数据帧上应用一堆函数。它在本地定义函数时起作用，例如：

但是当我们将代码拆分为模块时它不起作用..

助手.py

主文件

它只是抛出 KilledWorker:("('assign-read-parquet-head-1-5-assign-77bd7b855e5e8eec82312c65361fc7c5', 0)",