Scala 大数据生态实战：Spark 核心与数据处理范式

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

val conf = new SparkConf().setAppName("demo").setMaster("local[*]")
val sc = new SparkContext(conf)

val rdd = sc.parallelize(1 to 1000)          // 造一个 RDD
val sum = rdd.map(_ * 2).reduce(_ + _)       // 变换 + 动作
println(sum)                                  // 1001000

val rdd = sc.parallelize(1 to 100)
val filtered = rdd.filter(_ % 2 == 0)     // transformation，不执行
val count = filtered.count()               // action，此时才算

rdd.map(_ * 2)              // 逐个变换
rdd.flatMap(x => Seq(x, x)) // 展开
rdd.filter(_ > 10)          // 过滤
rdd.distinct()              // 去重
rdd.groupByKey()            // 按键分组
rdd.reduceByKey(_ + _)      // 按键聚合（比 groupByKey 高效）
rdd.join(other)             // 连接

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("demo").master("local[*]").getOrCreate()
import spark.implicits._

val df = spark.read.option("header", true).csv("data/users.csv")
df.show(5)
df.printSchema()

import org.apache.spark.sql.functions._

val result = df
  .filter(col("age") > 18)
  .groupBy("city")
  .agg(count("*").as("cnt"), avg("age").as("avg_age"))
  .orderBy(desc("cnt"))

result.show()

case class User(name: String, age: Int)

val ds: Dataset[User] = spark.read.json("data/users.json").as[User]
val adults = ds.filter(_.age > 18)   // 类型安全，编译期检查

df.createOrReplaceTempView("users")

val sqlDf = spark.sql("""
  SELECT city, COUNT(*) AS cnt, AVG(age) AS avg_age
  FROM users
  WHERE age > 18
  GROUP BY city
  ORDER BY cnt DESC
""")
sqlDf.show()

// 读
val fromParquet = spark.read.parquet("s3://bucket/data/part-*")
// 写
df.write.mode("overwrite").partitionBy("city").parquet("out/users")

df.groupBy("city").count().show()
df.rolling(...)                     // 窗口（时间序列）
df.withColumn("ratio", col("a") / col("b")).show()

// 调整分区数控制并行度
val repartitioned = df.repartition(200)     // 增分区（可能 shuffle）
val coalesced     = df.coalesce(20)         // 减分区（尽量不 shuffle）

val cached = df.cache()      // 缓存在内存
// 或
df.persist(StorageLevel.MEMORY_AND_DISK)

import org.apache.spark.sql.functions.broadcast

val smallDf = spark.read.csv("data/dim.csv")   // 小维表
val joined = bigDf.join(broadcast(smallDf), "key")  // 广播小表，避免 shuffle

import org.apache.flink.streaming.api.scala._

val env = StreamExecutionEnvironment.getExecutionEnvironment
val text = env.socketTextStream("host", 9999)
text
  .flatMap(_.split(" "))
  .map((_, 1))
  .keyBy(_._1)
  .sum(1)
  .print()
env.execute("word count")