如何用Scala将CSV字符串及指定表头的RDD转为Spark DataFrame？

Hey there! Let's walk through how to turn your string RDD into a properly structured DataFrame using that header variable in Spark 2.2. I'll show you two reliable approaches that work well for this scenario.

Approach 1: Manual Schema + Row Conversion #

This method gives you full control over each column's data type, which is great if you need precision (like converting age to an integer instead of leaving it as a string).

First, import the necessary Spark SQL classes:

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

Then follow these steps with code:

// Your existing RDD and header variable
val dataRDD = sc.parallelize(Seq("Mike,2222-003330,NY,34", "Kate,3333-544444,LA,32", "Abby,4444-234324,MA,56"))
val header = "name,account,state,age"

// Split header into individual column names
val columnNames = header.split(",")

// Define schema with correct data types (age as integer, others as string)
val schema = StructType(
  columnNames.zipWithIndex.map { case (name, idx) =>
    val dataType = if (idx == 3) IntegerType else StringType
    StructField(name, dataType, nullable = true)
  }
)

// Convert each string row to a Row object with typed values
val rowRDD = dataRDD.map { line =>
  val parts = line.split(",")
  Row(parts(0), parts(1), parts(2), parts(3).toInt)
}

// Create the final DataFrame
val df = spark.createDataFrame(rowRDD, schema)

// Verify the result
df.show()

Approach 2: Treat RDD as an In-Memory CSV Source #

If you prefer a more concise workflow, you can combine the header with your data and use Spark's built-in CSV reader. This is great for quick setup, and Spark can even infer data types automatically.

// Your existing RDD and header variable
val dataRDD = sc.parallelize(Seq("Mike,2222-003330,NY,34", "Kate,3333-544444,LA,32", "Abby,4444-234324,MA,56"))
val header = "name,account,state,age"

// Combine header with data and convert to a Dataset[String]
val csvDataset = spark.createDataset(header +: dataRDD.collect())

// Read as CSV with header enabled and schema inference
val df = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv(csvDataset)

// Check the output
df.show()

Note: If you need strict control over data types instead of relying on inference, you can pass the same schema from Approach 1 using .schema(schema) in the reader chain.

Either method will give you a fully structured DataFrame with the columns name, account, state, and age—pick the one that fits your use case best!

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