共计 1768 个字符,预计需要花费 5 分钟才能阅读完成。
今天就跟大家聊聊有关怎么实现 Spark SchemaRDD 隐式转换,可能很多人都不太了解,为了让大家更加了解,丸趣 TV 小编给大家总结了以下内容,希望大家根据这篇文章可以有所收获。
SchemaRDD 在 Spark SQL 中已经被我们使用到,这篇文章简单地介绍一下如果将标准的 RDD(org.apache.spark.rdd.RDD) 转换成 SchemaRDD,并进行 SQL 相关的操作。
01 /** 这是因为 people 是普通的 RDD,而 registerTempTable 函数不属于 RDD 类,只有通过 SchemaRDD 的实例才可以调用,所以这么调用会出现错误,解决办法有两个:
(1)registerTempTable 函数是 SQLContext 类中的,所以我们可以将 people 转换成 SchemaRDD,如下:
02 * User: 过往记忆
03 * Date: 14-12-16
04 * Time: 下午 10:16
05 * bolg: http://www.iteblog.com
06 * 本文地址:http://www.iteblog.com/archives/1224
07 * 过往记忆博客,专注于 hadoop、hive、spark、shark、flume 的技术博客,大量的干货
08 * 过往记忆博客微信公共帐号:iteblog_hadoop
09 */
10 scala val peopleSchema =sqlContext.createSchemaRDD(people)
11 peopleSchema:org.apache.spark.sql.SchemaRDD =
12 SchemaRDD[29] at RDD at SchemaRDD.scala:103
13 ==Query Plan ==
14 ==Physical Plan ==
15 ExistingRdd [name#4,age#5], MapPartitionsRDD[28] at
16 mapPartitions at basicOperators.scala:217
17
18 scala peopleSchema.registerTempTable(people)
19 warning:there were 1deprecation warning(s); re-run with-deprecation fordetails
这么调用就可以将 people 转成 SchemaRDD。
(2)、上面的方法是通过显示地调用 sqlContext.createSchemaRDD 将普通的 RDD 转成 SchemaRDD。其实我们还可以通过 Scala 的隐式语法来进行转换。我们先来看看 createSchemaRDD 函数的定义
1 /**
2 * Creates a SchemaRDD from an RDD of case classes.
4 * @group userf
6 implicitdefcreateSchemaRDD[A :Product:TypeTag](rdd:RDD[A]) ={7 SparkPlan.currentContext.set(self)
8 newSchemaRDD(this, SparkLogicalPlan(ExistingRdd.fromProductRdd(rdd))(self))
9 }
在定义 createSchemaRDD 的时候用到了 implicit 关键字,所以我们在使用的时候可以通过下面语句使用
1 scala import sqlContext.createSchemaRDD
2 import sqlContext.createSchemaRDD
3
4 scala people.registerAsTable(people)
5 warning:there were 1deprecation warning(s); re-run with-deprecation fordetails
这样就隐身地将 people 转换成 SchemaRDD 了。这是因为 Spark 可以隐式地将包含 case class 的 RDD 转换成 SchemaRDD。
看完上述内容,你们对怎么实现 Spark SchemaRDD 隐式转换有进一步的了解吗?如果还想了解更多知识或者相关内容,请关注丸趣 TV 行业资讯频道,感谢大家的支持。