如何用Spark CSV Reader读取预过滤的RDD[String]以降低HDFS IO？

我完全懂你的需求——不想为了拆分不同Schema的记录额外写入HDFS，徒增IO开销，毕竟已经通过RDD过滤出了PHM开头的行，就想直接把这些内存里的数据转换成结构化的DataFrame对吧？

你之前尝试用spark.read.option("schema","phcschema").csv(fileRDD.toDS())的思路是对的，但细节上有两个关键问题没处理好：一是Schema的指定方式不对，二是没匹配你数据的分隔符（你的数据用的是竖线|，而CSV Reader默认用逗号）。下面给你可行的完整方案：

步骤1：定义PHM记录对应的Schema #

因为PHM开头的记录结构固定（5列），咱们直接手动定义StructType，这样比自动推断更可靠：

import org.apache.spark.sql.types._

// 对应PHM|MERC|PHARMA|BLUEDRUG|50的结构
val phmSchema = StructType(Array(
  StructField("record_type", StringType, nullable = false),
  StructField("vendor", StringType, nullable = false),
  StructField("category", StringType, nullable = false),
  StructField("drug_name", StringType, nullable = false),
  StructField("amount", IntegerType, nullable = false)
))

步骤2：用CSV Reader直接读取过滤后的Dataset[String] #

现在把你转换好的Dataset[String]传给CSV Reader，同时配置正确的选项：

// 这里的fileRDD是你已经过滤好的RDD[String]
val phmDF = spark.read
  .option("delimiter", "|")  // 必须指定分隔符为竖线，否则默认逗号会解析错误
  .option("header", "false") // 你的数据没有表头行，一定要设为false
  .schema(phmSchema)         // 传入咱们定义好的Schema
  .csv(fileRDD.toDS())       // 直接读取内存中的Dataset，无需落地文件

为什么你之前的写法不生效？ #

• Schema参数错误：option("schema","phcschema")是错误的用法，schema方法需要接收StructType对象，而不是字符串形式的Schema名称。如果你的phcschema是已经注册好的表，那可以用.schema(spark.table("phcschema").schema)来引用，但手动定义更直接。
• 未指定分隔符：你的数据用|分隔，CSV Reader默认用逗号，不指定的话会把整行当成一列，完全不符合预期。

这样处理后，你就能直接得到结构化的PHM数据DataFrame，全程不需要写入HDFS，完美避免了额外的IO开销。同理，CLM开头的记录也可以用同样的方式：先过滤出CLM的RDD，定义对应的Schema，再用CSV Reader读取即可。

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