如何查看Dask计算任务进度？
Dask是一个用于并行计算的开源框架，可以在分布式环境中执行高性能的数据分析任务。要查看Dask计算任务的进度，可以使用以下方法：
- 使用Dask的
progress函数：Dask提供了一个progress函数，可以在计算任务执行期间显示进度条。可以通过在任务执行的代码中插入progress()函数来启用进度条。例如：
import dask
from dask.distributed import Client
client = Client() # 连接到Dask集群
@dask.delayed
def my_task(x):
# 执行计算任务
...
result = my_task(10)
result.compute() # 启动计算任务
dask.distributed.progress(result) # 显示进度条- 使用Dask的
distributed模块：Dask的distributed模块提供了更多的功能来监视和管理计算任务。可以使用Client对象的progress方法来查看任务的进度。例如：
from dask.distributed import Client
client = Client() # 连接到Dask集群
@dask.delayed
def my_task(x):
# 执行计算任务
...
result = my_task(10)
result.compute() # 启动计算任务
client.progress(result) # 显示进度条- 使用Dask的Dashboard：Dask提供了一个Web界面的Dashboard，可以实时监视和管理计算任务。可以通过在代码中插入以下代码来启动Dashboard：
from dask.distributed import Client
client = Client() # 连接到Dask集群
client.get_dashboard_link() # 获取Dashboard链接然后，可以在浏览器中打开链接，查看计算任务的进度、性能指标和其他相关信息。
以上是查看Dask计算任务进度的几种方法。根据具体的需求和使用场景，可以选择适合的方法来监视和管理Dask计算任务的进度。对于更详细的信息和示例，请参考腾讯云Dask相关文档和产品介绍：
相关·内容
我试图计算Dask数据的相关矩阵。
数据非常庞大--围绕着带有900列的1 000 000行。，我想从关联矩阵创建一个热图。--我知道热图可能是完全混乱的，视觉上也不讨人喜欢，但我还是想做。
我使用Dask dataframe来获得parallelism的优势，这样就可以更快地计算它：
from dask.distributed import Client
import dask.dataframe as df
#Create cluster on local machine
client = Client(n_workers = 4, threads_per_worker = 1, mem
浏览 1提问于2021-04-04得票数 1
回答已采纳
我正在尝试在整个脚本周围包装一个进度指示器。但是，set_index(..., compute=False)仍然在调度程序上运行任务，这在web界面中可见。
如何报告set_index步骤的进度？
import dask.dataframe as dd
from dask.distributed import Client, progress
if __name__ == '__main__':
with Client() as client:
df = dd.read_csv('big.csv')
# I can see on th
浏览 2提问于2017-10-25得票数 5
tqdm和dask都是python中令人惊奇的迭代包。虽然tqdm实现了所需的进度条，但dask实现了多线程平台，它们都可以减少迭代过程的沮丧。但是-我很难把两者结合在一起。
例如，下面的代码在dask中使用tqdm.trange进度条实现延迟执行。问题是，由于delayed执行得很快，进度条立即结束，而真正的计算时间则是在compute部分期间完成的。
from dask import delayed,compute
from tqdm import trange
from time import sleep
ct = time()
result= []
def fun(x):
浏览 3提问于2017-06-11得票数 8
回答已采纳
环境信息： Python==3.7
Win7 i5-2310:2-cores-4-threads 4GB-memory Dask Ini： from dask.distributed import Client
Client()
# Client Scheduler: tcp://127.0.0.1:50723 Dashboard:
# http://127.0.0.1:8787/status Cluster Workers: 4 Cores: 4 Memory: 4.20 GB 第一个:带有For循环的延迟函数： @dask.delayed
def teststr(x,y):
浏览 25提问于2020-01-27得票数 0
回答已采纳
我非常熟悉为CPU分发的Dask。我想探索如何在GPU内核上运行我的代码。当我向LocalCUDACluster提交任务时，我会得到以下错误：
ValueError: tuple is not allowed for map key
这是我的测试用例：
import cupy as cp
import numpy as np
from dask_cuda import LocalCUDACluster
from dask.distributed import Client
cluster = LocalCUDACluster()
c = Client(cluster)
def test_f
浏览 5提问于2020-06-03得票数 0
回答已采纳
1回答
如何监控逐行Dask DataFrame apply操作的进度？
使用ProgressBar()包装该行似乎没有任何作用，即控制台上没有打印任何内容？
from dask.diagnostics import ProgressBar
with ProgressBar():
df_calc = ddf.apply(myfunc, axis=1)
浏览 0提问于2018-05-14得票数 0
1回答
报告Dask任务的进度
、、、、
我正在Dask调度程序上运行一些缓慢的任务，我需要每个任务的进度报告。这些任务将从处理进度报告的同一台机器上提交，因此可以保持在同一个过程中，但现在让我们假设任务是在不同的过程中提交的，进度报告是在不同的过程中处理的。
Dask提供了，它的目标用例包括能够监视进度：
这些工具可用于控制对外部资源的访问，跟踪正在进行的计算的进度，或在许多工作人员、客户端和任务之间通过侧通道共享数据。
我能想到的最简单的例子是：
任务提交者：
from dask.distributed import Client, Pub
import time
c = Client('tcp://127.0.
浏览 2提问于2019-12-24得票数 1
1回答
我在达克上读到：
Worker节点和Scheduler节点同时运行。它们为多个重叠请求提供服务，同时在不阻塞的情况下执行多个重叠计算。
我一直认为单线程并发编程最适合于昂贵的I/O，而不是CPU绑定的作业。然而，我希望许多dask任务(例如dask.pandas、dask.array)都是CPU密集型的。
分布式系统是否只对客户端/服务器通信使用旋风，并有单独的进程/线程来运行dask任务？实际上，dask-worker有--nprocs和--nthreads参数，所以我希望是这样的。
如何与“旋风协同”和处理每个dask任务的更常见的进程/线程并发在分布式中？
浏览 3提问于2016-10-04得票数 4
回答已采纳
我没有对实验室环境的浏览器访问权限，并且到目前为止，可用于实验室的dask扩展对我来说还不起作用。我希望能够看到我的dask项目的进度和性能数据，但目前还没有运气。 compute()有时需要几个小时(没问题，有很多数据)，整个过程让我感觉很盲目。 您知道实验1.2是否支持任何功能吗？progressbar也不走运。该实例位于AWS SageMaker上，因此我无法访问web仪表板。 谢谢!
浏览 15提问于2021-04-01得票数 2
回答已采纳
我正在使用大型CSV (~60 in；~250M行)和Dask in Jupyter。 加载DF后，我要做的第一件事是连接两个字符串列。我可以成功地做到这一点，但是我注意到单元执行时间似乎并没有随着工作进程数量的增加而减少(我在一台具有64个逻辑核心的机器上尝试了5、10和20 )。如果说有什么不同的话，那就是每五个左右的工人似乎就会增加一分钟的执行时间。 同时，Dask仪表板的进度条表明，任务与工作人员数量有很好的比例关系。在5个工作进程中，任务完成(ac。到仪表板)大约需要10-15分钟。在20个工人中，流可视化建议在大约3-5分钟内完成任务。但单元执行时间保持在25分钟左右，即在5-w
浏览 21提问于2021-07-23得票数 1
回答已采纳
我在一台机器上使用Dask (LocalCluster有4个进程，16个线程，68.56 to内存)，当我试图一次计算两个共享依赖关系的结果时，遇到了工作内存问题。
在下面显示的示例中，仅用一次计算就可以很好地快速计算result，工作人员的总内存使用量最高可达1 1GB左右。但是，当使用两次计算来计算results时，当总内存使用量约为40 is时，工作人员会迅速耗尽所有内存，并开始写入磁盘。计算最终将完成，但一旦开始写入磁盘，就会有一个巨大的减速。
直观地说，如果读入一个块，然后立即计算它的两个和，那么该块可以被丢弃，并且内存使用率保持在较低水平。然而，看起来Dask正在优先加载数据，而不
浏览 18提问于2021-02-27得票数 2
回答已采纳
假设一个dask集群有一些CPU设备和一些GPU设备。每个设备运行一个单独的dask-worker。现在，问题是我如何发现dask-worker的底层设备是CPU或GPU。
例如：-如果dask-worker在CPU设备上运行，我应该知道它在CPU上运行，或者如果dask-worker在GPU设备上运行，我应该以编程方式知道设备类型。有没有什么方法可以通过编程的方式知道这一点？
浏览 0提问于2018-04-19得票数 0
我使用的是dask_jobqueue.SGECluster()，当我向网格提交作业时，它们都被列为dask-worker。我希望每个提交的工作都有不同的名字。
以下是一个例子：
futures = []
for i in range(1,10):
res = client.submit(slow_pow, i,2)
futures.append(res)
[future.result() for future in as_completed(futures)]
当使用dask-worker检查它们的状态时，所有10个作业都以名称qsub出现。
我尝试过在循环中添加cl
浏览 4提问于2022-01-13得票数 0
有什么方法可以获取Dask诊断数据，而不是Dask.distributed client的仪表板？
Dask已经提供了一个不错的Bokeh dashboard，它绘制了相当多的诊断信息。然而，我想要的不是情节，而是它们的价值。与时间戳、进度值、cpu和内存使用类似。我希望将这些值存储在数据库中，以供我自己进行监视。
到目前为止，我已经尝试使用函数Dask.distributed.get_task_stream()，它提供了列表中的工作人员的信息，但是我想以流的方式了解Task 图到底在仪表板上显示了什么。
注意:有一个名为dask.diagnostics的包，您可以在那里导入一个Progres
浏览 2提问于2019-07-03得票数 1
最近我正在学习达斯克，听说它比熊猫记忆更高效，速度更快。
这可能是内存效率高，但一个简单的基准测试给我的达斯克比熊猫的速度。
Qn:如何提高DASK的速度?
下面是设置：
import numpy as np
import pandas as pd
import dask
import dask.dataframe as dd
import dask.array as da
import time
import seaborn as sns
df = sns.load_dataset('titanic')
df = df = pd.concat([df for _ in
浏览 0提问于2019-02-17得票数 0
回答已采纳
我的Dask计算很慢。当我查看诊断仪表板的状态页面时，我看到大部分时间都花在disk-read-*和disk-write-*任务上。
这是什么意思？
如何诊断此问题？
浏览 0提问于2018-02-22得票数 3
我有一个程序，其中每个任务都是通过subprocess.Popen调用一个subprocess.Popen外部程序。任务以图形形式排列，一切都通过dask get命令执行。
我有这个程序的一个节点版本，它可以很好地与dask.threaded一起工作，并且我正在尝试将这个版本扩展到一个分布式设置。我的目标是在Slurm集群上运行它，但我在部署工作人员时遇到了困难。当我运行以下命令时：
screen -d -m dask-scheduler --scheduler-file scheduler.json
screen -d -m srun dask-worker --scheduler-file
浏览 0提问于2018-05-18得票数 1
回答已采纳
我是Dask的新手。我把它和一个xarray数据集一起使用。我将数据集保存在内存中，jupyter单元显示它已经就绪(不再有星号)。但是dask仪表板正忙着执行任务。我不明白。当这种情况发生时，我应该等到dask仪表板稳定下来，还是可以自由运行下一个单元？
浏览 6提问于2022-02-07得票数 0
下面的代码将一个函数映射到一个迭代器上。应用于每个元素的函数运行一个docker容器，以计算其返回值：
import subprocess
def task(arg):
return subprocess.check_output(
["docker", "run", "ubuntu", "bash", "-c", f"echo 'result_{arg}'"]
)
args = [1, 2, 3]
for result in map(task,
浏览 18提问于2020-03-05得票数 1
回答已采纳
我正在使用分布式调度程序和分布式进度条。是否有办法让进度条工作为Dataframe.map_partition或延迟？我认为缺乏期货是导致酒吧不起作用的原因。如果我将代码更改为client.submit，进度条就会起作用。
代码如下：
import dask.dataframe as dd
from distributed import Client
from distributed.diagnostics.progressbar import progress
client = Client("tcp://....")
...
ddf = dd.read_parquet(&
浏览 3提问于2020-12-01得票数 2
回答已采纳
我有这样一个小数据集：
<xarray.Dataset>
Dimensions: (time: 24)
Coordinates:
* time (time) datetime64[ns] 2022-09-28 ... 2022-09-28T23:00:00
spatial_ref int64 0
Data variables:
CO (time) float32 dask.array<chunksize=(24,), meta=np.ndarray>
NO2 (time) fl
浏览 10提问于2022-10-13得票数 0
一个简单的代码片段如下:后面跟着###的注释很重要。
from dask.distributed import Client
### this code-piece will get executed on a dask worker.
def task_to_perform():
print("task in progress.")
## do something here..
print("task is over.!")
### whereas the below code will run on client side,
#
浏览 4提问于2018-02-19得票数 2
回答已采纳
2回答
dask计算结果存储吗？
、、、、
考虑以下代码
import dask
import dask.dataframe as dd
import pandas as pd
data_dict = {'data1':[1,2,3,4,5,6,7,8,9,10]}
df_pd = pd.DataFrame(data_dict)
df_dask = dd.from_pandas(df_pd,npartitions=2)
df_dask['data1x2'] = df_dask['data1'].apply(lambda x:2*x,meta=('data1x2&#
浏览 8提问于2022-04-03得票数 2
回答已采纳
1回答
1)如何取消数据帧申请？e.g
dask_df.apply(myfunc) # Long running
2)一般情况下，如何停止所有正在运行的任务？
这就是client.cancel(futures)。但我没有关于未来的参考。dask会存储它们的列表吗？
浏览 0提问于2020-02-26得票数 0
1回答
我是Dask库的新手，我想知道如果我们在两个系统上使用dask实现并行计算，那么我们应用计算的数据框架是否存储在这两个系统上？并行计算究竟是如何进行的，文档中并不清楚。
浏览 0提问于2018-07-03得票数 0
1回答
如果我尝试使用dask并行for循环，那么它的执行速度将比常规版本慢。基本上，我只是遵循dask教程中的介绍性示例，但由于某种原因，它在我的头上失败了。我做错了什么？
In [1]: import numpy as np
...: from dask import delayed, compute
...: import dask.multiprocessing
In [2]: a10e4 = np.random.rand(10000, 11).astype(np.float16)
...: b10e4 = np.random.rand(10000, 11).astype(
浏览 1提问于2018-02-12得票数 5
回答已采纳
1回答
我对达斯克比较陌生。我正在尝试并行化一个不使用Dask容器的“自定义”函数。我只想加快计算速度。但是我的结果是，当我尝试使用dask.delayed并行化时，它的性能要比运行串行版本差得多。下面是一个演示问题的最小实现(我实际上想要这样做的代码涉及得更多:)
import dask,time
def mysum(rng):
# CPU intensive
z = 0
for i in rng:
z += i
return z
# serial
b = time.time(); zz = mysum(range(1, 1_000_000_000
浏览 1提问于2022-09-25得票数 2
2回答
当我使用Dask运行一个计算任务时，我希望在Jupyter笔记本上看到一个进度条，我正在从一个大的csv文件+4 4GB中计算id列的所有值，所以你有什么想法吗？
import dask.dataframe as dd
df = dd.read_csv('data/train.csv')
df.id.count().compute()
浏览 13提问于2018-03-01得票数 24
回答已采纳
1回答
我正在进行一个项目，它使用历史数据和输入的数据进行分析。我想学习如何管理在dask上更新传入的数据，同时不必每次都发送所有的历史数据。
我收集数据用于时间序列进行分析，但是时间序列随着传入数据的增长而增长，每个流的传入数据都需要发送给适当的工作人员，以便进行ARMA分析。如果我对天气进行ARMA分析，我希望将气压与温度分开，并通过比较压力与温度和温度与温度来进行分析。我不想将新的温度数据附加到以前的温度数据中，并将现在更大的系列分派给新的工作人员。我只想将新的温度数据发送给已经拥有所有先前温度数据的dask工人，以此类推。如何确保先前的温度数据在员工上持续存在，以及如何(仅)将新的温度数据分配
浏览 4提问于2022-11-08得票数 1
回答已采纳
1回答
Dask分布式库说：
默认情况下，distributed假定所有函数都是纯的。..。调度程序避免了冗余计算。如果结果已经从以前的调用中内存中，那么这个旧结果将被使用，而不是重新计算它。
当基准测试函数运行时，这种缓存行为将成为障碍，因为我们希望用相同的输入多次调用相同的函数。那么有什么方法可以完全禁用它吗？
我知道，对于submit和map，有一个可以使用的参数。但是对于dask集合的计算，我还没有找到任何解决方案。
浏览 6提问于2022-11-01得票数 0
回答已采纳
2回答
dask中的点产品似乎比在numpy中运行得慢得多：
import numpy as np
x_np = np.random.normal(10, 0.1, size=(1000,100))
y_np = x_np.transpose()
%timeit x_np.dot(y_np)
# 100 loops, best of 3: 7.17 ms per loop
import dask.array as da
x_dask = da.random.normal(10, 0.1, size=(1000,100), chunks=(5,5))
y_dask = x_dask.transpose
浏览 7提问于2015-12-23得票数 4
回答已采纳
1回答
任务期间的任务进度
、、、、
使用dask dataframe
df = dask.dataframe.from_pandas(df, npartitions=5) series = df.apply(func) future = client.compute(series) progress(future)
在jupyter笔记本中，我可以看到进度条显示每个分区完成了多少次apply()调用(例如2/5)。
有没有办法让dask报告每个分区的进度？
就像熊猫的tqdm progress_apply()。
浏览 2提问于2020-03-01得票数 0
我正在设置一个Dask脚本，以便在PSC Bridge P100图形处理器节点上执行。这些节点提供2个CPU和32个CPU核心。我想启动基于CPU和GPU的dask-workers。
CPU工作进程将启动：
dask-worker --nprocs 1 --nthreads 1
而GPU工作者则表现为：
CUDA_VISIBLE_DEVICE=0 dask-worker --nprocs 1 --nthreads 1
我的工作流由一组仅CPU任务和一组GPU任务组成，它们取决于CPU任务的结果。是否有一种方法可以将GPU任务仅绑定到GPU工作者？
此外，我希望确保GPU任务与它们所依赖的CPU
浏览 0提问于2019-08-20得票数 3
1回答
我有以下基本代码，这些代码(我认为)应该设置xarray来使用LocalCluster。
from dask.distributed import Client
client = Client("tcp://127.0.0.1:46573") # this points to a LocalCluster
import xarray as xr
ds = xr.open_mfdataset('*.nc', combine='by_coords') # Uses dask to defer actually loading data
我现在开始一
浏览 6提问于2020-06-08得票数 0
回答已采纳
2回答
dask.compute(...)应该是一个阻塞调用。然而，当我嵌套了dask.compute，并且内部的dask.compute执行I/O (如dask.dataframe.read_parquet)时，内部的and不是阻塞的。下面是一个伪代码示例：
import dask, distributed
def outer_func(name):
files = find_files_for_name(name)
df = inner_func(files).compute()
# do work with df
return result
def inner
浏览 11提问于2017-08-10得票数 0
我知道有一种方法可以在全球注册dask.diagnostics.ProgressBar，虽然它很不错，但它破坏了我的单元格输出。我还看到了一个很好的distributed.diagnostics.progress函数，它可以显示木星笔记本进度栏小部件的执行进度，但它期望收到未来的结果。
我在这里遇到的问题是，dask.diagnostics.ProgressBar使用stdout (所以我在使用进度栏时不能打印任何东西)，需要使用Dask/Distributed期货显式地调用distributed.diagnostics.progress，但是我有自己的函数，它们计算一些东西并返回一个“结果”
浏览 4提问于2016-08-15得票数 2
在将一些代码从Pandas迁移到Dask时，我发现通过使用多列调用DataFrame.assign()来修改Dask数据帧与使用多个DataFrame.__setitem__() (也称为dataframe[x]=y)调用来修改它之间存在巨大的性能差异。
使用导入
import pandas, dask, cProfile
对于定义如下的Dask数据帧：
dd = dask.dataframe.from_pandas(pandas.DataFrame({'a':[1]}), npartitions=1)
cProfile.run('for i in range(100)
浏览 3提问于2020-04-16得票数 0
为了更好地理解并行，我正在比较一组不同的代码。
这是基本的一个(code_piece_1)。
用于循环
import time
# setup
problem_size = 1e7
items = range(9)
# serial
def counter(num=0):
junk = 0
for i in range(int(problem_size)):
junk += 1
junk -= 1
return num
def sum_list(args):
print("sum_list fn:", arg
浏览 1提问于2019-09-06得票数 3
回答已采纳
我有拼花文件，也就是800 K行x8.7K列。我把它装进了一个dask数据仓库：
import dask.dataframe as dd
dask_train_df = dd.read_parquet('train.parquet')
dask_train_df.info()
这产生了：
<class 'dask.dataframe.core.DataFrame'>
Columns: 8712 entries, 0 to 8711
dtypes: int8(8712)
当我尝试执行像dask_train_df.head()或dask_train_df
浏览 0提问于2018-12-24得票数 6
回答已采纳
我将尝试重新表述我的问题：
如何将dask.dataframe与zip这样的函数结合起来？
假设我们有一个名为"accounts.0.csv“的文件，其中包含以下数据
id,names,amount
352,Dan,4837
387,Tim,208
42,Jerry,21
129,Patricia,284
我写了这段代码
import dask.dataframe as dd
import itertools
from dask.threaded import get
df = dd.read_csv('accounts.0.csv')
dsk = {'a
浏览 5提问于2015-10-21得票数 2
回答已采纳
我有一个最初使用concurrent.futures.ProcessPoolExecutor运行的计算工作量，我将其转换为使用dask，这样我就可以利用dask与分布式计算系统的集成来扩展到一台机器之外。工作负载由两种任务类型组成：
Task :接受字符串/浮点数输入并生成一个矩阵(大约为2000x2000)。任务持续时间通常是60秒或更短。任务B:从任务A获取矩阵，并使用它和其他一些小输入来求解一个常微分方程。解决方案被写入磁盘(因此没有返回值)。任务持续时间最长可达15分钟。
每个A任务可以有多个B任务。
最初，我的代码如下所示：
a_results = client.map(calc_a
浏览 2提问于2021-07-19得票数 1
回答已采纳
我有一个定制的工作流，它需要使用resample来获得更高的时间频率，应用ufunc和groupby + mean来计算最终结果。
我想将它应用到一个大型xarray数据集中，该数据集由一个块的dask数组支持。对于计算，我想使用dask.distributed。
但是，当我将此应用于完整的数据集时，任务数量急剧增加，使客户端不堪重负，如果提交的话，调度程序和工作人员也很可能会受到影响。
解释道：
在管道的早期执行空间和时间索引(例如.sel()或.isel())，特别是在调用resample()或groupby()之前。分组和重采样会在所有块上触发一些计算，理论上应该与索引交换，但是这种优
浏览 1提问于2020-12-16得票数 3
我很难理解何时什么时候不要在Dask数据仓库中使用compute()。我通常通过添加/删除compute()来编写代码，直到代码生效，但这是非常容易出错的。我应该如何在达斯克中使用compute()？它在Dast分发中有区别吗？
浏览 2提问于2021-05-23得票数 1
回答已采纳
1回答
使用dask，我定义了一个很长的计算管道；在给定apis和版本中的某些约束的情况下，我需要计算一些小的结果(而不是惰性)，并将其提供给惰性操作。我的问题是，在这一点上，整个计算图将被执行，以便我可以产生中间结果。在接下来的步骤中，当我将最终结果存储到磁盘时，有没有办法不丢失在这一点上完成的工作，并且必须从头开始重新计算所有内容？
使用persist会对此有所帮助吗？
任何帮助都将不胜感激。
浏览 0提问于2018-11-20得票数 1
1回答
我想对dask数据的单个列进行频率计数。代码可以工作，但是我收到一个warning抱怨meta没有定义。如果我试图定义meta，就会得到一个错误AttributeError: 'DataFrame' object has no attribute 'name'。对于这个特殊的用例，我似乎不需要定义meta，但是我想知道如何做这件事，以供将来参考。
虚拟数据和列频率
import pandas as pd
from dask import dataframe as dd
df = pd.DataFrame([['Sam', 'Alex&#
浏览 7提问于2017-06-08得票数 25
回答已采纳
像许多人一样，我在使用Jupyter笔记本的集群上使用Dask。这意味着SSH为Jupyter开辟了一个端口。
隧道一个端口是可行的，更多的端口会变得有问题。我想知道是否有计划为Dask做一个Jupyter扩展，这样你就可以有一个标签页，上面有所有漂亮的Bokeh图(或者我可能遗漏了一些基本的东西)
或者，(或者另外)我知道你可以得到一些可视化的东西，比如进度(未来)有没有办法把更多的小部件放在笔记本里？
具体来说，我是#dask提供的分析工具的铁杆粉丝
浏览 2提问于2018-08-29得票数 2
1回答
我所处的HPC环境中有集群、紧密耦合的互连和支持光泽的文件系统。我们一直在探索如何利用Dask不仅提供计算，而且充当一个分布式缓存，以加快我们的工作流。我们的专有数据格式是n维和规则的，我们编写了一个懒散的读取器，以传递给from_array/from_delayed方法。
我们在Dask集群中加载和持久化比内存更大的数据集时遇到了一些问题。
使用hdf5的示例：
# Dask scheduler has been started and connected to 8 workers
# spread out on 8 machines, each with --memory-limit=15
浏览 0提问于2018-06-11得票数 2
回答已采纳
1回答
我可以在pandas中使用tqdm进度条，例如： tqdm.pandas()
df = df['var'].progress_apply(lambda x: something(x)) 我可以在thing cudf或dask_cudf中做同样的事情吗?如果不能，那么我如何在其中使用tqdm进度条，
浏览 58提问于2021-07-31得票数 0
回答已采纳
1回答
与类似，我遇到了Dask分发版的内存问题。但是，在我的例子中，解释并不是客户端试图收集大量数据。
这个问题可以根据一个非常简单的任务图来说明:一个delayed操作列表生成一些固定大小为500 MB的随机DataFrames (以模拟从文件加载许多分区)。任务图中的下一个操作是获取每个DataFrame的大小。最后，将所有大小缩小为一个总大小，即必须返回给客户端的数据很小。
出于测试目的，我正在运行本地调度程序/工作线程单线程，限制为2GB内存，即：
$ dask-scheduler
$ dask-worker localhost:8786 --nthreads 1 --memory-limi
浏览 4提问于2017-06-03得票数 13
回答已采纳
从一个库中，我得到一个函数，它读取一个文件并返回一个numpy数组。
我想用多个文件中的多个块构建一个Dask数组。
每个块都是在文件上调用函数的结果。
当我要求Dask计算时，Dask会要求函数同时从硬盘读取多个文件吗？
如果是这样的话，如何避免呢？我的电脑没有并行文件系统。
示例：
import numpy as np
import dask.array as da
import dask
# Make test data
n = 2
m = 3
x = np.arange(n * m, dtype=np.int).reshape(n, m)
np.save('0.npy'
浏览 0提问于2018-08-06得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
