SchemaRDD在Spark SQL中已经被我们使用到,这篇文章简单地介绍一下如果将标准的RDD(org.apache.spark.rdd.RDD)转换成SchemaRDD,并进行SQL相关的操作。
scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc) sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@6edd421f scala> case class Person(name: String, age:Int) defined class Person scala> var people = sc.textFile("/home/iteblog/person.txt") | .map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)) people: org.apache.spark.rdd.RDD[Person] = MappedRDD[3] at map at <console>:14 scala> people.registerTempTable("people") <console>:17: error: value registerTempTable is not a member of org.apache.spark.rdd.RDD[Person] people.registerTempTable("people") ^
这是因为people是普通的RDD,而registerTempTable函数不属于RDD类,只有通过SchemaRDD的实例才可以调用,所以这么调用会出现错误,解决办法有两个:
(1)registerTempTable函数是SQLContext类中的,所以我们可以将people转换成SchemaRDD,如下:
/** * User: 过往记忆 * Date: 14-12-16 * Time: 下午10:16 * bolg: * 本文地址:/archives/1224 * 过往记忆博客,专注于hadoop、hive、spark、shark、flume的技术博客,大量的干货 * 过往记忆博客微信公共帐号:iteblog_hadoop */ scala> val peopleSchema = sqlContext.createSchemaRDD(people) peopleSchema: org.apache.spark.sql.SchemaRDD = SchemaRDD[29] at RDD at SchemaRDD.scala:103 == Query Plan == == Physical Plan == ExistingRdd [name#4,age#5], MapPartitionsRDD[28] at mapPartitions at basicOperators.scala:217 scala> peopleSchema.registerTempTable("people") warning: there were 1 deprecation warning(s); re-run with -deprecation for details
这么调用就可以将people转成SchemaRDD。
(2)、上面的方法是通过显示地调用sqlContext.createSchemaRDD将普通的RDD转成SchemaRDD。其实我们还可以通过Scala的隐式语法来进行转换。我们先来看看createSchemaRDD函数的定义
/** * Creates a SchemaRDD from an RDD of case classes. * * @group userf */ implicit def createSchemaRDD[A <: Product: TypeTag](rdd: RDD[A]) = { SparkPlan.currentContext.set(self) new SchemaRDD(this, SparkLogicalPlan(ExistingRdd.fromProductRdd(rdd))(self)) }
在定义createSchemaRDD的时候用到了implicit 关键字,所以我们在使用的时候可以通过下面语句使用
scala> import sqlContext.createSchemaRDD import sqlContext.createSchemaRDD scala> people.registerAsTable("people") warning: there were 1 deprecation warning(s); re-run with -deprecation for details
这样就隐身地将people转换成SchemaRDD了。这是因为Spark可以隐式地将包含case class的RDD转换成SchemaRDD。
关于什么是SchemaRDD,官方文档将的很详细:
An RDD of [[Row]] objects that has an associated schema. In addition to standard RDD functions, SchemaRDDs can be used in relational queries。也就是包含了Row对象以及模式的RDD。它继承自标准的RDD类,所以拥有标准RDD类的所有方法;并且可以用于关系性数据库的查询在中。
本博客文章除特别声明,全部都是原创!An RDD of [[Row]] objects that has an associated schema. In addition to standard RDD functions, SchemaRDDs can be used in relational queries。也就是包含了Row对象以及模式的RDD。它继承自标准的RDD类,所以拥有标准RDD类的所有方法;并且可以用于关系性数据库的查询在中。
原创文章版权归过往记忆大数据(过往记忆)所有,未经许可不得转载。
本文链接: 【Spark: SchemaRDD隐式转换】(https://www.iteblog.com/archives/1224.html)