如何将C++生成的DataFrame零拷贝传递给Rust的Apache Datafusion

核心思路 #

依托Apache Arrow跨语言一致的内存布局，我们可以通过共享内存或MPI内存窗口实现零拷贝进程间通信——让C++生成Arrow的RecordBatch/Table后，直接将内存区域暴露给Rust进程，Rust端无需序列化/持久化，直接映射为Arrow Rust数据结构，再交给DataFusion处理。

具体实现步骤 #

1. C++端生成Arrow数据并共享内存 #

使用C++版Apache Arrow库生成数据，再通过共享内存（或MPI内存窗口）将数据内存暴露给Rust进程：

#include <arrow/api.h>
#include <arrow/ipc/writer.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <unistd.h>

// 生成测试用Arrow RecordBatch
std::shared_ptr<arrow::RecordBatch> create_test_batch() {
  auto schema = arrow::schema({
    arrow::field("a", arrow::int32()),
    arrow::field("b", arrow::int32())
  });

  auto a_array = arrow::MakeArrayFromValues<int32_t>({1, 2, 3, 4, 5});
  auto b_array = arrow::MakeArrayFromValues<int32_t>({2, 3, 1, 5, 4});

  return arrow::RecordBatch::Make(schema, 5, {a_array, b_array});
}

// 将RecordBatch写入共享内存
void write_to_shared_memory(const std::shared_ptr<arrow::RecordBatch>& batch) {
  // 创建并配置共享内存
  int fd = shm_open("/arrow_shm", O_CREAT | O_RDWR, 0666);
  ftruncate(fd, batch->serialized_size());
  void* shm_ptr = mmap(nullptr, batch->serialized_size(), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);

  // 将RecordBatch序列化为Arrow IPC格式写入共享内存
  auto out_stream = arrow::io::BufferOutputStream::Create((uint8_t*)shm_ptr, batch->serialized_size()).ValueOrDie();
  auto writer = arrow::ipc::RecordBatchWriter::Open(out_stream, batch->schema()).ValueOrDie();
  writer->WriteRecordBatch(*batch).ValueOrDie();
  writer->Close().ValueOrDie();

  // 同步通知Rust进程数据就绪（可通过信号、管道或MPI消息实现）
}

2. Rust端读取共享内存并转为DataFusion DataFrame #

在Rust中直接映射共享内存，读取Arrow数据后转为DataFusion的DataFrame处理：

use arrow::ipc::reader::StreamReader;
use arrow::io::buffer::BufferReader;
use datafusion::prelude::*;
use std::os::unix::io::AsRawFd;
use std::fs::OpenOptions;
use memmap2::Mmap;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    // 打开共享内存并映射
    let file = OpenOptions::new()
        .read(true)
        .open("/arrow_shm")?;
    let mmap = unsafe { Mmap::map(&file)? };
    let reader = BufferReader::new(&mmap);
    
    // 读取Arrow IPC格式的RecordBatch
    let mut stream_reader = StreamReader::try_new(reader, None)?;
    let batch = stream_reader.next().unwrap()?;
    
    // 转为DataFusion DataFrame并执行处理逻辑
    let ctx = SessionContext::new();
    let df = ctx.read_batch(batch)?;
    
    let df = df.filter(col("a").lt_eq(col("b")))?
               .aggregate(vec![col("a")], vec![min(col("b"))])?
               .limit(0, Some(100))?;
    
    df.show().await?;
    
    Ok(())
}

3. MPI场景下的零拷贝优化 #

如果是MPI集群环境，可直接使用**MPI内存窗口（MPI_Win）**替代共享内存：

• C++端通过MPI_Win_create将Arrow数据的内存区域注册为共享窗口
• Rust端借助rsmpi crate，通过MPI_Win_attach映射该内存区域，直接读取Arrow数据结构
  这种方式完全跳过序列化步骤，实现真正的零拷贝进程间通信。

关键注意事项 #

• 确保C++与Rust使用的Apache Arrow版本完全一致，避免内存布局不兼容
• 通过同步机制（如信号量、MPI消息）协调内存生命周期，防止Rust读取时C++释放内存
• 处理大型数据集时，可分批次传递RecordBatch，避免内存占用过高

内容的提问来源于stack exchange，提问作者cpchung