python - 如何使用 RAPIDS 加速管道中由容器分隔的模块

Question

我们有一个功能，可以让用户通过UI界面拖拽一个模块，形成一个数据处理流水线，比如读取数据，做预处理，做分类训练等等，拖拽之后，这些模块会是顺序执行。

每个模块都会启动一个容器(通过k8s)运行，上一个模块处理后的结果作为文件保存到cephfs，下一个模块读取该文件然后执行操作。这个序列化/反序列化过程很慢。我们计划使用 RAPIDS 来加速此管道:通过将数据放入 GPU MEM 来改进模块间数据交换。并使用 cuDF/cuML 代替 Pandas/SKLearn 以获得更快的处理速度。

目前，我们已经确认这些模块可以从Pandas/SKLearn移植到cuDF/cuML，但是因为每个模块都运行在一个容器中，一旦模块运行完，容器就消失了，进程也就消失了，所以，相应的cuDF数据不能继续存在于GPU MEM中。

在这种情况下，如果想使用RAPIDS来改进pipeline，有什么好的建议吗？

Answer 1

如果您希望进程产生和终止并远程访问它们的内存，那么您需要在过渡期间保存数据的东西。一种解决方案是构建一个进程来进行分配，然后从 ipc 创建 cudf 列。我不确定如何在 python 中执行此操作。在 C++ 中，它非常简单。

类似的东西

//In the code handling your allocations
gdf_column col;

cudaMemHandle_t handle_data, handle_valid;
cudaIpcGetMemHandle(&handle,col.data);
cudaIpcGetMemHandle(&valid,col.valid);



//In the code consuming it
gdf_column col;

//deserialize these by reading from a file or however you want to make this 
//binary data avaialable
cudaMemHandle_t handle_data, handle_valid;

cudaIpcOpenMemHandle ( (void**) &col.data, cudaIpcMemHandle_t handle, cudaIpcMemLazyEnablePeerAccess );
cudaIpcOpenMemHandle ( (void**) &col.valid, cudaIpcMemHandle_t handle, cudaIpcMemLazyEnablePeerAccess );

还有来自 RAPIDs 贡献者的第三方解决方案，如 BlazingSQL，它在 python 中提供此功能，并为 cudfs 提供 SQL 接口(interface)。

在这里你会做类似的事情

#run this code in your service to basically select your entire table and get it
#as a cudf
from blazingsql import BlazingContext
import pickle
bc = BlazingContext()
bc.create_table('performance', some_valid_gdf) #you can also put a file or list of files here
result= bc.sql('SELECT * FROM main.performance', ['performance'])
with open('context.pkl', 'wb') as output:
    pickle.dump(bc, output, pickle.HIGHEST_PROTOCOL)
with open('result.pkl', 'wb') as output:
    pickle.dump(result, output, pickle.HIGHEST_PROTOCOL)


#the following code can be run on another process as long as result
# contains the same information from above, its existence is managed by blazingSQL 
from blazingsql import BlazingContext
import pickle
with open('context.pkl', 'rb') as input:
  bc = pickle.load(input)
with open('result.pkl', 'rb') as input:
  result = pickle.load(input)

#Get result object
result = result.get()
#Create GDF from result object
result_gdf = result.columns

免责声明，我为 Blazing 工作。