Spark之 RDD转换成DataFrame的Scala实现 Spark之 RDD转换成DataFrame的Scala实现

依赖

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.1.3</version>
</dependency>

RDD转化成DataFrame:通过StructType指定schema

package com.zy.sparksql
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
/**
* RDD转化成DataFrame:通过StructType指定schema
*/
object StructTypeSchema {
def main(args: Array[String]): Unit = {
//创建sparkSession对象
val sparkSession: SparkSession = SparkSession.builder().appName("StructTypeSchema").master("local[2]").getOrCreate()
//获取sparkContext
val sc: SparkContext = sparkSession.sparkContext
//设置日志级别
sc.setLogLevel("WARN")
//读取文件
val textFile: RDD[String] = sc.textFile("D:\person.txt")
//切分文件
val lineArrayRDD: RDD[Array[String]] = textFile.map(_.split(","))
//关联对象
val rowRDD: RDD[Row] = lineArrayRDD.map(x => Row(x(0).toInt, x(1), x(2).toInt))
//创建rdd的schema信息
val schema: StructType = (new StructType)
.add("id", IntegerType, true, "id")
.add("name", StringType, false, "姓名")
.add("age", IntegerType, true, "年龄")
//根据rdd和schema信息创建DataFrame
val personDF: DataFrame = sparkSession.createDataFrame(rowRDD, schema)
//DSL操作

personDF.show()
//sql 操作
//将df注册成表
personDF.createTempView("person")
sparkSession.sql("select * from person where id =3").show()
sparkSession.stop()
}
}

 

RDD转化成DataFrame:利用反射机制推断schema

package com.zy.sparksql
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
* RDD转化成DataFrame:利用反射机制推断schema
*/
//todo 定义一个样例类
case class Person(id: Int, name: String, age: Int)
object CaseClassSchema {
def main(args: Array[String]): Unit = {
//构建sparkSession 指定appName和master地址(本地测试local)
val sparkSession: SparkSession = SparkSession.builder().appName("CaseClassSchema").master("local[2]").getOrCreate()
//获取sparkContext
val sc: SparkContext = sparkSession.sparkContext
//设置日志输出级别
sc.setLogLevel("WARN")
//加载数据
val dataRDD: RDD[String] = sc.textFile("D:\person.txt")
//切分数据
val lineArrayRDD: RDD[Array[String]] = dataRDD.map(_.split(","))
//将rdd和person样例类关联
val personRDD: RDD[Person] = lineArrayRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))
//将rdd转换成dataFrame 导入隐式转换
import sparkSession.implicits._
val personDF: DataFrame = personRDD.toDF
//DSL语法

personDF.show()
personDF.printSchema()
personDF.select("name").show()
personDF.filter($"age" > 30).show()
println("---------------------------------------------")
//sql语法
//首先要创建临时视图
personDF.createTempView("person")
sparkSession.sql("select * from person where id>1").show()
sparkSession.stop()
}
}