spark中RDD、DataFrame创建及互相转换

1. RDD创建方式

parallelize 从一个Seq集合创建RDD。
例如： var rdd = sc.parallelize(1 to 10)

makeRDD 这种用法和parallelize完全相同

例如： var collect = Seq((1 to 10,Seq("slave007.lxw1234.com","slave002.lxw1234.com")),  
(11 to 15,Seq("slave013.lxw1234.com","slave015.lxw1234.com")))
var rdd = sc.makeRDD(collect) 

textFile 从外部存储创建RDD

例如： var rdd = sc.textFile("hdfs:///tmp/lxw1234/1.txt")  //hdfs上获取

 var rdd = sc.textFile("file:///etc/hadoop/conf/core-site.xml")  //本地文件系统获取

2. Dataframe创建方式

toDF

import sqlContext.implicits._
val df = Seq((1, "First Value", java.sql.Date.valueOf("2010-01-01")),

(2, "Second Value", java.sql.Date.valueOf("2010-02-01"))

).toDF("int_column", "string_column", "date_column")

createDataFrame

import org.apache.spark.sql.types._

val schema = StructType(List(
StructField("integer_column", IntegerType, nullable = false),
StructField("string_column", StringType, nullable = true),
StructField("date_column", DateType, nullable = true)))
val rdd = sc.parallelize(Seq(
Row(1, "First Value", java.sql.Date.valueOf("2010-01-01")),
Row(2, "Second Value", java.sql.Date.valueOf("2010-02-01"))))

val df = sqlContext.createDataFrame(rdd, schema)

通过文件直接创建DataFrame

val df = sqlContext.read.parquet("hdfs:/path/to/file")

val df = spark.read.json("examples/src/main/resources/people.json")

3. RDD转换为Dataframe
一、通过反射的方式来推断RDD元素中的元数据

// 从原来的RDD创建一个Row格式的RDD
// 创建与RDD 中Rows结构匹配的StructType，通过该StructType创建表示RDD 的Schema
// 通过SQLContext提供的createDataFrame方法创建DataFrame，方法参数为RDD 的Schema

val conf = new SparkConf().setMaster ("local").setAppName ("Test1")
val sc = new SparkContext (conf)
val sqlContext = new SQLContext(sc)
// import sqlContext.implicits._ case class Person(name:String,age:Int)
val people = sc.textFile ("d:/people.txt")
val schemaString = "name age"

val
schema = StructType
(
schemaString.split(" ").map(fieldName =>
StructField(fieldName,StringType,true))
)
val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))
val peopleSchemaRDD =
sqlContext.createDataFrame(rowRDD, schema)
peopleSchemaRDD .registerTempTable("people" )
val results = sqlContext . sql ("SELECT name FROM people" )
results.printSchema()
println(results.count())
results.map(t => "Name: " + t(0)).collect().foreach(println)

二、利用反射来推断包含特定类型对象的RDD的schema。
//2. 先创建一个bean类，然后将Rdd转换成DataFrame
case class Person(name: String, age: Int)
def main (args : Array[String]) : Unit =
{
val conf = new SparkConf().setMaster ("local").setAppName ("Test1")
val sc = new SparkContext (conf)
val sqlContext = new SQLContext(sc)

//隐式转换实现（关键点）
import sqlContext.implicits._

val people = sc.textFile("d:/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
people.registerTempTable("people")
val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)
teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)
teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)