Spark之 RDD转换成DataFrame的Scala实现-蒲公英云

Spark之 RDD转换成DataFrame的Scala实现

水深无声 2022-01-07 10:57 382阅读 0赞

依赖

<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()

}
}

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
/**
* 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()