处理大量镶木地板文件时出现 CUDF 错误
CUDF error processing a large number of parquet files
我在一个目录中有 2000 个 parquet 文件。每个 parquet 文件的大小大约为 20MB。使用的压缩是 SNAPPY。每个镶木地板文件都有如下所示的行：
+------------+-----------+-----------------+
| customerId | productId | randomAttribute |
+------------+-----------+-----------------+
| ID1 | PRODUCT1 | ATTRIBUTE1 |
| ID2 | PRODUCT2 | ATTRIBUTE2 |
| ID2 | PRODUCT3 | ATTRIBUTE3 |
+------------+-----------+-----------------+
每列条目都是一个字符串。
我正在使用具有以下配置的 p3.8xlarge EC2 实例：
- 内存: 244GB
- vCPU: 32
- GPU 内存：64GB（每个 GPU 核心有 16GB 内存）
- GPU：4 个 Tesla V100
我正在尝试以下代码：
def read_all_views(parquet_file_lst):
df_lst = []
for file in parquet_file_lst:
df = cudf.read_parquet(file, columns=['customerId', 'productId'])
df_lst.append(df)
return cudf.concat(df_lst)
这在处理前 180 个文件后崩溃并出现以下运行时错误：
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 9, in read_all_views
File "/home/ubuntu/miniconda3/lib/python3.7/site-packages/cudf/io/parquet.py", line 54, in read_parquet
use_pandas_metadata,
File "cudf/_lib/parquet.pyx", line 25, in
cudf._lib.parquet.read_parquet
File "cudf/_lib/parquet.pyx", line 80, in cudf._lib.parquet.read_parquet
RuntimeError: rmm_allocator::allocate(): RMM_ALLOC: unspecified launch failure
在任何给定时间，GPU 和 CPU RAM 都只使用了 10%。
任何想法如何调试这个或相同的解决方法是什么？
cuDF 是一个单一的 GPU 库。 2000 个 20 MB 的文件大约是 40 GB 的数据，这超出了单个 V100 GPU 的内存容量。
对于需要更多单个 GPU 的工作流程，cuDF 依赖于 Dask。以下示例说明如何使用 cuDF + Dask 将数据读取到单个节点中具有多个 GPU 的分布式 GPU 内存中。这不会回答您的调试问题，但应该有望解决您的问题。
首先，我使用几行代码创建了一个包含两个 GPU 的 Dask 集群。
from dask.distributed import Client
from dask_cuda import LocalCUDACluster
import dask_cudf
cluster = LocalCUDACluster() # by default use all GPUs in the node. I have two.
client = Client(cluster)
client
# The print output of client:
#
# Client
# Scheduler: tcp://127.0.0.1:44764
# Dashboard: http://127.0.0.1:8787/status
# Cluster
# Workers: 2
# Cores: 2
# Memory: 404.27 GB
接下来我将为此示例创建几个镶木地板文件。
import os
import cudf
from cudf.datasets import randomdata
if not os.path.exists('example_output'):
os.mkdir('example_output')
for x in range(2):
df = randomdata(nrows=10000,
dtypes={'a':int, 'b':str, 'c':str, 'd':int},
seed=12)
df.to_parquet('example_output/df')
让我们用 nvidia-smi 查看每个 GPU 上的内存。
nvidia-smi
Thu Sep 26 19:13:46 2019
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 410.104 Driver Version: 410.104 CUDA Version: 10.0 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
|===============================+======================+======================|
| 0 Tesla T4 On | 00000000:AF:00.0 Off | 0 |
| N/A 51C P0 29W / 70W | 6836MiB / 15079MiB | 0% Default |
+-------------------------------+----------------------+----------------------+
| 1 Tesla T4 On | 00000000:D8:00.0 Off | 0 |
| N/A 47C P0 28W / 70W | 5750MiB / 15079MiB | 0% Default |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: GPU Memory |
| GPU PID Type Process name Usage |
|=============================================================================|
+-----------------------------------------------------------------------------+
注意这两个值。 GPU 0 上 6836 MB 和 GPU 1 上 5750 MB（我碰巧在这些 GPU 的内存中已经有不相关的数据）。现在让我们用 Dask cuDF 读取两个镶木地板文件的整个目录，然后 persist 它。坚持它会强制计算——Dask 执行是惰性的，所以只需调用 read_parquet 只会向任务图中添加一个任务。 ddf 是一个 Dask DataFrame。
ddf = dask_cudf.read_parquet('example_output/df')
ddf = ddf.persist()
现在让我们再看看nvidia-smi。
Thu Sep 26 19:13:52 2019
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 410.104 Driver Version: 410.104 CUDA Version: 10.0 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
|===============================+======================+======================|
| 0 Tesla T4 On | 00000000:AF:00.0 Off | 0 |
| N/A 51C P0 29W / 70W | 6938MiB / 15079MiB | 2% Default |
+-------------------------------+----------------------+----------------------+
| 1 Tesla T4 On | 00000000:D8:00.0 Off | 0 |
| N/A 47C P0 28W / 70W | 5852MiB / 15079MiB | 2% Default |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: GPU Memory |
| GPU PID Type Process name Usage |
|=============================================================================|
+-----------------------------------------------------------------------------+
Dask 负责为我们在两个 GPU 之间分发数据。
我在一个目录中有 2000 个 parquet 文件。每个 parquet 文件的大小大约为 20MB。使用的压缩是 SNAPPY。每个镶木地板文件都有如下所示的行：
+------------+-----------+-----------------+
| customerId | productId | randomAttribute |
+------------+-----------+-----------------+
| ID1 | PRODUCT1 | ATTRIBUTE1 |
| ID2 | PRODUCT2 | ATTRIBUTE2 |
| ID2 | PRODUCT3 | ATTRIBUTE3 |
+------------+-----------+-----------------+
每列条目都是一个字符串。 我正在使用具有以下配置的 p3.8xlarge EC2 实例：
- 内存: 244GB
- vCPU: 32
- GPU 内存：64GB（每个 GPU 核心有 16GB 内存）
- GPU：4 个 Tesla V100
我正在尝试以下代码：
def read_all_views(parquet_file_lst):
df_lst = []
for file in parquet_file_lst:
df = cudf.read_parquet(file, columns=['customerId', 'productId'])
df_lst.append(df)
return cudf.concat(df_lst)
这在处理前 180 个文件后崩溃并出现以下运行时错误：
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 9, in read_all_views
File "/home/ubuntu/miniconda3/lib/python3.7/site-packages/cudf/io/parquet.py", line 54, in read_parquet
use_pandas_metadata,
File "cudf/_lib/parquet.pyx", line 25, in
cudf._lib.parquet.read_parquet
File "cudf/_lib/parquet.pyx", line 80, in cudf._lib.parquet.read_parquet
RuntimeError: rmm_allocator::allocate(): RMM_ALLOC: unspecified launch failure
在任何给定时间，GPU 和 CPU RAM 都只使用了 10%。 任何想法如何调试这个或相同的解决方法是什么？
cuDF 是一个单一的 GPU 库。 2000 个 20 MB 的文件大约是 40 GB 的数据，这超出了单个 V100 GPU 的内存容量。
对于需要更多单个 GPU 的工作流程，cuDF 依赖于 Dask。以下示例说明如何使用 cuDF + Dask 将数据读取到单个节点中具有多个 GPU 的分布式 GPU 内存中。这不会回答您的调试问题，但应该有望解决您的问题。
首先，我使用几行代码创建了一个包含两个 GPU 的 Dask 集群。
from dask.distributed import Client
from dask_cuda import LocalCUDACluster
import dask_cudf
cluster = LocalCUDACluster() # by default use all GPUs in the node. I have two.
client = Client(cluster)
client
# The print output of client:
#
# Client
# Scheduler: tcp://127.0.0.1:44764
# Dashboard: http://127.0.0.1:8787/status
# Cluster
# Workers: 2
# Cores: 2
# Memory: 404.27 GB
接下来我将为此示例创建几个镶木地板文件。
import os
import cudf
from cudf.datasets import randomdata
if not os.path.exists('example_output'):
os.mkdir('example_output')
for x in range(2):
df = randomdata(nrows=10000,
dtypes={'a':int, 'b':str, 'c':str, 'd':int},
seed=12)
df.to_parquet('example_output/df')
让我们用 nvidia-smi 查看每个 GPU 上的内存。
nvidia-smi
Thu Sep 26 19:13:46 2019
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 410.104 Driver Version: 410.104 CUDA Version: 10.0 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
|===============================+======================+======================|
| 0 Tesla T4 On | 00000000:AF:00.0 Off | 0 |
| N/A 51C P0 29W / 70W | 6836MiB / 15079MiB | 0% Default |
+-------------------------------+----------------------+----------------------+
| 1 Tesla T4 On | 00000000:D8:00.0 Off | 0 |
| N/A 47C P0 28W / 70W | 5750MiB / 15079MiB | 0% Default |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: GPU Memory |
| GPU PID Type Process name Usage |
|=============================================================================|
+-----------------------------------------------------------------------------+
注意这两个值。 GPU 0 上 6836 MB 和 GPU 1 上 5750 MB（我碰巧在这些 GPU 的内存中已经有不相关的数据）。现在让我们用 Dask cuDF 读取两个镶木地板文件的整个目录，然后 persist 它。坚持它会强制计算——Dask 执行是惰性的，所以只需调用 read_parquet 只会向任务图中添加一个任务。 ddf 是一个 Dask DataFrame。
ddf = dask_cudf.read_parquet('example_output/df')
ddf = ddf.persist()
现在让我们再看看nvidia-smi。
Thu Sep 26 19:13:52 2019
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 410.104 Driver Version: 410.104 CUDA Version: 10.0 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
|===============================+======================+======================|
| 0 Tesla T4 On | 00000000:AF:00.0 Off | 0 |
| N/A 51C P0 29W / 70W | 6938MiB / 15079MiB | 2% Default |
+-------------------------------+----------------------+----------------------+
| 1 Tesla T4 On | 00000000:D8:00.0 Off | 0 |
| N/A 47C P0 28W / 70W | 5852MiB / 15079MiB | 2% Default |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: GPU Memory |
| GPU PID Type Process name Usage |
|=============================================================================|
+-----------------------------------------------------------------------------+
Dask 负责为我们在两个 GPU 之间分发数据。