import org.apache.spark.{SparkConf, SparkContext}

object SparkWordCount {
    def main (args:Array [String]): Unit = {
        //setMaster("local[9]") 表示在本地运行 Spark 程序，使用 9 个线程。local[*] 表示使用所有可用的处理器核心。
        //这种模式通常用于本地测试和开发。
        val conf = new SparkConf ().setAppName ("Word Count").setMaster("local[9]");
        val sc = new SparkContext (conf);
        sc.setLogLevel("ERROR")

            val data = List("Hello World", "Hello Spark")
            val textFile = sc.parallelize(data)
            val wordCounts = textFile.flatMap (line => line.split (" ")).map (
            word => (word, 1)).reduceByKey ( (a, b) => a + b)
            wordCounts.collect().foreach(println)
    }
}

val <Name>[:<type>] = <value>   // 常量（值），不可再赋值
var <Name>[:<type>] = <value>   // 变量，可再赋值

// 1. 字符串拼接
"Hello " + "World"              // = Hello World

// 2. 字符串内插（s 前缀）
val item = "World"
s"Hello $item"                  // = Hello World
s"Hello ${item}s"               // = Hello Worlds（引用名后接非字用 {}）
s"Hello ${item * 3}"            // = Hello WorldWorldWorld（{} 内可算式）

// 3. 格式化（f 前缀 / printf）
val pi = 3.1415926
f"pi = $pi%3.2f"                // = pi = 3.14

math.sin(1) + math.cos(1)       // Double = 1.3817...

// 表达式块：块中最后一个表达式作为整个块的返回值
val pi_approx = {
  val numerator = 355; val denominator = 113f
  numerator / denominator
}                               // Float = 3.141593

// if / if...else
if (<Boolean expression>) <expression>
if (<Boolean expression>) <expression> else <expression>

var x = 1; var y = 8
val max = if (x > y) x else y   // Int = 8

<expression> match {
  case <pattern> => <expression>
  [case ...]
}

// 基本匹配
val day = "mon"
val kind = day match {
  case "mon" | "tue" | "wed" | "thu" | "fri" => "weekday"
  case "sat" | "sun" => "weekend"
}                               // = weekday

// 带守卫条件 if 的匹配
val score = 95
score match {
  case score if (90 <= score) & (score < 100) => "A"
  case score if (70 <= score) & (score < 90)  => "B"
  case score if (60 <= score) & (score < 70)  => "C"
  case score if score < 60                    => "F"
}                               // = A

// 定义数值范围
1 to 5          // (1, 2, 3, 4, 5)   —— 包括头尾
1 until 5       // (1, 2, 3, 4)      —— 不包括最后一个
1 to 10 by 2    // (1, 3, 5, 7, 9)   —— 步长

// for 循环
for (x <- 1 to 7) { print(s"Day $x ") }       // Day 1 ... Day 7

// yield：返回值作为集合返回
for (x <- 1 to 7) yield { s"Day $x " }         // Vector("Day 1 ", ...)

// 迭代器哨位（带 if 过滤）
val threes = for (i <- 1 to 20 if i % 3 == 0) yield i   // Vector(3,6,9,12,15,18)

// while / do-while
var x = 10; while (x > 0) x -= 1               // 0
var x = 0; do println(s"x = $x") while (x > 0) // 至少执行一次 → x = 0

def hi = "hi"                                   // 无参

def hi: String = "hi"                           // 指定返回类型

def <id>(<id>:<type>[, ...]): <type> = <expr>   // 带参
def logit(p: Double): Double = {
  if (p >= 0 && p <= 1) math.log(p / (1 - p))
  else throw new Error(s"logit: parameter $p is out of range")
}

val f = (x: Double) => x * x                    // 匿名函数，f(2) = 4.0

val originalData = sc.textFile("FileStore/tables/03.08data/score.txt")
val distData = sc.parallelize(List(1, 2, 3, 4, 5))

// 创建键值对 RDD
val rdd = sc.parallelize(List("a", "b", "c"))
val pairRDD = rdd.map((_, 1))    // Array((a,1), (b,1), (c,1))

// 获取键 / 值
val keys = pairRDD.keys          // Array(a, b, c)
val values = pairRDD.values      // Array(1, 1, 1)

// mapValues：只对 Value 操作
// 类似 map，只对 `(Key, Value)` 中的 Value 做 map，不处理 Key
val result = pairRDD.mapValues((_, 1))
// Array[(String, (Int, Int))] = Array((a,(1,1)), (b,(1,1)), (c,(1,1)))

