例外:
val people = sc.textFile("resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
value toDF is not a member of org.apache.spark.rdd.RDD[Person]
这是
TestApp.scala
文件:
package main.scala
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
case class Record1(k: Int, v: String)
object RDDToDataFramesWithCaseClasses {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("Simple Spark SQL Application With RDD To DF")
// sc is an existing SparkContext.
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
// this is used to implicitly convert an RDD to a DataFrame.
import sqlContext.implicits._
// Define the schema using a case class.
// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,package main.scala
还有
TestApp.scala
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
case class Record1(k: Int, v: String)
object RDDToDataFramesWithCaseClasses {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("RDD To DF")
// sc is an existing SparkContext.
// you can use custom classes that implement the Product interface.
case class Person(name: String, age: Int)
// Create an RDD of Person objects and register it as a table.
val people = sc.textFile("resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
people.registerTempTable("people")
// SQL statements can be run by using the sql methods provided by sqlContext.
val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")
// The results of SQL queries are DataFrames and support all the normal RDD operations.
// The columns of a row in the result can be accessed by field index:
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)
// or by field name:
teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)
// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)
// Map("name" -> "Justin", "age" -> 19)
}
}
和SBT 文件
name := "SparkScalaRDBMS"
version := "1.0"
scalaVersion := "2.11.7"
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.5.1"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "1.5.1"
现在我找到了原因,你应该在对象中定义案例类,并在主函数之外。 看这里
好的,我终于解决了这个问题。需要做 2 件事:
导入隐式:请注意,只有在创建 org.apache.spark.sql.SQLContext 实例后才应执行此操作。应该写成:
val sqlContext= new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._
将案例类移到方法之外:案例类(用于定义 DataFrame 的架构)应该在需要它的方法之外定义。您可以在这里阅读更多相关信息:https://issues.scala-lang.org/browse/SI-6649
在 Spark 2 中,您需要从 SparkSession 导入隐式:
val spark = SparkSession.builder().appName(appName).getOrCreate()
import spark.implicits._
有关创建 SparkSession 时的更多选项,请参阅 Spark 文档。
您的代码有两个问题
对于 Spark V 1.0,您需要导入 import sqlContext.implicits._ ;如果您使用 Spark V 2.0 或更高版本,则需要导入 Spark.implicits._
第二种情况 class Record1(k: Int, v: String) 需要位于 main 函数内部但位于 def main(args: Array[String]) { val conf = new SparkConf().setAppName("RDD 转 DF") …
}
scala> case class Employee(id: Int, name: String, age: Int)
defined class Employee
scala> val sqlContext= new org.apache.spark.sql.SQLContext(sc)
warning: there was one deprecation warning; re-run with -deprecation for details
sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@1f94e3a
scala> import sqlContext.implicits._
import sqlContext.implicits._
scala> var empl1= empl.map(_.split(",")).map(e=>Employee(e(0).trim.toInt,e(1),e(2).trim.toInt)).toDF
empl1: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
scala> val allrecords = sqlContext.sql("SELECT * FROM employee")
allrecords: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
scala> allrecords.show();
+----+--------+---+
| id| name|age|
+----+--------+---+
|1201| satish| 25|
|1202| krishna| 28|
|1203| amith| 39|
|1204| javed| 23|
|1205| prudvi| 23|
+----+--------+---+
在 Scala 工作表中运行 Spark 时遇到此问题。基本上,由于工作表的性质,在这些情况下您不能使用 toDF()。相反,请使用
spark.createDataFrame
。