Spark IDEA 编程环境配置

D:\spark-2.4.7-bin-hadoop2.7\examples\src\main\scala\spark\ml\cookbook\chapter1

// package 打包
package spark.ml.cookbook.chapter1

// 导入 SparkSession 上下文
import org.apache.spark.sql.SparkSession
import org.apache.log4j.Logger
import org.apache.log4j.Level // 减少输出内容

object MyFirstSpark20 {
    def main(args: Array[String]): Unit = {
        // 输出日志级别为 error，减少输出
        Logger.getLogger("org").setLevel(Level.ERROR)
        // 使用 Builder 模型，指定配置初始化 SparkSession，作为访问Spark集群的入口
        val spark = SparkSession
            .builder
            .master("local[*]")
            .appName("myFirstSpark20")
            .config("spark.sql.warehouse.dir", ".")
            .getOrCreate()
        // 以上为创建 SparkSession 的典型方法

        val x = Array(1.0,5.0,8.0,10.0,15.0,21.0,27.0,30.0,38.0,45.0,50.0,64.0)
        val y = Array(5.0,1.0,4.0,11.0,25.0,18.0,33.0,20.0,30.0,43.0,55.0,57.0)

        // 创建数组的 RDD
        val xRDD = spark.sparkContext.parallelize(x)
        val yRDD = spark.sparkContext.parallelize(y)

        // 创建新的 打包后的 rdd
        val zipedRDD = xRDD.zip(yRDD)
        zipedRDD.collect().foreach(println)
        // (1.0,5.0)
        // (5.0,1.0)
        // ...

        val xSum = zipedRDD.map(_._1).sum()
        val ySum = zipedRDD.map(_._2).sum()
        val xySum = zipedRDD.map(c => c._1 * c._2).sum()
        val n = zipedRDD.count()

        println("RDD X Sum: " +xSum) //314.0
        println("RDD Y Sum: " +ySum) //302.0
        println("RDD X*Y Sum: "+xySum) // 11869.0
        println("Total count: "+n) // 12
        spark.stop() // 退出程序，释放资源
    }
}