将包含dask的数据集复制到所有工作进程
Dask是一个用于并行计算的灵活的开源库，它可以在分布式环境中处理大规模数据集。它提供了类似于Pandas和NumPy的API，但可以处理比内存更大的数据集。
将包含Dask的数据集复制到所有工作进程是指在分布式计算中，将数据集复制到所有参与计算的工作进程中，以便并行处理数据。这样做的好处是可以减少数据传输的开销，提高计算效率。
在云计算领域，可以使用腾讯云的一些相关产品来实现将包含Dask的数据集复制到所有工作进程。以下是一些推荐的腾讯云产品和产品介绍链接地址：
- 腾讯云容器服务（Tencent Kubernetes Engine，TKE）：TKE是一种高度可扩展的容器管理服务，可以帮助用户快速部署、管理和扩展容器化应用。通过在TKE上部署Dask集群，可以实现数据集的复制到所有工作进程。了解更多信息，请访问：腾讯云容器服务
- 腾讯云弹性MapReduce（EMR）：EMR是一种大数据处理和分析服务，可以帮助用户快速处理大规模数据集。通过在EMR上配置Dask集群，可以实现数据集的复制到所有工作进程。了解更多信息，请访问：腾讯云弹性MapReduce
- 腾讯云云服务器（CVM）：CVM是一种高性能、可扩展的云服务器，可以用于部署和运行Dask集群。通过在多个CVM实例上部署Dask集群，可以实现数据集的复制到所有工作进程。了解更多信息，请访问：腾讯云云服务器
需要注意的是，以上推荐的腾讯云产品仅供参考，具体的选择和配置取决于实际需求和场景。同时，还可以结合其他腾讯云产品和服务，如对象存储、负载均衡、虚拟专用网络等，来构建完整的分布式计算环境。
相关·内容
如何让dask中的所有其他工作节点都可以访问一个8 GB的大文件？我已经在chunksize和client.scatter上尝试过pd.read_csv()，但这需要很长时间。我正在macOS上运行它。
这是我的代码：
import time
import pandas as pd
import dask as dask
import dask.distributed as distributed
import dask.dataframe as dd
import dask.delayed as delayed
from dask.distributed import Client, pr
浏览 4提问于2018-07-24得票数 1
3回答
我是新来Dask的，所以如果你觉得这个问题很愚蠢，请原谅我。在Dask中，我正在使用一个包含大约50 of数据的Dask数据帧。此数据是字符串数据，在将其提供给机器学习算法之前，我需要对其进行预处理(使用该过程快速)(使用线程快速)。现在的问题是，当我设计一个关于进程的集群时，数据帧操作是快的，但是关于线程，它是慢的(但线程在机器学习中是快的)。因此，我正在寻找一种解决方案，在这种方案中，我们可以从进程切换到线程环境。
目前，我正在使用进程集群保存预处理数据，然后关闭它，并使用线程环境启动一个新的集群，以应用机器学习。
有什么替代方案可以解决这个问题吗？
请在这方面帮助我。
浏览 30提问于2021-02-17得票数 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
回答已采纳
我正在使用大型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实现从一台机器到Slurm集群的机器学习(Scikit-Learn)。根据一些教程(例如https://examples.dask.org/machine-learning/scale-scikit-learn.html)，使用job_lib.parallel_backend('dask')非常简单。然而，读入数据的位置让我感到困惑，并且没有任何教程提到它。我应该使用dask.dataframe来读取数据以确保数据被传递到集群，还是使用pd.dataframe读取数据并不重要(然后数据存储在我运行Jupiter notebook的机器的随机存取存储器中)
浏览 28提问于2020-07-05得票数 1
回答已采纳
1回答
我在试着决定是Spark还是Dask能给我们的工作带来更好的表现。我有一个简单的脚本，可以在DataFrame上运行一些操作。
我不相信我使用分布式版本是正确的，因为时间比本地使用dask慢得多。下面是我的脚本：
def CreateTransactionFile(inputFile, client):
startTime = time.time()
df = dd.read_csv(inputFile)
mock = pd.DataFrame([[1,1, datetime.today(), 1,1]], columns=['A', '
浏览 2提问于2018-11-29得票数 1
我正在尝试使用dask_cudf/dask读取单个大的parquet文件(size > gpu_size)，但它当前正在将其读取到单个分区中，我猜测这是从文档字符串推断出的预期行为： dask.dataframe.read_parquet(path, columns=None, filters=None, categories=None, index=None, storage_options=None, engine='auto', gather_statistics=None, **kwargs):
Read a Parquet file into a D
浏览 18提问于2019-10-18得票数 3
回答已采纳
1回答
当在集群上设置工作人员时，我对dask和dask.distributed中使用的不同术语感到有点困惑。
我遇到的术语是:线程、进程、处理器、节点、工作人员、调度程序。
我的问题是如何设置每一个的数目，如果有一个严格的或建议的关系，其中任何一个。例如：
每个节点上有n个进程的每个节点有一个工作人员
线程和进程是相同的概念吗？在dask中，我必须设置n线程，但它们在客户端显示为进程。
还有其他建议吗？
浏览 1提问于2018-06-29得票数 26
回答已采纳
我有这样一个小数据集：
<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
1回答
我有下面的项目树
.
└── src
└── dask_test
├── helpers
│ ├── commandline.py
│ ├── data
│ │ ├── dataset0.json
│ │ ├── dataset1000.json
│ │ ├── dataset300.json
│ │ ├── dataset5000.json
│ │ ├── dataset500.json
│
浏览 5提问于2017-06-24得票数 2
我需要从一个大的numpy数组中保存一个dask数组。下面是一个最小的工作示例，展示了该过程。请注意，a是使用numpy.random创建的，但不幸的是，我不能使用dask创建数组。
import numpy as np
import dask.array as da
from dask.distributed import Client
a = numpy.random.randint(0,2,size=4000000*8000).reshape((4000000,8000))
# here the conversion and saving
client = Client(n_worke
浏览 22提问于2020-02-20得票数 0
回答已采纳
假设您想要指定Dask.array中的工作进程数，如所示，您可以设置：
dask.set_options(pool=ThreadPool(num_workers))
这对于我已经运行过的一些模拟非常有效，例如，montecarlo，但是对于一些线性代数操作，Dask似乎覆盖了用户指定的配置，例如：
import dask.array as da
import dask
from multiprocessing.pool import ThreadPool
dask.set_options(pool=ThreadPool(num_workers))
mat1 = da.random.rand
浏览 4提问于2017-02-24得票数 2
1回答
我正在尝试使用sklearn来集群相当大的数据集(150 k样本和150 k特性)。我想我可以更快地使用来自dask_ml的增量来将我的数据放入块中。下面是虚拟数据集上我的代码片段：
from dask_ml.datasets import make_blobs
from dask_ml.wrappers import Incremental
from sklearn.cluster import MiniBatchKMeans
import dask.array as da
import dask
dataset = da.random.ra
浏览 4提问于2021-06-14得票数 0
1回答
寻找嵌套并行作业的最佳实践。我不能嵌套dask delayed或futures，所以我混合了这两种方式来让它正常工作。这不是建议的做法吗？有没有更好的方法呢？示例：
import dask
from dask.distributed import Client
import random
import time
client = Client()
def rndSeries(x):
time.sleep(1)
return random.sample(range(1, 50), x)
def sqNum(x):
time.sleep(1)
return x*
浏览 5提问于2021-02-05得票数 0
1回答
dask.distributed将数据保存在中，直到不再需要该数据为止。(谢谢@MRocklin!)
虽然这一过程在网络使用方面是有效的，但它仍然会导致频繁的数据酸洗和数据的不筛选。我假设这不是一个零副本筛选，就像plasma或joblib在numpy数组的情况下所做的一样。
显然，当另一个主机或其他工作人员需要计算结果时，必须对其进行腌制和发送。当依赖同一主机内的线程并行时，可以避免这种情况，因为所有计算都访问相同的内存(--nthreads=8)。dask正在使用这个技巧，只访问计算结果而不是酸洗。
当我们使用基于进程的并行化，而不是在单个主机(--nproc=8 --nthreads=1
浏览 3提问于2020-07-03得票数 1
1回答
我正在尝试使用Dask读取一个包含非常大的csv文件的文件夹(这些文件都可以放在内存中，它们非常大，但我有很多RAM) --我目前的解决方案如下：
val = 'abc'
df = dd.read_csv('/home/ubuntu/files-*', parse_dates=['date'])
# 1 - df_pd = df.compute(get=dask.multiprocessing.get)
ddf_selected = df.map_partitions(lambda x: x[x['val_col'] == va
浏览 3提问于2017-10-06得票数 2
我完全被困住了，所以我正在寻求善意的建议。我的目标是并行读取多个hdf5文件，提取内部的多个模糊数组，并将每个数组存储在一个dataframe的一行，准确地说是一个单元格中。我不选择熊猫df，因为我相信它会太大。
不可能使用read_hdf()从用h5py创建的hdf5文件中读取。我能做些什么来用paralleL中的dask导入数千个hdf5 5文件，并访问其中的多个do数组？我想要创建一个dataframe，其中每个2d数组(从hdfs中的n个暗数组中提取)都存储在dataframe的一个单元格中。因此，行数对应于在所有文件中找到的总数组数，在这里9。我将数组存储在一个列中。在将来我希望将更
浏览 0提问于2021-09-30得票数 1
回答已采纳
3回答
我使用Dask读取了2.5 to的csv文件，Python给了我错误。这是我写的代码：
import pandas as pd
import numpy as np
import time
from dask import dataframe as df1
s_time_dask = time.time()
dask_df = df1.read_csv('3SPACK_N150_7Ah_PressureDistributionStudy_Data_Matrix.csv')
e_time_dask = time.time()
以下是我从Python得到的错误：
dask_df
浏览 0提问于2021-06-17得票数 0
我有一个dask Client和它的工作人员。 我希望ho把我的计算分成两步： 1)每个worker运行一次预计算代码(吃小的设置对象，计算速度慢，生成相当大的中间结构)，保存每个worker的中间数据。 2)运行计算函数(比预计算快很多，每个worker运行多次，并使用每个worker保存的中间数据)。 我该怎么做呢？
浏览 33提问于2019-03-01得票数 1
1回答
我有一个数据集存储在一个标签分隔的文本文件中。该文件如下所示：
date time temperature
2010-01-01 12:00:00 10.0000
...
其中temperature列包含以摄氏度(°C)为单位的值。我用达斯克计算日平均温度。这是我的代码：
from dask.distributed import Client
import dask.dataframe as dd
client = Client("<scheduler URL")
inputDataFrame = dd.read_table("<in
浏览 1提问于2018-10-12得票数 0
2回答
Dask仪表板是空的
、、、
我有几个python脚本，它们从不同的数据库运行不同的Dask任务，我使用Python多处理模块同时运行所有Python脚本。我与任务管理器检查了脚本是否并行运行，并且能够访问我的dask仪表板。然而，我的达斯克仪表盘没有显示任何东西。这是我的dask仪表盘的截图。
这是我的Python代码片段示例(简化)：
if __name__ == '__main__':
# Setup Dask Distributed Client
client = Client(n_workers=4, threads_per_worker=4)
""
浏览 7提问于2022-03-01得票数 0
回答已采纳
1回答
我有一个API，它从MongoDB (用pymongo)加载数据，然后在熊猫上应用相对“复杂”的数据转换，比如在datetime列上groupby，将频率参数化等等。由于我对熊猫比芒果更精通，所以我更喜欢按原样做，但我不知道用mongo聚合查询来编写这些转换是否会更快。
为了简化这个问题，不要考虑在双方编写查询时遇到的困难:在mongo上做一个简单的群并选择*结果或选择*在熊猫/dask(在分布式场景中)中这样做更快吗？在大型数据集中，前者比第二个更快/更慢，还是更小？
浏览 2提问于2020-05-19得票数 1
回答已采纳
我可以在一台机器上使用Dask没有问题。
import dask.dataframe as dd
a = dd.read_parquet("c:/data/p03.parquet", columns="var0")
aa=a.value_counts().compute()
但是我如何知道它使用了多少个工作人员(或进程)呢？我试过了
from dask.distributed import Client
Client()
它显示了
Client:调度器=‘tcp://127.0.0.1:56688’processes=4 cores=12
这是否意味着它
浏览 0提问于2019-08-23得票数 0
1回答
我正在尝试理解Dask的Client()函数参数。假设我使用Dask的Client()函数编写了以下代码：
from dask.distributed import Client
import joblib
client = Client()
如果我没有为Client()函数中的参数指定任何值，那么参数的默认值是什么：
(i) n_workers
(ii) threads_per_worker
(iii) memory_limit
据我所知，Python具有防止多线程的全局解释器锁(GIL)功能。如果是这样的话，为什么当多线程在Python中被阻止时，达斯克-ML的Client()函数有参数
浏览 1提问于2020-05-28得票数 1
回答已采纳
1回答
这与有关
再考虑一下这个简单(但很有趣)的例子：
import dask.dataframe as dd
import dask.multiprocessing
import dask.threaded
from fuzzywuzzy import fuzz
import pandas as pd
master= pd.DataFrame({'original':['this is a nice sentence',
'this is another one',
'stackoverflow is nice']})
slave
浏览 1提问于2018-04-15得票数 3
回答已采纳
我试图通过Dataframe保存到与Dask调度程序/工作人员所在的同一台机器上。但是，我在这段时间里遇到了麻烦。
My安装程序：我的python脚本在本地计算机(笔记本电脑16 GB RAM)上执行，但该脚本为运行在远程计算机上的Dask调度程序(用于并行计算的具有400 GB RAM的服务器)创建了Dask客户端。Dask调度程序和工作人员都位于同一台服务器上，因此它们都共享本地可用的相同的文件系统。由于我的团队的所有成员都使用这个远程Dask调度程序，我们正在处理的文件也位于同一服务器上，通过同一个Dask集群向所有成员提供对所有文件的公共访问。
我试过：
# This saves th
浏览 2提问于2019-12-20得票数 2
回答已采纳
我正在以一种自动化的方式启动一个dask集群，方法是通过ssh-ing到一堆机器并运行dask-worker。我注意到，当以前的实验中的进程仍在运行时，我有时会遇到问题。在达斯克之后，什么是最好的清理方法？killall dask-worker dask-scheduler似乎没有做到这一点，可能是因为达斯克以某种方式启动了新的进程。
浏览 0提问于2018-07-29得票数 2
我正在努力将字典的dask.bag转换为dask.delayed pandas.DataFrames，使之成为最终的dask.dataframe
我有一个函数(make_dict)将文件读入相当复杂的嵌套字典结构，另一个函数(make_df)将这些字典转换为pandas.DataFrame (由此产生的数据为每个文件大约100 mb )。为了进一步分析，我想将所有数据文件附加到一个dask.dataframe中。
到目前为止，我一直在使用dask.delayed对象来加载、转换和附加所有工作正常的数据(参见下面的示例)。但是，对于以后的工作，我想使用dask.bag将加载的字典存储在dask.
浏览 3提问于2019-03-22得票数 5
回答已采纳
嗨，我是新来的达斯克，似乎找不到相关的例子，这个标题的主题。如有任何相关文件或帮助，将不胜感激。
我使用的示例是使用dask_cloudprovider库对蔚蓝环境中的图像数据集进行预处理，我希望通过在一组机器上划分工作来提高处理速度。
根据我所阅读和测试的内容，我可以(1)将数据加载到客户端计算机上的内存中，并将其推送给工作人员或
'''psudo code
load data into array and send it to workers through delayed func'''
(2)我可以在每个工作节点和数据存储之间建立一个链接
浏览 1提问于2021-02-02得票数 0
回答已采纳
我使用的是在GCP上Kubernetes引擎的Kubernetes集群上使用Helm部署的Dask。我当前的集群设置有5个节点，每个节点有8个cpus，30 gb：
我运行了名为05-nyc-taxi.ipynb的笔记本，这导致了工作人员的死亡。
当我重新启动Dask客户端时，它显示我现在没有工作进程和内存：
但是，当我运行kubectl get services和kubectl get pods时，它显示我的pods和服务正在运行：
你知道为什么会这样吗？
浏览 1提问于2019-03-06得票数 1
1回答
我使用以下代码创建了一个Dask SSH Cuda集群：
from dask.distributed import Client, SSHCluster
cluster = SSHCluster(
["localhost", "192.168.1.119", "192.168.1.191"],
connect_options={"known_hosts": None,"username": "vinhdiesal"},
worker_options={
浏览 0提问于2020-07-01得票数 0
1回答
我正在努力理解BlazingSQL是一个竞争对手还是dask的补充。
我有一些中等大小的数据(10-50GB)保存在Azure blob的存储空间中。
IIUC我可以使用BlazingSQL语法查询、连接、聚合和使用CuDF，但我也可以使用dask_cudf将数据读入CuDF，并使用python/dataframe语法执行所有相同的操作。
在我看来，他们是直接的竞争对手？
使用dask的好处之一是它可以对分区进行操作，这样就可以对大于GPU内存的数据集进行操作，而BlazingSQL仅限于适用于GPU的数据集，这是正确的吗？
为什么要选择使用BlazingSQL而不是dask呢？
编辑：
谈到
浏览 1提问于2020-01-18得票数 3
回答已采纳
2回答
据信，Dask框架能够处理比RAM更大的数据集。然而，我未能成功地将它应用于我的问题，听起来如下所示：
我有一个巨大的.csv文件(1.8Gb)，其中包含用户评论的文本，内存为8Gb。其目标是对给定的数据进行预处理(首先对句子进行标记)。为了实现这一点，我运行以下代码：
if __name__ == '__main__':
client = Client(n_workers=3, memory_limit='1.5GB', processes=True)
df = dd.read_csv('texts_no_n', dtype={'us
浏览 5提问于2020-05-20得票数 1
回答已采纳
1回答
有了500个连续增长的DataFrames，我想向dask提交(对于每个DataFrame独立的)数据的操作。我的主要问题是：dask 保存连续提交的数据，因此我可以对所有提交的数据(而不仅仅是新提交的？)执行 submit 函数。
但让我们在一个例子中解释一下：
创建dask_server.py
from dask.distributed import Client, LocalCluster
HOST = '127.0.0.1'
SCHEDULER_PORT = 8711
DASHBOARD_PORT = ':8710'
def run_cluster()
浏览 5提问于2020-05-13得票数 5
开发人员是否应该通过发布大型数组来使用Dask作为数据库？如果是这样的话，建议的工作流应该做什么来增加ram Dask的访问权限(除了修改机器本身)？
浏览 4提问于2017-06-13得票数 0
回答已采纳
1回答
如何处理并行返回大结果的小数据帧
、、、、
我有一个大约6000万行的Pandas DataFrame。前60行对应于第一组，依此类推。这些组中的每一个都需要并行处理，并且每个组返回一个大于4 4GB的NumPy数组。我有足够的RAM和内核来并行处理大约100个这样的组。最后，我必须将每个组的结果相加才能得到最终结果，因此最终结果不会大于每个组的结果。
处理过程大致是:将给定大小的网格(NumPy数组)上的一组数据打包，并计算该网格与自身的张量外积(numpy.multiply.outer)。然后对得到的乘积求和，得到最终的数组。
如果我按顺序处理这些组，处理过程可能需要几天时间。因此，我需要并行处理它们。
首先，我尝试了multipr
浏览 28提问于2021-03-11得票数 1
我正在尝试使用dask将一个大的dask数组(46 GB，124 -- 370MB块)写入zarr文件。如果我的dask数组被命名为dask_data，那么一个简单的dask_data.to_zarr("my_zarr.zarr")就可以工作。但据我所知，这是一个同步的、受CPU限制的进程。
我想要做的是使用并行，将大部分工作分配给Quadro GV100图形处理器。我尝试通过dask_data_cupy = dask_data.map_blocks(cupy.asarray)将numpy.ndarray转换为cupy.ndarray，并将其写出到zarr文件中，但我收到以下消
浏览 0提问于2020-02-08得票数 1
1回答
我正在钻研达斯克，(大部分)感觉很舒服。但是，我无法理解在下面的场景中发生了什么。我肯定以前有人问过这样的问题，但经过一段时间的搜索后，我似乎找不到一个真正击中头部的问题。所以我们到了！
在下面的代码中，您可以看到一个简单的python函数，函数上有一个Dask延迟装饰器。在我真正的用例场景中，这将是一个“黑匣子”类型的函数，在这个函数中，我不关心发生了什么，只要它保持4GB内存预算，并最终返回一个熊猫数据。在本例中，我特别选择了值N=1.5e8，因为这导致内存占用总量接近2.2GB(很大，但仍在预算范围内)。最后，当以脚本的形式执行这个文件时，我有一个“数据管道”，它只运行一些ID的黑匣子函
浏览 2提问于2021-04-26得票数 3
Local dask允许使用进程调度程序。dask distributed中的工作人员正在使用ThreadPoolExecutor计算任务。在dask distributed中用ProcessPoolExecutor替代ThreadPoolExecutor是可能的吗？谢谢。
浏览 1提问于2019-01-11得票数 1
我想使用Dask来处理大型数据帧。然而，当我尝试使用它时，我得到了一个内存错误，如下所示。 df = pandas.DataFrame({'x': my_very_large_array})
ddf = dask.dataframe.from_pandas(df, npartitions=100) 我认为Dask应该处理比内存更大的数据。这是怎么回事？
浏览 7提问于2019-03-06得票数 0
回答已采纳
我希望通过在6核macbook pro上使用Dask dataframe而不是Pandas来提高性能。然而，Dask的执行速度与Pandas数据帧一样慢，大约需要5分钟。
我在这里做错了什么？
ddf = dd.from_pandas(df.set_index('customer seq').sort_index(), npartitions = 8)
ddf = ddf.set_index(ddf.index, sorted = True)
paired = ddf.groupby(ddf.index, group_keys =
False).apply(retention_
浏览 1提问于2019-07-12得票数 0
2回答
在对包含numpy数组的FutureWarning进行foldby操作时，我从dask / numpy获得了一条信息非常少的foldby消息。
def binop(a, b):
print('binop')
return a + b[1]
def combine(a, b):
print('combine')
return a + b[1]
seq = ((np.random.randint(0, 5, size=1)[0], np.ones(5,)) for _ in range(50))
db.from_sequence(
浏览 5提问于2017-05-05得票数 3
回答已采纳
已经有了一个很好的问题，但最好的答案是现在已经5年了，所以我认为2018年应该有更好的选择。
我目前正在寻找一个比内存数据集更大的特性工程管道(使用适当的dtype)。
初始文件是一个不适合内存的csv。以下是我的需要：
创建特性(主要使用多列的groupby操作)。
将新特性合并到以前的数据(磁盘上，因为它不适合内存)
对某些ML应用程序使用子集(或全部)列/索引
重复1/2/3 (这是一个类似于day1的迭代过程:创建4个特性，day2: create 4个.)
试着用胸罩和丹克：
首先，我把大csv文件分割成多个小的“地板”文件。这样，dask对于计算新特性是非常
浏览 0提问于2018-03-29得票数 2
我正在使用Dask分布式，并试图从存储在HDFS中的CSV创建一个数据格式。我认为与HDFS的连接是成功的，因为我能够打印dataframe列的名称。但是，当我试图在dataframe上使用len函数或任何其他函数时，会出现以下错误：
pyarrow.lib.ArrowIOError: HDFS file does not exist: /user/F43479/trip_data_v2.csv
我不明白为什么我有这个错误。我想听听你的意见。
这是我的代码：
# IMPORTS
import dask.dataframe as dd
from dask.distributed import C
浏览 2提问于2019-04-30得票数 0
回答已采纳
1回答
我有一个转换(映射)数据的管道。输出是输入数据中的大行，包含二进制格式的音频，输出数据中的行包含提取的二进制特性。
我正在从一个分区的拼花文件中读取输入，并将它写回一个不同的拼花文件--这两个文件都在网络共享上。
据我所知，在分布式dask中，每个工作人员都会将输出发送回调度程序(然后调度程序可能会将输出发送回客户端？)只有这样，调度程序(或客户端)才会将其写入网络共享。这是正确的吗？
如果是的话，如果数据很大，并且带宽是一个问题，那么在这种情况下，似乎存在冗余通信--为什么工作人员不能将输出直接发送到最终目的地(在这种情况下，网络共享)？当然，共享需要对所有工作人员可用，并且需要有人同步写入
浏览 1提问于2019-03-27得票数 1
回答已采纳
1回答
我有4台机器，M1，M2，M3和M4。调度程序、客户端和工作人员运行在M1上。我已经在M1中放置了一个csv文件。其余的机器都是工人。
当我在dask中使用read_csv文件运行程序时。它给了我错误，文件找不到
浏览 1提问于2018-06-22得票数 3
回答已采纳
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
回答已采纳
有时，使用dask.dataframe.map_partitions进行合并之类的操作很有吸引力。在某些情况下，当使用map_partitions在left_df和right_df之间进行合并时，我希望在执行合并之前先预缓存right_df，以减少网络开销/本地混洗。有什么明确的方法可以做到这一点吗？感觉应该可以使用client.scatter(the_df)、client.run(func_to_cache_the_df)或其他智能广播中的一种或两者的组合。
在使用小得多的right_df (本质上是一个查找表)的大型left_df上执行左连接时，这一点尤为突出。感觉这个right_df应该
浏览 0提问于2019-07-30得票数 2
我有一个函数f，它使用一个变量x作为输入，它是一个大型np.ndarray (伦特20000)。
执行f所需的时间非常少(大约5ms)。
多行矩阵M上的一个M循环
for x in M:
f(x)
与使用多处理并行化相比，需要花费大约5倍的时间。
import multiprocessing
with multiprocessing.Pool() as pool:
pool.map(f, M)
我曾经尝试过与dask并行化，但是它甚至在顺序执行中失败了。相关的职位是，但接受的答案不适用于我。我尝试过很多事情，如使用数据的分区(如所说的)或使用dask.bag。我在本地机器上运
浏览 7提问于2022-03-07得票数 0
1回答
我使用dask_yarn (knit的一部分)启动一个Dask Yarn集群，如下所示：
import dask_yarn
cluster = dask_yarn.DaskYARNCluster(env='/home/hadoop/reqs/dvss.zip', lang='en_US.UTF-8')
cluster.start(n_workers=4, memory=5120, cpus=3)
这将在核心节点上为AM请求1个vCore，并将vCores的其余部分提供给工作程序。
在Spark中，我可以选择在yarn-client模式下运行应用程序，这将使驱动
浏览 0提问于2018-04-28得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
