dask

6热度

1回答

如何在一台机器上利用Pandas Dataframe上的所有内核来并行化apply（）？

截至2017年8月，Pandas DataFame.apply()不幸仍然局限于使用单核，这意味着当您运行df.apply(myfunc, axis=1)时，多核机器将浪费大部分计算时间。如何使用所有内核并行运行数据帧？

2热度

1回答

我有几个文件以一列，其被称为idx，我想用它作为指标。获得的数据帧大约有13M行。我知道我可以阅读并以这种方式分配指标（这是慢〜40秒） df = dd.read_parquet("file-*.parq") df = df.set_index("idx") 或在此的其他方式（这是快速〜40毫秒） df = dd.read_parquet("file-*.parq", index = "id

0热度

2回答

嵌套dask.compute不会阻止

dask.compute（...）预计是阻止呼叫。但是，当我嵌套dask.compute，并且内部一个I/O（如dask.dataframe.read_parquet）时，内部dask.compute不会阻塞。下面是一个伪代码示例：如果我开始2名工人8个处理每个像 import dask, distributed def outer_func(name): files = find_

2热度

1回答

复杂的过滤

我已经习惯了这样做的熊猫数据帧的对象“复杂”过滤： import numpy as np import pandas as pd data = pd.DataFrame(np.random.random((10000, 2)) * 512, columns=["x", "y"]) data2 = data[np.sqrt((data.x - 200)**2 + (data.y - 200)*

1热度

2回答

如何区分排队和正在运行的期货（并杀死运行时间过长的期货）

在dask.distributed上使用期货时，有没有办法区分目前正在评估的期货pending和仍然在队列中？原因是我将大量任务（〜8000）提交给较小的工作人员（100），因此不是所有任务都可以立即处理。这些任务涉及调用第三方可执行文件（，通过subprocess.check_output），在一些极少数情况下会进入无限循环。因此，我想取消期货运行时间太长（使用任意超时）。然而，似乎没有办法

1热度

1回答

将dask集合异步存储到文件/ CSV

我正在使用dask.distributed来实现各种数据处理管道。通常从S3读取原始数据，最后处理（大）集合也将在S3上写入CSV。我可以异步运行处理并监视进度，但我注意到存储集合到文件的所有to_xxx（）方法似乎都是同步调用。其中一个缺点是电话会阻塞很长时间。其次，我不能轻易构建一个完整的图形，以便稍后执行。有没有办法运行例如to_csv（）异步获取未来的对象而不是阻塞？ PS：我非常确定

1热度

1回答

Dask worker优雅任务失败

当我运行dask.distributed workers时，任务函数中抛出的任何异常都会传播到调度程序并终止整个工作。有没有办法使任务失败，以便调度程序负责重试它（可能在另一个工作者上）？

1热度

1回答

分布式工作人员的Dask工作人员资源

在分布式多处理工作人员中定义工作人员资源（http://distributed.readthedocs.io/en/latest/resources.html）时，是否为所有进程定义了资源池？例如，工人主机上我运行： dask-worker --nprocs 8 --resources HOST=1 现在，如果我认为需要resources={"HOST":1}任务（s）没有这样的保证，只有那

1热度

1回答

为数据帧操作定义dask工作者资源

我正在将多个操作应用于dask数据框。我可以为特定操作定义分布式工作人员资源需求吗例如我打电话是这样的： df.fillna(value="").map_partitions(...).map(...) 我想指定map_partitions资源需求（）（比那些潜在的不同的地图（）），但好像方法不接受资源参数。 PS。或者，我发现我可以在map_partitions（）之后调用client.p

0热度

2回答

Dask：在Dataframe组上的nunique方法通过

我想知道是否有可能在使用Dask进行groupBy聚合之后获得来自给定列的唯一项的数量。在文档中我没有看到类似的东西。它可以在熊猫数据框中使用，非常有用。我已经看到一些与此相关的问题，但我不确定它是否已实施。有人可以给我一些提示吗？