如何在Spark中读取InputStream中的数据？

Spark本身没有提供直接读取InputStream的API，因为它的分布式模型更适配文件系统、数据库这类可分布式访问的数据源。不过可以通过以下两种常见方式实现需求：

方法一：将InputStream写入临时文件后读取 #

如果数据量不大，可以先把InputStream的内容写入本地或分布式文件系统的临时文件，再用Spark常规的文件读取API加载数据。

Scala 示例 #

import org.apache.spark.sql.SparkSession
import java.io.{FileOutputStream, InputStream}
import java.nio.file.{Files, Paths}

val spark = SparkSession.builder().appName("ReadInputStream").master("local[*]").getOrCreate()

// 假设你已经有一个InputStream实例
val inputStream: InputStream = ... 

// 创建临时文件
val tempPath = Files.createTempFile("spark_input", ".tmp").toString
val fos = new FileOutputStream(tempPath)
inputStream.transferTo(fos)
fos.close()
inputStream.close()

// 用Spark读取临时文件
val df = spark.read.text(tempPath)
df.show()

// 可选：删除临时文件
Files.delete(Paths.get(tempPath))

Java 示例 #

import org.apache.spark.sql.SparkSession;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Paths;

public class ReadInputStream {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("ReadInputStream").master("local[*]").getOrCreate();
        
        // 假设已获取InputStream实例
        InputStream inputStream = ...;
        
        try {
            // 创建临时文件
            String tempPath = Files.createTempFile("spark_input", ".tmp").toString();
            FileOutputStream fos = new FileOutputStream(tempPath);
            inputStream.transferTo(fos);
            
            // 读取数据
            spark.read().text(tempPath).show();
            
            // 删除临时文件
            Files.delete(Paths.get(tempPath));
            
            fos.close();
            inputStream.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

方法二：将InputStream内容并行化为RDD #

如果数据量较小（能完全加载到Driver内存），可以先把InputStream的内容读取为内存中的集合，再通过sparkContext.parallelize生成RDD，进而转换为DataFrame。

Scala 示例 #

import org.apache.spark.sql.SparkSession
import java.io.{BufferedReader, InputStreamReader}
import java.io.InputStream

val spark = SparkSession.builder().appName("ReadInputStream").master("local[*]").getOrCreate()
val sc = spark.sparkContext

// 假设已有InputStream实例
val inputStream: InputStream = ...

// 读取InputStream为字符串列表
val reader = new BufferedReader(new InputStreamReader(inputStream))
val lines = Iterator.continually(reader.readLine()).takeWhile(_ != null).toList
reader.close()
inputStream.close()

// 并行化为RDD并转为DataFrame
val rdd = sc.parallelize(lines)
val df = spark.createDataFrame(rdd.map(Tuple1.apply)).toDF("content")
df.show()

注意事项 #

• 数据量限制：方法二仅适用于数据能完全放入Driver内存的场景，数据量过大时会触发OOM；方法一的临时文件如果写入本地，集群模式下需确保所有Executor都能访问到（建议写入HDFS、S3这类分布式存储）。
• InputStream的单向性：InputStream只能读取一次，处理时要避免重复调用读取操作。
• 资源清理：临时文件使用后记得删除，避免占用存储空间。

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