// groupByKey：按键分组
// 按键分组，在 `(K, V)` 上调用时返回 `(K, Iterable<V>)` 组成的新 RDD
val rdd = sc.parallelize(List("a", "b", "c", "c")).map((_, 1))
val result = rdd.groupByKey()
// Array((a,CompactBuffer(1)), (b,CompactBuffer(1)), (c,CompactBuffer(1, 1)))

// reduceByKey：统计每个键出现的次数
val result = rdd.reduceByKey((x, y) => x + y)
// Array((a,1), (b,1), (c,2))

// join：相同键的值整合，只保留两边都有的键  把键值对数据中相同键的值整合起来
val rdd_1 = sc.parallelize(List(("K1","V1"), ("K2","V2"), ("K4","V4")))
val rdd_2 = sc.parallelize(List(("K1","W1"), ("K2","W2"), ("K3","W3")))
rdd_1.join(rdd_2)
// Array((K1,(V1,W1)), (K2,(V2,W2)))

// lookup：返回指定键的所有值
rdd.collect                      // Array((a,1), (b,1), (c,1), (c,1))
val result = rdd.lookup("c")     // Seq[Int] = WrappedArray(1, 1)

// collect / count / first / take
val data = sc.parallelize(List(1, 2, 3, 4))
data.collect()   // Array(1, 2, 3, 4)
data.count()     // Long = 4
data.first()     // Int = 1
data.take(2)     // Array(1, 2)

// 向量函数：按下标生成
val a = DenseVector.tabulate(3)(i => i * i)      // DenseVector(0, 1, 4)
val a = DenseVector.tabulate(0 to 2)(i => i * i) // DenseVector(0, 1, 4)

// 按元素赋值：将向量 a 的元素分别加上其所在位置的指标
val a = DenseVector(Array(1.0, 2.0, 3.0))        // DenseVector(1.0, 2.0, 3.0)
a := a + DenseVector((0 to a.length - 1).toArray.map(_.toDouble))
// DenseVector(1.0, 3.0, 5.0)

DenseMatrix((1.0, 2.0, 3.0), (4.0, 5.0, 6.0))
new DenseMatrix(2, 3, Array(1.0, 2.0, 3.0, 4.0, 5.0, 6.0))
/* 列优先填充：
1.0  3.0  5.0
2.0  4.0  6.0 */

val Matrix = DenseMatrix.tabulate(3, 4){ (i, j) => i * i + j * j }
/*
0  1  4  9
1  2  5  10
4  5  8  13 */

Matrix.cols        // 列数
Matrix.rows        // 行数
Vector.length      // 向量长度

Matrix.toDenseVector               // 拉直矩阵
Vector.toDenseMatrix.reshape(n, m) // 堆叠向量
Matrix.reshape(n, m)               // 矩阵变形
diag(Matrix)                       // 取出对角元素

Matrix.toDenseVector  // 拉直矩阵 DenseVector(0, 1, 4, 1, 2, 5, 4, 5, 8, 9, 10, 13)

// 堆叠向量
a.toDenseMatrix.reshape(3, 1)
/*
1.0
3.0
5.0 */

// 矩阵变形
Matrix.reshape(4, 3)
/*
0  2  8
1  5  9
4  4  10
1  5  13 */

val matrix1 = DenseMatrix.ones[Double](3, 3)
val matrix2 = DenseMatrix.eye[Double](3)

DenseMatrix.horzcat(matrix1, matrix2)
/*
1.0  1.0  1.0  1.0  0.0  0.0
1.0  1.0  1.0  0.0  1.0  0.0
1.0  1.0  1.0  0.0  0.0  1.0 */

DenseMatrix.vertcat(matrix1, matrix2)
/*
1.0  1.0  1.0
1.0  1.0  1.0
1.0  1.0  1.0
1.0  0.0  0.0
0.0  1.0  0.0
0.0  0.0  1.0 */

val vec1 = DenseVector.ones[Double](3)
val vec2 = DenseVector.rangeD(0, 3)
DenseVector.horzcat(vec1, vec2)   // 成矩阵 3×2
DenseVector.vertcat(vec1, vec2)   // DenseVector(1.0, 1.0, 1.0, 0.0, 1.0, 2.0)

val vec1 = DenseVector.ones[Double](3)   // (1.0, 1.0, 1.0)
val vec2 = DenseVector.rangeD(1, 4)      // (1.0, 2.0, 3.0)

vec1 >:> vec2   // BitVector()   （逐元素比较，返回满足的下标）
vec1 :== vec2   // BitVector(0)
vec1 == vec2    // false

