大数据Spark DataFrame

1）、分布式的数据集，并且以列的方式组合的，相当于具有schema的RDD；
2）、相当于关系型数据库中的表，但是底层有优化；
3）、提供了一些抽象的操作，如select、filter、aggregation、plot；
4）、它是由于R语言或者Pandas语言处理小数据集的经验应用到处理分布式大数据集上；
5）、在1.3版本之前，叫SchemaRDD；

import org.apache.spark.sql._
// Create a Row from values.
Row(value1, value2, value3, ...)
// Create a Row from a Seq of values.
Row.fromSeq(Seq(value1, value2, ...))

/**
 * 封装电影评分数据
 *
 * @param userId 用户ID
 * @param itemId 电影ID
 * @param rating 用户对电影评分
 * @param timestamp 评分时间戳
 */
case class MovieRating(
                        userId: String,
                        itemId: String,
                        rating: Double,
                        timestamp: Long
                      )

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
 * 采用反射的方式将RDD转换为DataFrame和Dataset
 */
object SparkRDDInferring {
  def main(args: Array[String]): Unit = {
    // 构建SparkSession实例对象
    val spark: SparkSession = SparkSession
      .builder() // 使用建造者模式构建对象
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[3]")
      .getOrCreate()
    import spark.implicits._
    // 读取电影评分数据u.data, 每行数据有四个字段，使用制表符分割
    // user id | item id | rating | timestamp.
    val rawRatingsRDD: RDD[String] = spark.sparkContext
      .textFile("datas/ml-100k/u.data", minPartitions = 2)
    // 转换数据
    val ratingsRDD: RDD[MovieRating] = rawRatingsRDD
      .filter(line => null != line && line.trim.split("\t").length == 4)
      .mapPartitions{iter =>
        iter.map{line =>
          // 拆箱操作, Python中常用
          val Array(userId, itemId, rating, timestamp) = line.trim.split("\t")
          // 返回MovieRating实例对象
          MovieRating(userId, itemId, rating.toDouble, timestamp.toLong)
        }
      }
    // 将RDD转换为DataFrame和Dataset
    val ratingsDF: DataFrame = ratingsRDD.toDF()
    /*
    root
    |-- userId: string (nullable = true)
    |-- itemId: string (nullable = true)
    |-- rating: double (nullable = false)
    |-- timestamp: long (nullable = false)
    */
    ratingsDF.printSchema()
    ratingsDF.show(10)
    // 应用结束，关闭资源
    spark.stop()
  }
}

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
/**
 * 自定义Schema方式转换RDD为DataFrame
 */
object SparkRDDSchema {
  def main(args: Array[String]): Unit = {
    // 构建SparkSession实例对象
    val spark: SparkSession = SparkSession
      .builder() // 使用建造者模式构建对象
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[3]")
      .getOrCreate()
    import spark.implicits._
    // 读取电影评分数据u.data, 每行数据有四个字段，使用制表符分割
    // user id | item id | rating | timestamp.
    val ratingsRDD: RDD[String] = spark
      .sparkContext.textFile("datas/ml-100k/u.data", minPartitions = 2)
    // a. RDD[Row]
    val rowsRDD: RDD[Row] = ratingsRDD.mapPartitions{ iter =>
      iter.map{line =>
        // 拆箱操作, Python中常用
        val Array(userId, itemId, rating, timestamp) = line.trim.split("\t")
        // 返回Row实例对象
        Row(userId, itemId, rating.toDouble, timestamp.toLong)
      }
    }
    // b. schema
    val rowSchema: StructType = StructType(
      Array(
        StructField("userId", StringType, nullable = true),
        StructField("itemId", StringType, nullable = true),
        StructField("rating", DoubleType, nullable = true),
        StructField("timestamp", LongType, nullable = true)
      )
    )
    // c. 应用函数createDataFrame
    val ratingDF: DataFrame = spark.createDataFrame(rowsRDD, rowSchema)
    ratingDF.printSchema()
    ratingDF.show(10, truncate = false)
    // 应用结束，关闭资源
    spark.stop()
  }
}

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
 * 隐式调用toDF函数，将数据类型为元组的Seq和RDD集合转换为DataFrame
 */
object SparkSQLToDF {
  def main(args: Array[String]): Unit = {
    // 构建SparkSession实例对象，通过建造者模式创建
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[3]")
      .getOrCreate()
    import spark.implicits._
    // TODO: 1、构建RDD，数据类型为三元组形式
    val usersRDD: RDD[(Int, String, Int)] = spark.sparkContext.parallelize(
      Seq(
        (10001, "zhangsan", 23),
        (10002, "lisi", 22),
        (10003, "wangwu", 23),
        (10004, "zhaoliu", 24)
      )
    )
    // 将RDD转换为DataFrame
    val usersDF: DataFrame = usersRDD.toDF("id", "name", "age")
    usersDF.printSchema()
    usersDF.show(10, truncate = false)
    println("========================================================")
    val df: DataFrame = Seq(
      (10001, "zhangsan", 23),
      (10002, "lisi", 22),
      (10003, "wangwu", 23),
      (10004, "zhaoliu", 24)
    ).toDF("id", "name", "age")
    df.printSchema()
    df.show(10, truncate = false)
    // TODO: 应用结束，关闭资源
    spark.stop()
  }
}