RDD 字符串到 Spark csv 阅读器答案

【问题标题】：RDD String to Spark csv ReaderRDD 字符串到 Spark csv 阅读器
【发布时间】：2019-05-31 02:15:36
【问题描述】：

我想使用 spark CSV 阅读器阅读 RDD[String]。我这样做的原因是，我需要在使用 CSV 阅读器之前过滤一些记录。

val fileRDD: RDD[String] = spark.sparkContext.textFile("file")

我需要使用 spark CSV 阅读器读取 fileRDD。我不希望提交该文件，因为它会增加 HDFS 的 IO。我查看了我们在 spark CSV 中的选项，但没有找到任何选项。

spark.read.csv(file)

样本数据

PHM|MERC|PHARMA|BLUEDRUG|50
CLM|BSH|CLAIM|VISIT|HSA|EMPLOYER|PAID|250
PHM|GSK|PHARMA|PARAC|70
CLM|UHC|CLAIM|VISIT|HSA|PERSONAL|PAID|72

如您所见，所有以 PHM 开头的记录具有不同的列数，而 clm 具有不同的列数。这就是我过滤然后应用模式的原因。 PHM 和 CLM 记录具有不同的模式。

val fileRDD: RDD[String] = spark.sparkContext.textFile("file").filter(_.startWith("PHM"))


 spark.read.option(schema,"phcschema").csv(fileRDD.toDS())

【问题讨论】：

我没有看到读取文本文件并再次填充到 CSV 阅读器并制作数据框的真正原因。使用它自己的数据框，您可以进行这些检查和过滤。你正在做的是一次往返。 AFAIK 我觉得没必要
此外，RDD 使用 java 序列化，而数据帧使用 tungstun 二进制格式，比 RDD 更高效
@RamGhadiyaram 该文件具有不同的架构，我必须在应用架构之前将其过滤掉。那就是为了这样做的原因。我实际上正在寻找带有 csv 阅读器的过滤器，它不存在
是的，在这种情况下，您可以使用具有不同模式的示例记录来提及该文件。您可能会得到回答者的优雅解决方案。
@RamGhadiyaram 你能详细说明吗，我没明白。

标签： apache-spark apache-spark-sql

【解决方案1】：

从 Spark 2.2 开始，方法“.csv”可以读取字符串数据集。可以这样实现：

val rdd: RDD[String] = spark.sparkContext.textFile("csv.txt")
// ... do filtering
spark.read.csv(rdd.toDS())

【讨论】：