以编程方式将列名添加到从 RDD 构建的 Spark DataFrame

Question

我有一个没有标题的管道分隔文本文件，并且行有不同数量的列（有些行是类型 A 有 400 列，其他类型是 B 有 200，所以我需要将它们分开第一）：

val textFileRaw = sc.textFile("./data.txt")
val textFile = textFileRaw.map(line => line.split("\\|", -1))
val dataA = textFile.filter(line => line(0) == "A")
val dataB = textFile.filter(line => line(0) == "B")

现在我想将这些 RDD 转换为 Spark DataFrame，但拆分返回的是单个数组，而不是 400 或 200 个不同的值。这会导致以下错误：

# ANames are my column names, length=400
val ANames = Array("Row ID", "City", "State", ...)
val dataADF = dataA.toDF(ANames: _*)

Name: java.lang.IllegalArgumentException
Message: requirement failed: The number of columns doesn't match.
Old column names (1): value
New column names (400): Row ID, City, State ...

This question 面临同样的问题，但所有答案都建议手动指定从数组到元组的映射，这在数百列的情况下并不是很好。

如果我使用Spark's csv loader，我想我可以让它工作，但这不适用于我的数据，因为行具有不同数量的字段（它不是真正的 csv 文件）。一种解决方法是首先拆分文件，编写格式良好的 csv 新文件，然后使用 csv 加载器，但如果可能的话，我想避免这种情况。如何将这些 RDD 转换为具有命名列的 DataFrame？

Answer 1

您应该创建一个架构并使用 SQLContext.createDataFrame api 作为

val dataA = textFile.filter(line => line(0) == "A")
val ANames = Array("Row ID", "City", "State", "kjl")
val dataADF = sqlContext.createDataFrame(dataA.map(Row.fromSeq(_)), StructType(ANames.map(StructField(_, StringType, true))))

它应该可以工作。但请注意，我已将所有 数据类型 用作StringType()。您可以根据需要进行更改。