【发布时间】：2023-01-12 18:01:46
【问题描述】：
我在使用时遇到了问题X阵列和...一起SLURM集群从达斯克.
我正在使用熊猫_plink将一些数据加载到X阵列，然后对其进行过滤并进行一些计算。只要我不使用SLUMCluster.在这种情况下，.compute() 会杀死内核。我认为它可能会加载整个文件，即使我只使用它的一部分。
import os
import numpy as np
import pandas as pd
import math
import time
import warnings
from joblib import parallel_backend
from pandas_plink import read_plink
from pandas_plink import read_plink1_bin
from pandas_plink import get_data_folder
from pandas_plink import Chunk
import dask as da
import dask.array
from dask.dataframe import from_pandas
import dask_ml.model_selection as dcv
import dask.multiprocessing
from dask_jobqueue import SLURMCluster
from dask.distributed import Client, progress, LocalCluster
import xarray as xr
np.random.seed(42)
da.config.set({"array.slicing.split_large_chunks": True})
os.environ["TUNE_MAX_PENDING_TRIALS_PG"] = "8"
G = read_plink1_bin('xxx.bed',verbose=True,chunk=Chunk(nsamples=4000, nvariants=4000))
G = G.sel(sample=G.sample[0:1000])
G = G.where(G.chrom == 11, drop=False)
G = G.sel(variant=G.variant[0:100])
G = G.fillna(G.mean(dim='sample'))
print('%.1f MB' % (G.nbytes / 1e6)) #Output: 0.4 MB
print(G.compute())
print(G)
#Output:
<xarray.DataArray 'genotype' (sample: 100, variant: 1000)>
dask.array<where, shape=(100, 1000), dtype=float32, chunksize=(50, 1000), chunktype=numpy.ndarray>
Coordinates: (12/14)
* sample (sample) object '12' '13' ... '15' '16'
* variant (variant) <U13 'variant0' 'variant1' ... 'variant998' 'variant999'
fid (sample) object '12' '13' ... '15' '16'
iid (sample) object '12' '13' ... '15' '16'
father (sample) object '0' '0' '0' '0' '0' '0' ... '0' '0' '0' '0' '0' '0'
mother (sample) object '0' '0' '0' '0' '0' '0' ... '0' '0' '0' '0' '0' '0'
... ...
chrom (variant) object '1' '1' '1' '1' '1' '1' ... '1' '1' '1' '1' '1'
snp (variant) object '123' '124' ... '999'
cm (variant) float64 0.0 0.0 0.0 0.0 0.0 0.0 ... 0.0 0.0 0.0 0.0 0.0
pos (variant) int32 123456 123457 123458 ... 987654 987655 987656
a0 (variant) object 'A' 'C' 'A' 'A' 'A' 'C' ... 'C' 'A' 'T' 'A' 'T'
a1 (variant) object 'G' 'T' 'G' 'C' 'G' 'T' ... 'T' 'G' 'C' 'G' 'C'
这很好用。但是，如果我之前启动了一个集群，它就不再起作用了。
cluster = SLURMCluster(name='dask-cluster',
cores=8,
n_workers=8,
memory='20GB',
walltime='12:00:00')
cluster.adapt(minimum=0, maximum=10)
client = Client(cluster)
在这种情况下，print(G.sel(sample='123',variant='variant0').values) 在内核死机并想要重新启动之前运行了一段时间。我猜出于某种原因它耗尽了内存，即使数组只有 0.4 MB。
对这种行为有什么解释吗？
更新：
我从集群中收到以下错误：
...
2022-04-29 15:11:22,923 - distributed.worker - INFO - -------------------------------------------------
2022-04-29 15:11:22,923 - distributed.core - INFO - Starting established connection
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
bytes_read = self.read_from_fd(buf)
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
response = await retry_operation(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
return await retry(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
return await coro()
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
return await send_recv(comm=comm, op=key, **kwargs)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
response = await comm.read(deserializers=deserializers)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
convert_stream_closed_error(self, e)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32856 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
bytes_read = self.read_from_fd(buf)
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
response = await retry_operation(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
return await retry(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
return await coro()
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
return await send_recv(comm=comm, op=key, **kwargs)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
response = await comm.read(deserializers=deserializers)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
convert_stream_closed_error(self, e)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32854 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
bytes_read = self.read_from_fd(buf)
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
response = await retry_operation(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
return await retry(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
return await coro()
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
return await send_recv(comm=comm, op=key, **kwargs)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
response = await comm.read(deserializers=deserializers)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
convert_stream_closed_error(self, e)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32852 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
bytes_read = self.read_from_fd(buf)
File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
response = await retry_operation(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
return await retry(
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
return await coro()
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
return await send_recv(comm=comm, op=key, **kwargs)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
response = await comm.read(deserializers=deserializers)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
convert_stream_closed_error(self, e)
File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32858 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken. Reconnecting...
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken. Reconnecting...
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken. Reconnecting...
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken. Reconnecting...
2022-04-29 15:21:07,367 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:33851
2022-04-29 15:21:07,368 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:41869
2022-04-29 15:21:07,368 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:41763
2022-04-29 15:21:07,368 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:40759
2022-04-29 15:21:07,379 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,380 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,380 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,380 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,653 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:46739'
2022-04-29 15:21:07,656 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:37165'
2022-04-29 15:21:07,656 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:38269'
2022-04-29 15:21:07,659 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:39827'
2022-04-29 15:21:07,659 - distributed.dask_worker - INFO - End worker
更新2：
以防万一，我正在对PLINK-文件，它会删除数组的其余部分吗？
G = G.isel(sample = slice(0,n_samples), variant = slice(0, n_features),drop=True)
我怀疑在调用 .compute() 时它加载了整个文件，当然这个文件太大而无法放入内存。
但是，我该如何运行模型并获得例如R^2 分数？
例如如果我正在运行如下所示的线性回归，则在调用 mod.score(G, yS).compute() 时内核会崩溃。
from dask_ml.linear_model import LinearRegression
mod = da.delayed(LinearRegression()).fit(G,yS)
mod.score(G, yS).compute()
【问题讨论】：
标签： python jupyter-notebook dask python-xarray slurm