Spark SQL 2.4.8 データフレームを操作するための2つの方法

package com.scala.demo.sql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{DataType, DataTypes, StructField, StructType}

object Demo01 {
  def main(args: Array[String]): Unit = {
    // 1. Create SparkContext and SparkSession objects
    val sc = new SparkContext(new SparkConf().setAppName("Demo01").setMaster("local[2]"))
    val sparkSession = SparkSession.builder().getOrCreate()

    // 2. Use StructType to define Schema
    val mySchema = StructType(List(
      StructField("empno", DataTypes.IntegerType, false),
      StructField("ename", DataTypes.StringType, false),
      StructField("job", DataTypes.StringType, false),
      StructField("mgr", DataTypes.StringType, false),
      StructField("hiredate", DataTypes.StringType, false),
      StructField("sal", DataTypes.IntegerType, false),
      StructField("comm", DataTypes.StringType, false),
      StructField("deptno", DataTypes.IntegerType, false)
    ))
    // 3. Read the data
    val empRDD = sc.textFile("file:///D:\\TestDatas\\\emp.csv")

    // 4. Map it to a ROW object
    val rowRDD = empRDD.map(line => {
      val strings = line.split(",")
      Row(strings(0).toInt, strings(1), strings(2), strings(3), strings(4), strings(5).toInt, strings(6), strings(7).toInt)
    })

    // 5. Create a DataFrame
    val dataFrame = sparkSession.createDataFrame(rowRDD, mySchema)

    // 6. Display the content DSL
	dataFrame.groupBy("deptno").sum("sal").as("result").sort("sum(sal)").show()
  }
}

package com.scala.demo.sql

import org.apache.spark.rdd.
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.types.{DataType, DataTypes, StructField, StructType}

// 0. data analysis
// 7499,allen,salesman,7698,1981/2/20,1600,300,30
// 1. Define the Emp sample class
case class Emp(empNo:Int,empName:String,job:String,mgr:String,hiredate:String,sal:Int,comm:String,deptNo:Int)

object Demo02 {
  def main(args: Array[String]): Unit = {
    // 2. Read the data and map it to a Row object
    val sc = new SparkContext(new SparkConf().setMaster("local[2]").setAppName("Demo02"))
    val mapRdd = sc.textFile("file:///D:\\TestDatas\\emp.csv")
      .map(_.split(","))

    val rowRDD:RDD[Emp] = mapRdd.map(line => Emp(line(0).toInt, line(1), line(2), line(3), line(4), line(5).toInt, line(6), line(7).toInt))

    // 3. Create the dataframe
    val spark = SparkSession.builder().getOrCreate()
    // Introduce the spark implicit conversion function
    import spark.implicits._
    // Convert the RDD to a Dataframe
    val dataFrame = rowRDD.toDF

    // 4.2 sql statement operations
    // 1. Register the dataframe as a temporary table
    dataFrame.createOrReplaceTempView("emp")
    // 2. write sql statements to perform the operation
    spark.sql("select deptNo,sum(sal) as total from emp group by deptNo order by total desc").show()

    // Close the resource
    spark.stop()
    sc.stop()
  }
}