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