从来自 java 中 InputStream 的字符串创建 Spark RDD 或数据帧

Posted 2023-04-17

【中文标题】从来自 java 中 InputStream 的字符串创建 Spark RDD 或数据帧

【英文标题】：Create Spark RDD or dataframe from Strings coming from InputStream in java

【发布时间】：2017-10-04 12:54:30

【问题描述】：

我在 java 中有一个字符串流。那来自其他机器上的 csv 文件。我正在创建一个 InputStream 并从 java 中的 BufferedReader 中逐行读取 csv 文件，如下所示。

        //call a method that returns inputStream 


        InputStream stream = getInputStreamOfFile();

        BufferedReader lineStream = new BufferedReader(new InputStreamReader(stream));

        while ((inputLine = lineStream.readLine()) != null) 
            System.out.println("******************new Line***********");
            System.out.println(inputLine);
        
        lineStream.close();
        stream.close();

现在，我想从中创建一个 spark RDD 或 DataFrame。

一个解决方案是，我不断在每一行创建新的 RDD 并维护 globle RDD 并继续进行 RDD 的联合。 还有其他解决方案吗？

注意：此文件不在同一台机器上。它来自一些远程存储。我确实有文件的 HTTP URL。

【问题讨论】：

继续追加到缓冲区，然后在 csv 完成流式传输后创建 RDD？或者将 csv 流式传输到您可以访问的文件系统，然后从该文件创建一个数据框？

文件很大。我无法在本地保存。

为什么不能用 sparkContext.textFile("filepath") 读取文件本身

如果可以在这里下载是给定的解决方案***.com/questions/29741082/…

@KrunalParmar 你找到解决方案了吗？

【参考方案1】：

如果 inputStream 的内容适合内存，我们可以使用以下内容：

private static List<String> displayTextInputStream(InputStream input) throws IOException 
    // Read the text input stream one line at a time and display each line.
    BufferedReader reader = new BufferedReader(new InputStreamReader(input));
    String line = null;
    List<String> result = new ArrayList<String>();
    while ((line = reader.readLine()) != null) 
        result.add(line);
    
    return result;

现在我们可以将List<String> 转换为对应的RDD。

S3Object fullObject = s3Client.getObject(new GetObjectRequest("bigdataanalytics", each.getKey()));
                            List<String> listVals = displayTextInputStream(fullObject.getObjectContent());
                            JavaRDD<String> s3Rdd = sc.parallelize(listVals);