如何将Spark DataFrame转换为JSON格式HTTP响应？附Parquet处理场景

一、Spark下的实现方式 #

1. 转换为JSON字符串 #

你不需要用dataset.write写入文件，Spark DataFrame自带toJSON() API可以直接将每行数据转为JSON字符串，再拼接成标准JSON数组：

// Scala 示例
import org.apache.spark.sql.DataFrame

def dfToJsonArray(df: DataFrame): String = {
  val jsonRows = df.toJSON.collect() // 将每行转为JSON字符串并拉取到Driver节点
  "[" + jsonRows.mkString(",") + "]"
}

注意：collect()会把全量数据加载到Driver内存，50万行数据只要Driver内存预留1GB以上就能稳定处理。如果担心内存压力，也可以分批拉取拼接，但这个规模下直接处理通常足够高效。

2. 直接流式传输到HTTP输出流 #

Spark可以结合HTTP服务的输出流实现流式响应，避免全量加载数据到内存：

// Scala + HTTP服务（如Play Framework）示例
def streamDfToHttp(df: DataFrame, outputStream: OutputStream): Unit = {
  val writer = new PrintWriter(outputStream)
  writer.write("[")
  val jsonIter = df.toJSON.toLocalIterator()
  var isFirstRow = true
  
  while (jsonIter.hasNext) {
    if (!isFirstRow) writer.write(",")
    writer.write(jsonIter.next())
    isFirstRow = false
    writer.flush() // 实时将数据刷入HTTP输出流
  }
  
  writer.write("]")
  writer.flush()
}

这种方式通过迭代器逐行输出，内存占用更低，适合中型数据的低延迟响应。

二、是否应该使用Spark？ #

Spark的核心优势是处理TB级以上的分布式大数据，但你的场景是100行到50万行的中小数据，用Spark存在明显劣势：

• 每次HTTP请求初始化SparkContext/SparkSession会带来几秒到十几秒的延迟，无法满足API服务的低延迟要求；
• 分布式框架的资源开销远大于单进程框架，属于“杀鸡用牛刀”。

因此如果你的服务对延迟敏感，或数据规模不会突破百万行级别，不建议使用Spark，单进程的Arrow、Pandas等框架是更优选择。

三、Arrow框架的替代方案 #

Arrow专为列式数据处理设计，读取Parquet性能优异，支持流式JSON输出，完美适配你的HTTP API场景：

1. 转换为JSON字符串（Java示例） #

import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.parquet.ParquetReader;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.JsonWriter;
import java.io.StringWriter;
import java.io.IOException;

public String parquetToJson(String parquetPath) throws IOException {
  try (RootAllocator allocator = new RootAllocator();
       ParquetReader<VectorSchemaRoot> reader = ParquetReader.open(allocator, parquetPath);
       StringWriter stringWriter = new StringWriter()) {
    VectorSchemaRoot root;
    JsonWriter jsonWriter = new JsonWriter(stringWriter, reader.getVectorSchemaRoot().getSchema());
    
    while ((root = reader.read()) != null) {
      jsonWriter.write(root);
      root.close();
    }
    jsonWriter.close();
    return stringWriter.toString();
  }
}

2. 直接流式传输到HTTP输出流 #

// Java + Servlet示例
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.parquet.ParquetReader;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.JsonWriter;
import javax.servlet.http.HttpServletResponse;
import java.io.OutputStreamWriter;
import java.io.IOException;

public void streamParquetToHttp(HttpServletResponse response, String parquetPath) throws IOException {
  response.setContentType("application/json");
  try (RootAllocator allocator = new RootAllocator();
       ParquetReader<VectorSchemaRoot> reader = ParquetReader.open(allocator, parquetPath);
       OutputStreamWriter writer = new OutputStreamWriter(response.getOutputStream())) {
    VectorSchemaRoot root;
    JsonWriter jsonWriter = new JsonWriter(writer, reader.getVectorSchemaRoot().getSchema());
    
    while ((root = reader.read()) != null) {
      jsonWriter.write(root);
      root.close();
      response.getOutputStream().flush();
    }
    jsonWriter.close();
  }
}

Arrow的优势是低延迟、内存高效，无需分布式集群，单进程即可快速处理中小规模Parquet数据，完全匹配HTTP API服务的需求。

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