延迟函数的dask计算字典
延迟函数是一种概念，它指的是在某些计算任务中，将计算推迟到必要的时候再进行，以提高计算效率和资源利用率的一种技术。Dask是一个开源的并行计算框架，可以用于处理大规模数据集和执行并行计算任务。在Dask中，延迟函数允许用户定义计算图，而不是立即执行计算，从而可以有效地管理和优化计算过程。
延迟函数在Dask中被称为"延迟对象"，它是一个表示待执行计算的特殊对象。当需要获取计算结果时，Dask会自动构建计算图，并通过任务调度器进行任务的执行。延迟函数的使用可以提高计算的效率和可扩展性，特别适用于处理大规模的数据集和复杂的计算任务。
Dask的延迟函数主要有以下特点和优势：
- 惰性计算：延迟函数在定义时并不会执行计算，只有在需要获取结果时才会触发实际的计算过程。这种惰性计算的特性可以有效地减少不必要的计算开销。
- 并行计算：Dask框架能够自动将计算任务分割成多个子任务，并使用多线程、多进程或分布式计算资源进行并行计算。延迟函数可以充分利用这种并行计算的能力，提高计算速度和吞吐量。
- 可扩展性：延迟函数可以处理大规模的数据集和复杂的计算任务，而不受内存限制。Dask通过适应性调度和内存管理来解决大规模计算的问题，可以在单机或分布式环境下进行扩展。
- 可视化调试：Dask提供了可视化工具，可以帮助用户理解和调试计算过程。延迟函数的计算图可以可视化展示，帮助用户检查和优化计算流程。
延迟函数的应用场景广泛，适用于各种需要处理大规模数据和复杂计算任务的场景。例如：
- 大规模数据处理：延迟函数可以有效地处理大规模的数据集，如数据清洗、数据分析、机器学习等任务。
- 图计算：延迟函数可以用于处理图结构数据的计算任务，如社交网络分析、网络图算法等。
- 高性能计算：延迟函数可以利用并行计算资源，加速科学计算、数值模拟、仿真等高性能计算任务。
- 数据流处理：延迟函数可以处理实时数据流，进行流式计算、数据流分析等任务。
- 任务调度优化：延迟函数可以用于优化任务调度和资源利用，提高计算效率和系统性能。
对于Dask框架中的延迟函数，腾讯云提供了一系列相关产品和服务，例如：
- 腾讯云弹性MapReduce（EMR）：提供大数据处理和分析的托管式服务，支持使用Dask进行并行计算和延迟函数的应用。详情请参考：腾讯云EMR产品介绍
- 腾讯云云服务器（CVM）：提供虚拟机实例，可用于搭建Dask集群和扩展计算资源。详情请参考：腾讯云云服务器产品介绍
- 腾讯云对象存储（COS）：提供可扩展的云存储服务，适用于存储大规模数据集和处理延迟函数的输入输出数据。详情请参考：腾讯云对象存储产品介绍
- 腾讯云容器服务（TKE）：提供容器编排和管理的服务，可用于部署和运行Dask集群和延迟函数应用。详情请参考：腾讯云容器服务产品介绍
总结：延迟函数是一种通过将计算推迟到必要时执行的技术，在Dask框架中可以应用于处理大规模数据和并行计算任务。它的优势包括惰性计算、并行计算、可扩展性和可视化调试。腾讯云提供了相关产品和服务，可用于支持和扩展延迟函数的应用。
相关·内容
1回答
我有一份delayed的字典清单。每个delayed对象的计算值必须转换为dask.Dataframe中的一个条目。 dfs = []
for source_list in list_of_list:
values1 = {}
values2 = {}
for source in source_list:
intermediate = dask.delayed(myfunc)(source)
source_name = string_manipulation(source)
values1[source_name]
浏览 26提问于2020-07-28得票数 0
我正在使用dask延迟函数，在函数上使用@dask.delayed装饰器时，我熟悉dask的do's和don。我意识到，有时我需要调用compute()两次才能得到结果，尽管我认为自己遵循了最佳实践。也就是说，不要在另一个dask延迟函数中调用dask延迟函数。
我在两个场景中遇到了这个问题:当有嵌套函数时，以及在类中调用成员函数时，类成员是延迟对象。
@dask.delayed
def add(a, b):
return a + b
def inc(a):
return add(a, 1)
@dask.delayed
def foo(x):
return
浏览 2提问于2019-07-09得票数 1
回答已采纳
我正在尝试使用dask并行构建一个字典，但我遇到了一个TypeError: Delayed objects of unspecified length are not iterable。
我尝试同时计算add、subtract和multiply，以便更快地构建字典。
下面是一些代表我的用例的代码：
import dask
from dask.delayed import delayed
x1 = {'a': 1, 'b': 2, 'c': 3}
x2 = {'a': 4, 'b': 5, 'c':
浏览 23提问于2018-02-11得票数 7
我无法使用分布式集群处理这个块。
import pandas as pd
from dask import dataframe as dd
import dask
df = pd.DataFrame({'reid_encod': [[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10]]})
dask_df = dd.from_pandas(df, npar
浏览 4提问于2019-11-30得票数 1
回答已采纳
我有一个文件夹，里面有497个熊猫的数据文件，存储在.parquet文件中。文件夹总维度为7.6GB。
我在尝试开发一个简单的交易系统。所以我创建了两个不同的类，其中一个是Portfolio类，然后这个类为数据文件夹中的每个数据文件创建一个Asset对象。
import os
import pandas as pd
from dask.delayed import delayed
class Asset(file):
def __init__:
self.data_path = 'path\\to\\data\\folder\\'
s
浏览 4提问于2017-11-22得票数 3
我真的很喜欢使用Dask。
有没有一种方法可以强制延迟对象在应用延迟函数之前要求计算它的所有参数？
简单的示例(使用集合的用例更有趣)：
def inc(x, y):
return x + y
dinc = dask.delayed(inc, pure=True)
类似这样的东西
def inc(x, y):
if hasattr(x, compute):
x = x.compute()
if hasattr(y, compute):
y = y.compute()
return x + y
dinc = dask.delay
浏览 0提问于2017-01-30得票数 1
1回答
我正在尝试将带有分类列的Dask数据存储到*.h5文件中，每个 - 1:23:25 - 1:23:45。
下面是我对store函数的调用：
stored = store(ddf,'/home/HdPC/Analyzed.h5', ['Tag'])
函数store是：
@delayed
def store(ddf,fp,c):
ddf.categorize(columns=c).to_hdf(fp, '/data2')
并使用。
ddf和stored类型如下：
print(type(ddf), type(stored))
>>&
浏览 4提问于2017-05-20得票数 1
回答已采纳
我想我遗漏了一些东西(仍然是Dask Noob)，但我正在尝试分批处理的建议，以避免在这里执行太多的Dask任务：
也不能让他们发挥作用。这就是我试过的：
import dask
def f(x):
return x*x
def batch(seq):
sub_results = []
for x in seq:
sub_results.append(f(x))
return sub_results
batches = []
for i in range(0, 1000000000, 1000000):
result_batch = d
浏览 0提问于2019-06-05得票数 0
1回答
我使用dask来延迟一些函数的计算，这些函数在我的代码库中返回序列。到目前为止，除了我使用np.average之外，大多数操作似乎都像预期的那样运行。
我所拥有的函数返回一个pd.Series，然后我要计算一个加权平均值。
下面是一个非达克和达斯克版本：
import dask
import numpy as np
import pandas as pd
s = pd.Series([1,2,3])
a = np.average(s, weights=s)
print(a)
ds = dask.delayed(lambda: s)()
a = np.average(ds, weights=
浏览 0提问于2018-10-04得票数 4
回答已采纳
在下面的dask代码中，我在执行两个map_partitions之前将x设置为1和2。结果似乎很好，但我并不完全理解。
如果dask只有在找到map_partitions时才等待运行这两个compute()，并且在找到compute() x时是2，那么dask如何知道x=1在第一个map_partitions中
pdf = pd.DataFrame({
'id': [1, 1, 1, 2, 2, 3, 4, 1, 2, 2, 1],
'balance': [150, 140, 130, 280, 260, 220, 230, 330, 420,
浏览 9提问于2022-06-02得票数 1
回答已采纳
1回答
我正在尝试使用dask的多线程模块。这段代码
def foo(arg):
return arg*2
jobs = []
t = delayed(foo)(100)
jobs.append(t)
j = delayed(jobs, pure=True)
#j = j.compute()
j = j.compute(get=dask.multiprocessing.get)
print("class:", type(j[0]))
执行时打印('class:', <type 'tuple'>)，但是如果我运行注释行(即不使用mult
浏览 0提问于2016-05-31得票数 3
我有一个函数crop_images_circle(file_dir,kmeans_dir,folders_dir,filename)。不会归还任何东西。试图使用达斯克来并行计算。
对大约100个奇数文件不使用dask实现：
for filename in os.listdir(file_dir):
crop_images_circle(file_dir,kmeans_dir,folders_dir,filename)
执行时间(秒)：53.58223843574524
与dask一起执行：
# Version 1
for filename in os.listdir(file_dir)
浏览 3提问于2022-06-17得票数 0
1回答
双数据帧Dask折叠
、、、
这是一个教科书的问题，如何添加两个DataFrames使用Dask (特别是与折叠).虽然我似乎无法让它起作用，所以我想伸出手来看看我做错了什么。
(我正在使用Python3.8.5和Dask 2021.4.1)
下面的代码显示了我的意图：
from dask import delayed, bag
import pandas as pd
def get_df1():
return pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
def get_df2():
return pd.DataFrame
浏览 2提问于2021-05-12得票数 1
回答已采纳
1回答
我想把我的df分成两个df。 import dask
import dask.dataframe as dd
df = dd.read_csv(r'D:\Amr\Amr.csv',error_bad_lines=False, engine="python")
import numpy as np
dfs = dask.delayed(np.split)(df,2)
df0=dfs[0]
dask.delayed(df0.to_csv)('file1.csv', header=False, index=False) 结果显示，Delayed(&#
浏览 15提问于2021-03-23得票数 0
回答已采纳
1回答
我正在创建一个函数，该函数读取整个文件夹，创建一个Dask数据帧，然后处理该数据帧的分区并对结果求和，如下所示： import dask.dataframe as dd
from dask import delayed, compute
def partitions_func(folder):
df = dd.read_csv(f'{folder}/*.csv')
partial_results = []
for partition in df.partitions:
partial = another_function(partit
浏览 6提问于2021-02-04得票数 1
回答已采纳
参考文献：
有几件事要注意：
我有48 my的ram这里是我使用的库的版本
- Python 3.7.7
- dask==2.18.0
- fbprophet==0.6
- pandas==1.0.3
我进口熊猫的原因仅仅是为了这条线。
pd.options.mode.chained_assignment = None
这有助于在im使用dask.distributed时执行dask错误。
所以，我有一个21 am的csv文件，我正在使用dask和jupyter笔记本.我试着从mysql数据库表中读取它，但是内核最终崩溃了
我尝试过多种组合使用本地网络的工作人员、线程和可用内存、可用st
浏览 2提问于2020-06-08得票数 0
我希望从使用Dask延迟的简单示例中可以看到，我基本上可以从scikit复制gridsearchcv --使用以下几个函数调用来学习。该模型似乎永远不适合(model.fit(.))因为循环的其余部分还在继续(pred(.))？
我如何嵌套函数有问题吗？我知道有一个用于dask的gridsearchcv，但问题是我的真实模型是一个多输入的Keras，而且您不能以'X‘的形式传递一个3d数组。在没有Dask的情况下，代码的串行工作很好。
下面是一个可复制的小例子：
import dask
import pandas as pd
import numpy as np
from sklear
浏览 1提问于2018-07-07得票数 0
2回答
我一直在四处寻找，但还没有找到解决方案。我一直在Dask字典中工作，但团队正在研究delayed。我需要将我的dsk{}转换为最后一步延迟的对象。
我现在要做的是：
def add(x, y):
return x+y
dsk = {
'step1' : (add, 1, 2),
'step2' : (add, 'step1', 3),
'final' : (add, 'step2', 'step1'),
}
dask.visualize(dsk)
cl
浏览 10提问于2020-06-03得票数 1
2回答
我不知道如何计算来自df.groupy.apply()操作的延迟对象。如果有人能帮忙我真的很感激。下面是我编写的示例代码
import pandas as pd
import dask
df = pd.DataFrame(columns=['id','id2','val1'])
df['id'] = ['A','A','A','B','C','C','D','D']
df['id2']=[
浏览 3提问于2021-12-22得票数 1
回答已采纳
从一个库中，我得到一个函数，它读取一个文件并返回一个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
问题综述
短版
我怎样才能从一袋潘达斯DataFrames变成一只达克DataFrame？
长版
我有许多文件无法被dask.dataframe的各种read函数(例如dd.read_csv或dd.read_parquet)读取。我有自己的函数，可以将它们读入Pandas DataFrames (函数一次只对一个文件工作，类似于pd.read_csv)。我想让所有这些单一的熊猫DataFrames在一个大达克DataFrame。
最小工作示例
下面是一些CSV数据示例(我的数据实际上并不在CSV中，但在这里使用它是为了方便示例)。要创建一个最低限度的工作示例，可以将其保存为CSV并复制几个副本
浏览 10提问于2019-12-13得票数 3
回答已采纳
1回答
dask延迟对象的计算列表
、、、、
我已经做了所有类似的问题和解决方案，但没有得到期望的输出。
我有一张达克延迟对象的清单。
for y in ys:
projection = Projection(data, X, y)
fi = projection.decode()
var.append(fi)
投影类和解码方法如下：
class Projection(object):
def __init__(self, data, X, y=0):
# data is dataframe, X is indecies of independent variables and y is index of d
浏览 4提问于2021-12-04得票数 1
1回答
我试图使用这样的方式来使用dask分发来并行一个嵌套循环：
@dask.delayed
def delayed_a(e):
a = do_something_with(e)
return something
@dask.delayed
def delayed_b(element):
computations = []
for e in element:
computations.add(delayed_a(e))
b = dask.compute(*computations, scheduler='distributed
浏览 1提问于2019-03-10得票数 0
回答已采纳
1回答
我需要读取mongo_db转储BSON文件。
我正在考虑如何通过编写自定义的DASK bson_reader来读取这些文件。
问题是解析mongo_db BSON文件并遍历所有文件。当迭代时，BSON需要找到BSON块的结尾，以分离每个文件块并防止将整个文件加载到内存中。pymongo bson模块可以做到这一点，但是只返回一个文件迭代器，这个迭代器不能用于例如:dask.bug.load_csv(文件).map(迭代器)。
写自定义dask.bug.bson_read的样板是什么？或者有什么想法？
浏览 2提问于2020-03-12得票数 0
我拥有的数据规模不太可能使DataFrame合并成功--以前的尝试已经导致过多的数据洗牌、调度程序上的内存错误和工作人员中的通信超时，甚至包括索引、分区、工作人员的重要计数、总内存等等。
通过将数据写入小文件并在需要查找时将其读取回来，我已经“手动”合并了一些成功。我们目前正在dask.delayed函数中这样做。这显然需要大量磁盘I/O。
Dask延迟最佳实践()警告不要发送DataFrame延迟，不要调用延迟形式延迟，并告诉我们在分布式场景中避免全局状态。这些最佳实践让我相信，没有一种安全的方法可以使用延迟函数中的DataFrame --我的理解是正确的吗？
不幸的是，数据的规模和敏感性使
浏览 2提问于2021-09-20得票数 1
回答已采纳
我有一个非常大的数据文件，我正在处理dask。dataframe大体如下所示：
Col_1 Col_2 Bool_1 Bool_2
A 1 True False
B 1 True True
C 1 False False
D 1 True False
A 2 False True
B 2 False False
C 2 True False
D
浏览 1提问于2020-03-17得票数 2
回答已采纳
1回答
将立即显示通过客户端提交的函数的日志。相反，预期日志将显示在client.gather(futures)上。预期的行为可以通过延迟而不是期货来实现。
下面是复制问题的代码：
from dask.distributed import Client
client = Client(processes=False, n_workers=2)
def inc(x):
warning(f"{x}")
return x + 1
output=[]
for x in [1, 2, 3, 4, 5]:
a = client.submit(inc, x)
o
浏览 6提问于2022-09-19得票数 0
1回答
如何将dask数据按降序排序？
、、、、
显然，dask中不存在升序关键字，这很有趣，因为dask被设计成类似于pandas。这样做是行不通的：
res = (ddf
.groupby(['An important column'])
.mean()
.sort_values('Score', ascending=False)
.compute()
)
使用dask进行降序排序的最佳方法是什么？
> NotImplementedError: The ascending= keyword is not supported
达斯克版本：2021.4.0
浏览 3提问于2021-04-21得票数 0
1回答
处理大熊猫数据帧
、、、、
我有一个巨大的数据库(500 it左右)，它可以放进熊猫里。数据库包含大约39705210个观测结果。正如您可以想象的那样，python甚至在打开它时都很困难。现在，我尝试使用Dask将其导出到cdv到20个分区，如下所示：
import dask.dataframe as dd
dask_merge_bodytextknown5 = dd.from_pandas(merge_bodytextknown5, npartitions=20) # Dask DataFrame has 20 partitions
dask_merge_bodytextknown5.to_csv('df_
浏览 2提问于2022-06-29得票数 2
回答已采纳
我正在尝试是否可以使用Dask在集群上对海量2D图像(约20-50 GB)中的对象进行按块并行检测和分割。
我检测/分割图像块中对象的逻辑将封装在一个函数中。
我遇到了一个名为的Dask函数，它允许我在dask数组的每个块/块上应用一个自定义函数。
但是，我看到我可以传递给map_blocks的函数的输出类型也应该是一个数组。
对于对象检测/分割，我希望我的函数能够返回在块中找到/检测到的每个对象的边界轮廓的坐标。请注意，任何块中的对象数量都是未知的，并且取决于图像。
我如何用map_blocks或Dask中的其他东西来解决这个用例？
浏览 3提问于2016-11-23得票数 0
我有一个最初使用concurrent.futures.ProcessPoolExecutor运行的计算工作量，我将其转换为使用dask，这样我就可以利用dask与分布式计算系统的集成来扩展到一台机器之外。工作负载由两种任务类型组成：
Task :接受字符串/浮点数输入并生成一个矩阵(大约为2000x2000)。任务持续时间通常是60秒或更短。任务B:从任务A获取矩阵，并使用它和其他一些小输入来求解一个常微分方程。解决方案被写入磁盘(因此没有返回值)。任务持续时间最长可达15分钟。
每个A任务可以有多个B任务。
最初，我的代码如下所示：
a_results = client.map(calc_a
浏览 2提问于2021-07-19得票数 1
回答已采纳
我有5个CSV文件，每个文件包含大约100万条记录。我正在使用dask尝试以并行的方式读取它们，并计算每个记录中的记录，然后求和以获得记录的总数。下面是我的代码： import dask.dataframe as dd
import dask
counts = []
def read_file(fn):
df = dask.delayed(dd.read_csv)(fn)
return len(df.index)
for i in range(5):
filename="c://parallel//test"+str(i)+".csv&#
浏览 22提问于2020-08-19得票数 0
环境信息： 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
回答已采纳
1回答
所以基本上我想要的是并行运行ML流水线。我一直在使用scikit learn，我决定使用DaskGridSearchCV。
我有一个gridSearchCV = DaskGridSearchCV(pipeline, grid, scoring=evaluator)对象的列表，我按顺序运行每个对象：
for gridSearchCV in list:
gridSearchCV.fit(train_data, train_target)
predicted = gridSearchCV.predict(test_data)
如果我有N个不同的GridSearch对象，我希望尽可能多
浏览 0提问于2017-05-08得票数 5
2回答
我有一个小的数据have (大约100 to )和一个昂贵的计算，我想为每一行执行。它不是一个可向量化的计算；它需要对每一行进行一些解析和一个DB查找。
因此，我决定尝试使用Dask来并行化任务。任务是“令人尴尬的平行”，执行顺序或重复执行是没有问题的。然而，由于一些未知的原因，内存使用量高达大约100 up。
下面是违规代码示例：
import pandas as pd
import numpy as np
import dask.dataframe as dd
from dask.distributed import Client
from dask_jobqueue import LS
浏览 1提问于2021-03-25得票数 1
回答已采纳
背景
我有一个列表，其中包含了经过预处理并保存为.npy二进制文件的1000个图像堆栈(3D numpy数组)的路径。
案例研究，，我想计算所有图像的平均值，为了加快分析速度，我想并行处理。
使用dask.delayed的方法
# List with the file names
flist_img_to_filter
# I chunk the list of paths in sublists. The number of chunks correspond to
# the number of cores used for the analysis
chunked_list
# Sc
浏览 0提问于2017-10-26得票数 4
回答已采纳
2回答
我有一个定制的文件格式，我想要懒洋洋地加载和处理，如果它是一种数据格式，它将是有用的。
我已经有了一个函数来创建熊猫的数据格式，但是运行起来代价很高。dask.dataframe.from_delayed逻辑地合并所有数据帧，就好像它们是一个可迭代的数据帧一样。当调用时，from_delayed遍历迭代(示例代码中的生成器)并创建延迟(pd.dataframe)的。
我的问题是，需要读取数据集才能生成数据格式，但我希望尽可能晚地读取文件(在生成器上迭代)。
def gen():
for i in range(10):
sleep(10) #Looping this generato
浏览 5提问于2022-05-04得票数 0
1回答
我的问题可能很蠢，但我才刚开始学习达斯克不被信任。任何帮助都是非常感谢的。
我有如下代码：
@dask.delayed
def do_something(date):
return x, y
get_item0 = dask.delayed(operator.itemgetter(0))
get_item1 = dask.delayed(operator.itemgetter(1))
def handle_x(list_x):
# do something
print(len(list_x))
浏览 2提问于2021-09-28得票数 0
这是pure在Dask中的有效使用吗？
让我们假设bar在计算过程中从未改变过，但是在我设置计算时它可能会改变。
from dask import delayed
class a:
def __init__(self):
self.bar = 1
def foo(self, b):
return self.bar + b
def dask_foo(self, b):
return delayed(self.foo, pure=True)(b)
浏览 1提问于2018-06-30得票数 2
回答已采纳
2回答
我有一个函数，返回两个元素的元组。函数通过池星图调用，以生成元组列表，这些元组被解压缩为两个列表。
def func():
#...some operations
return (x,y)
def MP_a_func(func,iterable,proc,chunk):
pool=multiprocessing.Pool(processes=proc)
Result=pool.starmap(func,iterable,chunksize=chunk)
pool.close()
return Result
##
if __name__ ==
浏览 0提问于2021-03-12得票数 2
回答已采纳
我有一个dask延迟计算的结果，它可以是一个浮点数字、一个数字列表或一个numpy数组。是否有一种标准的dask方法将这些结果保存到本地磁盘或HDFS？
浏览 1提问于2019-05-21得票数 0
回答已采纳
我不了解dask分布式系统中的弹性模型。
问题
工人提出的异常会杀死令人尴尬的并行dask操作。如果任何员工遇到异常，所有员工和客户都会死亡。
预期行为
在这里阅读：建议，异常应该包含在工人身上，随后的任务将不受干扰地进行。
“当函数引发错误时，该错误将根据请求保存并发送给客户端。任何收集该结果的尝试或任何依赖的结果都会导致exception...This不会以任何方式影响调度程序或工作人员的顺利运行。”
我在这里跟踪一个令人尴尬的并行用例：
可复制示例
import numpy as np
np.random.seed(0)
from dask import compute, de
浏览 1提问于2018-12-22得票数 0
1回答
我有这样的代码，它调用一个@delayed函数，该函数以N个dask数据格式作为输入，并返回一个dataframe作为输出。
有两个问题：(1)函数内部的数据类型是熊猫而不是达斯克；(2)当我得到函数的结果时，它也是熊猫而不是达斯克。
作为熊猫数据输入而不是达克，@delayed背后的逻辑是什么？我只需要和达克数据处理一起工作。
这是代码：
df = pd.DataFrame({
'height': [6.21, 5.12, 5.85, 5.78, 5.98],
'weight': [150, 126, 133, 164, 203]
})
d
浏览 4提问于2022-03-04得票数 0
回答已采纳
我在一台机器上使用Dask (LocalCluster有4个进程，16个线程，68.56 to内存)，当我试图一次计算两个共享依赖关系的结果时，遇到了工作内存问题。
在下面显示的示例中，仅用一次计算就可以很好地快速计算result，工作人员的总内存使用量最高可达1 1GB左右。但是，当使用两次计算来计算results时，当总内存使用量约为40 is时，工作人员会迅速耗尽所有内存，并开始写入磁盘。计算最终将完成，但一旦开始写入磁盘，就会有一个巨大的减速。
直观地说，如果读入一个块，然后立即计算它的两个和，那么该块可以被丢弃，并且内存使用率保持在较低水平。然而，看起来Dask正在优先加载数据，而不
浏览 18提问于2021-02-27得票数 2
回答已采纳
1回答
我有一个17 GRIB的GRIB文件，其中包含2020年每小时的温度()数据。数据集的维度是longitude、latitude和time。
我的目标是计算全年数据中每个坐标的最高温度(lon，lat)。我可以使用Xarray很好地加载文件，尽管它需要4到5分钟：
import xarray as xr
xarray_dataset = xr.open_dataset('cds/2020_hourly_t2m.grib', engine='cfgrib')
但是，调用会导致Google会话崩溃。这可能是因为它需要比可用内存更多的内存。
因此，我可能需要使用Das
浏览 2提问于2022-01-02得票数 2
回答已采纳
1回答
我有一个dataframe，它包含列中的图像路径(称为img_paths)。在接下来的步骤中，我想要做的是使用这些图像路径将图像加载到另一列(称为img_loaded)中，然后应用一些预处理功能。
但是，在加载(或，图像读取)过程中，我总是得到不同的结果，包括对imread函数的一个时间延迟包装，另一个时间正确加载图像(我可以看到数组)和其余时间：FileNotFoundError。
除了下面的例子之外，我还使用了map_partitions函数，但是除了没有数组之外，我还在类似的输出中结束。最后，我想使用map_partitions函数而不是apply函数。
以下是我的代码和有关问题的描述：
浏览 0提问于2019-06-16得票数 1
回答已采纳
1回答
我有一个dask数组，X，大约是。3500x700000。每一行都有缺失值，我需要根据该行的平均值进行推算。
目前这个数组被存储为dask数组，并且我有一个可以执行补偿的延迟函数。然而，当我使用计算方法时，内存被填满了，因为它似乎试图给我留下一个numpy数组。
有没有一种方法可以将这种估算应用于适当的数组，以便最终将估算的数组作为内存不足的dask数组？
浏览 11提问于2018-09-09得票数 2
1回答
迭代dask延迟的字典
、、、
我有很多从dask延迟函数返回的延迟dict。我想将它们聚合成一个summary_dict，如下所示。items函数不适用于延迟对象。 @dask.delayed
def get_dict(date):
return {
'a': {'date': date},
'b': {'date': date}
}
summary_dict = {}
dates = [d1, d2, d3, ...]
for date in dates:
date_dict = get_dict(
浏览 14提问于2021-11-19得票数 1
1回答
我想看看Dask是否适合我的项目，并编写了一些非常简单的测试用例来研究它的性能。然而，Dask只需花费相对较长的时间来执行延迟初始化。
@delayed
def normd(st):
return st.lower().replace(',', '')
@delayed
def add_vald(v):
return v+5
def norm(st):
return st.lower().replace(',', '')
def add_val(v):
return v+5
test_lis
浏览 0提问于2018-12-04得票数 3
当我们需要训练一堆数据(大约22GiB)时，我用两种方法进行了测试，生成随机数据，并尝试用Dask训练它，然而，当Dask.array one工作时，Numpy生成的数据会引发异常(消息包:字节对象太大)。有人知道为什么吗？ from dask.distributed import Client
from dask_cuda import LocalCUDACluster
from dask import array as da
import numpy as np
import xgboost as xgb
import time
def main(client):
regre
浏览 14提问于2021-04-10得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