vec1.t * vec2   // Double = 6.0  （内积）
vec1 dot vec2   // Double = 6.0
vec1 * vec2     // DenseVector(1.0, 2.0, 3.0)  （向量按元素乘）

any(a)   // 任意元素为真则为真
all(a)   // 所有元素为真才为真

// 矩阵：<:< 返回 Boolean 矩阵，* 是矩阵乘法
matrix2 <:< matrix1
/*
false  false  false
false  true   false
false  false  true */
matrix1 * matrix2   // 矩阵乘法

// -1 代表最后一个元素；:: 代表所有元素
Matrix(i, j)              Vec(i)                 // 提取单个元素
Matrix(i to j, m to n)    Vec(i to j)            // 提取多个元素（含尾）
Matrix(i until j, m to n) Vec(i until j [by seq]) // until 不含尾
Vector.slice(start, end)  // 包括 start，不包括 end

Matrix(1, ::)   // Transpose(DenseVector(2.0, 4.0, 6.0))  取第 1 行
Matrix(::, 1)   // DenseVector(3.0, 4.0)                  取第 1 列
vec.slice(0, 3) // DenseVector(1.0, 2.0, 3.0)

val matrix1 = diag(DenseVector[Double](1, 2, 3))
eig(matrix1)
// Eig(特征值实部, 特征值虚部, 特征向量矩阵)

val svd.SVD(u, s, v) = svd(matrix1)
// u: 左奇异向量矩阵；s: 奇异值向量 DenseVector(3.0, 2.0, 1.0)；v: 右奇异向量矩阵

sum(Matrix)          // 所有元素求和         → 21.0
sum(Matrix(*, ::))   // 对所有行求和         → DenseVector(9.0, 12.0)
Vec.reduce((x, y) => x + y)   // 向量可用 reduce 代替 sum

// 数学计算包
import breeze.numerics._
// 提供：sin/sinh/asin/asinh、cos/cosh/acos/acosh、tan/tanh/atan/atanh、
//       log、exp、log10、sqrt、pow

// 统计计算包
import breeze.stats._
breeze.stats.mean(Vec)        // 求均值，也可用 sum(Vec)/Vec.length 代替

// 产生随机数 / 密度与分布计算
val RandNumbers = new DistributionName(param).sample(size)
DistributionName.pdf()    // 计算密度
DistributionName.cdf()    // 计算分布

val Distribution = DistributionName(Parameter)
val RandNumbers = DenseMatrix.rand(n, m, Distribution)
val rdd = sc.parallelize(RandNumbers)
val rdd = sc.parallelize(Array(Vector, Vector, …))

import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.mllib.linalg.{Matrix, Matrices}

// 稠密向量
val dv = Vectors.dense(1.0, 2.0, 3.0)                      // [1.0,2.0,3.0]
// 稀疏向量（两种写法）
val sv1 = Vectors.sparse(3, Array(0, 2), Array(1.0, 3.0)) // (向量大小, 非零位置, 非零数值)
val sv2 = Vectors.sparse(3, Seq((0, 1.0), (2, 3.0)))      // (3, [0,2], [1.0,3.0])

// 稠密矩阵（列优先填充）
val dm = Matrices.dense(4, 3, Array(1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0))
/*
1.0  5.0  9.0
2.0  6.0  10.0
3.0  7.0  11.0
4.0  8.0  12.0 */

import org.apache.spark.mllib.linalg.distributed._

// 1. 行矩阵 RowMatrix
val data = Array(
  Vectors.sparse(5, Seq((1, 1.0), (3, 7.0))),
  Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
  Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0))
val MAT = new RowMatrix(sc.parallelize(data))

// 2. 索引行矩阵 IndexedRowMatrix
val data = Array(
  IndexedRow(1, Vectors.sparse(5, Seq((1, 1.0), (3, 7.0)))),
  IndexedRow(2, Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0)),
  IndexedRow(3, Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)))
val MAT = new IndexedRowMatrix(sc.parallelize(data))

// 3. 坐标矩阵 CoordinateMatrix
val ent1 = new MatrixEntry(0, 1, 0.5)
val ent2 = new MatrixEntry(2, 2, 1.8)
val CorrMat = new CoordinateMatrix(sc.parallelize(Array(ent1, ent2)))

// 4. 分块矩阵 BlockMatrix
val m1 = Matrices.dense(2, 2, Array(1.0, 2.0, 3.0, 4.0))
val m2 = Matrices.dense(2, 2, Array(5.0, 6.0, 7.0, 8.0))
val blocks = sc.parallelize(Seq(((0, 0), m1), ((0, 1), m2)))
val MAT = new BlockMatrix(blocks, 2, 2)