无法使用案例类将 rdd 转换为 Dataframe

Question

我正在尝试使用 Case Class 将 rdd 转换为 DataFrame，如下所示

1.)从“id,name,country”的文本文件中获取数据，由“,”分隔，但没有标题

val x = sc.textFile("file:///home/hdadmin/records.txt")

2.)创建一个带有如下标题定义的案例类“rec”：

case class rec(id:Int, name:String, country:String)

3.) 现在我定义转换

val y = x.map(x=>x.split(",")).map(x=>rec(x(0).toInt,x(1),x(2)))

4.) 然后我导入了隐式库

import spark.implicits._

5.) 使用 toDF 方法将 rdd 转换为数据帧：

val z = y.toDF()

6.) 现在，当我尝试使用以下命令获取记录时：

z.select("name").show()

我收到以下错误：

17/05/19 12:50:14 错误 LiveListenerBus：SparkListenerBus 已经 停了！删除事件 SparkListenerSQLExecutionStart(9,show at ：49，org.apache.spark.sql.Dataset.show（Dataset.scala:495） $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw $$iw.(:49) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw .(:54) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(: 56) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:58) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:60) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:62) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:64) $line105.$read$$iw$$ iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:66) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:68) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:70) $line105.$read$$iw$$iw$$iw$$iw$$iw$$iw.(:72) $line105.$read$$iw$$iw$$iw$$iw$$iw.(:74) $line105.$read$$iw$$iw$$iw$$iw.(:76) $line105.$read$$iw$$iw$$iw.(:78) $line105.$read$$iw$$iw.(:80) $line105.$read$$iw.(:82) $line105.$read.(:84) $line105.$read$.(:88) $line105.$read$.(),== 解析的逻辑计划 == 全球限制 21 +- LocalLimit 21 +- 项目 [名称#91] +- 逻辑RDD [id#90, name#91, country#92]

== 分析的逻辑计划 == 名称：字符串 GlobalLimit 21 +- LocalLimit 21 +- 项目 [名称#91] +- 逻辑RDD [id#90, name#91, country#92]

== 优化逻辑计划 == GlobalLimit 21 +- LocalLimit 21 +- 项目 [名称#91] +- 逻辑RDD [id#90, name#91, country#92]

== 物理计划 == CollectLimit 21 +- *Project [name#91] +- 扫描 ExistingRDD[id#90,name#91,country#92],org.apache.spark.sql.execution.SparkPlanInfo@b807ee,1495223414636) 17/05/19 12:50:14 错误 LiveListenerBus：SparkListenerBus 已经 停了！删除事件 SparkListenerSQLExecutionEnd(9,1495223414734) java.lang.IllegalStateException：SparkContext 已在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1863) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1884) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1897) 在 org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:347) 在 org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:39) 在 org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$execute$1$1.apply(Dataset.scala:2183) 在 org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57) 在 org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2532) 在 org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$execute$1(Dataset.scala:2182) 在 org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collect(Dataset.scala:2189) 在 org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:1925) 在 org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:1924) 在 org.apache.spark.sql.Dataset.withTypedCallback(Dataset.scala:2562) 在 org.apache.spark.sql.Dataset.head(Dataset.scala:1924) 在 org.apache.spark.sql.Dataset.take(Dataset.scala:2139) 在 org.apache.spark.sql.Dataset.showString(Dataset.scala:239) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:526) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:486) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:495) ... 56 省略

问题可能出在哪里？

Answer 1

在为几个文本文件尝试了相同的代码之后，我实际上更正了文本文件中的文本格式以防任何差异。

下面代码中的列分隔符是“，”，在我仔细扫描后，它在文本文件中的 1 个位置丢失。

val y = x.map(x=>x.split(",")).map(x=>rec(x(0).toInt,x(1),x(2)))

代码运行良好，并在更改后以结构化表格格式显示结果。

因此，重要的是要注意内部给出的分隔符​​（“，”，“\t”，“|”）

x.split("")

应该与源文件和整个源文件中的相同。