在 Apache Spark 中读取具有可变部分内容的多个部分标题的文件答案

【问题标题】：Reading a file with multiple section headers in Apache Spark with variable section content在 Apache Spark 中读取具有可变部分内容的多个部分标题的文件
【发布时间】：2018-10-08 09:01:37
【问题描述】：

是否可以使用 Spark API 读取包含具有不同标题的多个部分的大型 CSV 文件？文件结构如下

BatchCode#1
Name,Surname,Address
AA1,BBB,CCC
AA2,BBB,CCC
AA3,BBB,CCC

BatchCode#2
Name,Surname,Address,Phone
XY1,BBB,CCC,DDD
XY2,BBB,CCC,DDD
XY3,BBB,CCC,DDD

在阅读记录时，我们需要注意标题以及各部分之间的文件格式可能不同。 BatchCode 信息需要从标头中提取，并且应该是该部分中每条记录的一部分 - 例如，第 1 行的数据应解析为：

Name: AAA1
Surname: BBB
Address:CCC
BatchCode:1

我想到了以下选项，但我不完全确定它是否会造成重大问题：

使用 wholeTextFile 读取文件。这将使用单个线程读取文件，但会将整个文件加载到内存中，并可能导致大文件出现内存问题。
使用 sc.textFile 上的 coalesce(1) 强制 Spark 在单个线程中读取文件。我不确定订单是否始终得到保证。一旦我们将文件作为 RDD 获取，我们将在读取文件时缓存标题行并将它们与相应的数据记录合并。

即使上述方法有效，它们是否有效？什么是最有效的方法？

【问题讨论】：

coalesce() 不会帮助读取文件，只会写入文件。 wholeTextFile 对你来说真的是个问题吗？你的数据集有多大？我认为您最好的解决方案是将文件预处理为多个文件，每个文件都有自己的 csv 模式（除非这会创建大量小文件，在这种情况下，您最好使用单个流文件解析器）
看我的回答。我可以分享一些示例代码。
wholeTextFiles 对我们不起作用，因为我们将数据设置为 20 GB 作为输入。预处理是一种选择，但我试图弄清楚我们是否可以使用 Spark 内联。我很好奇——我在合并文档中遗漏了什么，它只是说——返回一个新的 RDD，它被缩减为 numPartitions 分区。
使用 binaryFiles 选项读取此类文件怎么样？
一个新的RDD类似于写。

标签： java scala csv apache-spark

【解决方案1】：

我只为更复杂的此类用例编写了 Scala 程序，从而保证了顺序性。否则太难了。如果文件首先来自 xls 或 xlsx，则通过 csvkit 处理这些文件。

【讨论】：

感谢您的回复。在scala中读取文件的方法是什么？预处理器？或者你在使用 SparkContext.someMethod
我正在使用仅 SCALA 的程序 - 我们查看了检查分区 idex = 0 等的内容，但这太难了。此外，架构师们一致认为，为更小的输入而旋转 SPARK 工作并不是真正的成功者。问题是，在你的情况下什么是大的？很高兴分享一个被认为是高级的示例，但您需要成为这种方法的优秀程序员。
可以并行化
我想坚持使用 Spark 以重用所有功能。它允许我们使用 HDFS 在不同的数据中心和区域之间进行水平扩展和复制。
这很好，但你使用 scala 或 pysrk 与 spark 或 java。我成功地做到了。成功。这个问题是一个开放的问题，这是我的答案。

【解决方案2】：

以下程序适合我：

JavaPairRDD<String, PortableDataStream> binaryFiles = sc.binaryFiles(file);

PortableRecordReader reader = new PortableRecordReader();
JavaPairRDD<String, Record> fileAndLines = binaryFiles.flatMapValues(reader);

PortableRecordReader 打开 DataInputStream 并将其转换为 InputStreamReader，然后使用 CSV 解析器将行转换为 Record 对象中的预期输出，并合并标题。

【讨论】：

我建议通过提供 PortableRecordReader 代码来改进这个答案。