Apache Spark 技术文档

// 1. 创建 RDD
val rdd = sc.parallelize(1 to 100)
// 说明：parallelize 将本地集合转换为 RDD
// 参数：1 to 100 表示创建包含 1 到 100 的 RDD

// 2. map 操作 - 一对一转换
val doubled = rdd.map(_ * 2)
// 说明：map 对 RDD 中的每个元素应用函数
// 结果：每个数字都乘以 2
// 特点：输入和输出元素数量相同

// 3. filter 操作 - 过滤数据
val filtered = rdd.filter(_ > 50)
// 说明：filter 根据条件过滤元素
// 结果：只保留大于 50 的数字
// 特点：输出元素数量可能少于输入

// 4. flatMap 操作 - 一对多转换
val words = rdd.flatMap(_.toString.split(""))
// 说明：flatMap 对每个元素应用函数，然后展平结果
// 结果：将每个数字转换为字符串，然后分割
// 特点：输入 1 个元素可能输出多个元素

// 5. groupBy 操作 - 分组
val grouped = rdd.groupBy(_ % 2)
// 说明：根据键对元素进行分组
// 结果：按奇偶数分组
// 特点：返回 (键, 值列表) 的 RDD

// 6. distinct 操作 - 去重
val unique = rdd.distinct()
// 说明：去除重复元素
// 特点：需要 shuffle 操作

// 7. union 操作 - 合并
val rdd1 = sc.parallelize(1 to 50)
val rdd2 = sc.parallelize(51 to 100)
val combined = rdd1.union(rdd2)
// 说明：合并两个 RDD
// 特点：不检查重复，只是简单合并

// 1. collect 操作 - 收集所有数据
val result = rdd.collect()
// 说明：将 RDD 中的所有数据收集到 Driver 节点
// 注意：数据量大时可能导致内存溢出
// 适用：小数据集或测试环境

// 2. count 操作 - 计数
val count = rdd.count()
// 说明：返回 RDD 中元素的总数
// 特点：不需要收集数据到 Driver
// 适用：统计数据集大小

// 3. reduce 操作 - 归约
val sum = rdd.reduce(_ + _)
// 说明：使用指定的函数归约 RDD 中的元素
// 结果：计算所有元素的总和
// 特点：需要可交换和可结合的函数

// 4. take 操作 - 取前 N 个元素
val first10 = rdd.take(10)
// 说明：返回 RDD 中的前 N 个元素
// 特点：不需要收集所有数据
// 适用：查看数据样本

// 5. first 操作 - 取第一个元素
val first = rdd.first()
// 说明：返回 RDD 中的第一个元素
// 特点：只计算第一个分区
// 适用：快速查看数据

// 6. saveAsTextFile 操作 - 保存为文本文件
rdd.saveAsTextFile("hdfs://path/to/output")
// 说明：将 RDD 保存为文本文件
// 特点：每个分区保存为一个文件
// 适用：持久化计算结果

// 7. foreach 操作 - 遍历执行
rdd.foreach(println)
// 说明：对每个元素执行函数
// 特点：在 Executor 上执行
// 适用：打印或发送数据

// 创建血缘关系链
val rdd1 = sc.parallelize(1 to 100)
val rdd2 = rdd1.map(_ * 2)
val rdd3 = rdd2.filter(_ > 100)
// rdd3 的血缘关系: rdd1 -> rdd2 -> rdd3

// 查看血缘关系
println(rdd3.toDebugString)
// 输出：
// (2) MapPartitionsRDD[2] at filter at <console>:1 []
//  |  MapPartitionsRDD[1] at map at <console>:1 []
//  |  ParallelCollectionRDD[0] at parallelize at <console>:1 []

// 示例：数据倾斜处理中的血缘关系
val originalRdd = sc.parallelize(1 to 1000)

// 第一步：加盐
val saltedRdd = originalRdd.map(x => (x % 10, x))
// 血缘关系：originalRdd -> saltedRdd

// 第二步：第一次聚合
val firstAgg = saltedRdd.reduceByKey(_ + _)
// 血缘关系：originalRdd -> saltedRdd -> firstAgg

// 第三步：去盐
val finalResult = firstAgg.map(x => (x._1, x._2))
// 血缘关系：originalRdd -> saltedRdd -> firstAgg -> finalResult

// 如果 finalResult 的某个分区丢失
// Spark 会从 originalRdd 开始重新计算整个链条

// 1. 合理使用缓存
val expensiveRdd = rdd.map(expensiveFunction)
expensiveRdd.cache() // 缓存中间结果

// 2. 避免过长的血缘关系
val result1 = rdd.map(f1).map(f2).map(f3)
// 可以合并为：
val result2 = rdd.map(x => f3(f2(f1(x))))

// 3. 使用检查点
rdd.checkpoint() // 切断血缘关系，保存到存储系统

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

val conf = new SparkConf()
  .setAppName("MySparkApp")
  .setMaster("local[*]")

val sc = new SparkContext(conf)

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("DataFrameExample")
  .master("local[*]")
  .getOrCreate()

// 创建 DataFrame
val df = spark.read.json("path/to/data.json")

// 显示数据
df.show()

// 选择列
df.select("name", "age").show()

// 过滤数据
df.filter(df("age") > 18).show()

// 分组聚合
df.groupBy("department").count().show()

case class Person(name: String, age: Int, department: String)

val ds = spark.read.json("path/to/data.json").as[Person]

// 类型安全的操作
val adults = ds.filter(_.age > 18)
val names = ds.map(_.name)

// 读取 JSON
val df = spark.read.json("data.json")

// 读取 CSV
val df = spark.read.option("header", "true").csv("data.csv")

// 读取 Parquet
val df = spark.read.parquet("data.parquet")

// 读取数据库
val df = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/db")
  .option("dbtable", "table")
  .option("user", "username")
  .option("password", "password")
  .load()

// 注册临时视图
df.createOrReplaceTempView("people")

// 执行 SQL 查询
val result = spark.sql("""
  SELECT department, AVG(age) as avg_age
  FROM people
  WHERE age > 18
  GROUP BY department
  ORDER BY avg_age DESC
""")

import org.apache.spark.streaming._

val ssc = new StreamingContext(sc, Seconds(1))

// 创建 DStream
val lines = ssc.socketTextStream("localhost", 9999)

// 处理数据
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)

// 输出结果
wordCounts.print()

// 启动流处理
ssc.start()
ssc.awaitTermination()

// 窗口大小为 30 秒，滑动间隔为 10 秒
val windowedCounts = wordCounts.window(Seconds(30), Seconds(10))

import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.Pipeline

// 特征向量化
val assembler = new VectorAssembler()
  .setInputCols(Array("feature1", "feature2", "feature3"))
  .setOutputCol("features")

// 逻辑回归
val lr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.01)

// 构建管道
val pipeline = new Pipeline()
  .setStages(Array(assembler, lr))

// 训练模型
val model = pipeline.fit(trainingData)

// 预测
val predictions = model.transform(testData)

import org.apache.spark.graphx._

// 创建顶点 RDD
val vertices = sc.parallelize(Array(
  (1L, "Alice"),
  (2L, "Bob"),
  (3L, "Charlie")
))

// 创建边 RDD
val edges = sc.parallelize(Array(
  Edge(1L, 2L, "friend"),
  Edge(2L, 3L, "colleague")
))

// 创建图
val graph = Graph(vertices, edges)

// PageRank 算法
val ranks = graph.pageRank(0.0001).vertices

// 连通组件
val cc = graph.connectedComponents().vertices

// 三角形计数
val triangles = graph.triangleCount().vertices

// 缓存策略
rdd.persist(StorageLevel.MEMORY_ONLY)
rdd.persist(StorageLevel.MEMORY_AND_DISK)

// 序列化优化
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

// 设置并行度
val rdd = sc.parallelize(data, 200)

// 重新分区
val repartitioned = rdd.repartition(100)

// 合并小分区
val coalesced = rdd.coalesce(50)

// 加盐技术
val saltedRdd = rdd.map(x => (x._1 + "_" + Random.nextInt(100), x._2))

// 两阶段聚合
val firstStage = rdd.map(x => (x._1 + "_" + Random.nextInt(100), x._2))
  .reduceByKey(_ + _)
val secondStage = firstStage.map(x => (x._1.split("_")(0), x._2))
  .reduceByKey(_ + _)

// 启用动态分配
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.minExecutors", "1")
spark.conf.set("spark.dynamicAllocation.maxExecutors", "10")

// 启用推测执行
spark.conf.set("spark.speculation", "true")

// 合并小文件
df.coalesce(1).write.parquet("output")

// 使用 repartition
df.repartition(10).write.parquet("output")

// 1. 数据清洗和预处理
val userBehaviorRdd = spark.read.json("hdfs://user-behavior/*.json")
  .filter($"timestamp".isNotNull)
  .filter($"userId".isNotNull)

// 2. 用户行为分析
val userStats = userBehaviorRdd
  .groupBy($"userId")
  .agg(
    count("*").as("totalActions"),
    countDistinct($"productId").as("uniqueProducts"),
    sum(when($"action" === "purchase", 1).otherwise(0)).as("purchases")
  )

// 3. 热门商品分析
val popularProducts = userBehaviorRdd
  .filter($"action" === "view")
  .groupBy($"productId")
  .count()
  .orderBy(desc("count"))
  .limit(100)

// 4. 用户画像构建
val userProfile = userBehaviorRdd
  .groupBy($"userId")
  .agg(
    collect_list($"category").as("interests"),
    avg($"sessionDuration").as("avgSessionTime"),
    count("*").as("activityLevel")
  )

// 1. 创建 StreamingContext
val ssc = new StreamingContext(spark.sparkContext, Seconds(10))

// 2. 读取日志流
val logStream = ssc.socketTextStream("localhost", 9999)

// 3. 解析日志
val parsedLogs = logStream.map(line => {
  val parts = line.split(" ")
  LogEntry(parts(0), parts(1), parts(2), parts(3))
})

// 4. 实时统计
val pageViews = parsedLogs
  .map(_.page)
  .countByValue()

val errorCount = parsedLogs
  .filter(_.statusCode.startsWith("4") || _.statusCode.startsWith("5"))
  .count()

// 5. 输出结果
pageViews.print()
errorCount.print()

// 1. 数据准备
val ratings = spark.read.option("header", "true")
  .csv("ratings.csv")
  .select($"userId".cast("int"), $"movieId".cast("int"), $"rating".cast("double"))

// 2. 特征工程
val assembler = new VectorAssembler()
  .setInputCols(Array("userId", "movieId"))
  .setOutputCol("features")

// 3. 模型训练
val als = new ALS()
  .setMaxIter(10)
  .setRegParam(0.01)
  .setUserCol("userId")
  .setItemCol("movieId")
  .setRatingCol("rating")

val pipeline = new Pipeline()
  .setStages(Array(assembler, als))

val model = pipeline.fit(ratings)

// 4. 推荐生成
val recommendations = model.transform(ratings)
  .select($"userId", $"movieId", $"prediction")
  .orderBy($"userId", desc("prediction"))

// 1. 数据预处理
val transactions = spark.read.parquet("transactions.parquet")
  .withColumn("hour", hour($"timestamp"))
  .withColumn("dayOfWeek", dayofweek($"timestamp"))

// 2. 特征工程
val features = transactions
  .groupBy($"cardId")
  .agg(
    count("*").as("transactionCount"),
    sum($"amount").as("totalAmount"),
    avg($"amount").as("avgAmount"),
    max($"amount").as("maxAmount")
  )

// 3. 异常检测
val anomalies = features
  .filter($"transactionCount" > 100)
  .filter($"maxAmount" > 10000)
  .filter($"avgAmount" > 1000)

// 4. 实时监控
val streamingTransactions = ssc.socketTextStream("localhost", 9999)
  .map(parseTransaction)
  .filter(_.amount > 5000)
  .foreachRDD(rdd => {
    // 发送告警
    rdd.foreach(transaction => {
      sendAlert(s"High value transaction: ${transaction}")
    })
  })

// 1. 调整内存配置
spark.conf.set("spark.executor.memory", "8g")
spark.conf.set("spark.executor.memoryFraction", "0.8")
spark.conf.set("spark.storage.memoryFraction", "0.6")

// 2. 使用合适的存储级别
val rdd = sc.textFile("large-file.txt")
rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)

// 3. 分批处理
val partitions = rdd.coalesce(200) // 增加分区数
val processed = partitions.mapPartitions(iter => {
  // 处理每个分区
  iter.map(processRecord)
})

// 1. 检测数据倾斜
val keyCounts = rdd.map(_._1).countByKey()
val maxCount = keyCounts.values.max
val avgCount = keyCounts.values.sum / keyCounts.size

if (maxCount > avgCount * 3) {
  println("数据倾斜 detected!")
}

// 2. 加盐处理
val saltedRdd = rdd.map(x => {
  val salt = Random.nextInt(100)
  (s"${x._1}_$salt", x._2)
})

// 3. 两阶段聚合
val firstStage = saltedRdd.reduceByKey(_ + _)
val secondStage = firstStage.map(x => {
  val originalKey = x._1.split("_")(0)
  (originalKey, x._2)
}).reduceByKey(_ + _)

// 1. 合并小文件
val df = spark.read.parquet("input/*.parquet")
df.coalesce(10).write.parquet("output")

// 2. 使用 repartition
val repartitioned = df.repartition(10, $"date")
repartitioned.write.parquet("output")

// 3. 动态分区
df.write
  .partitionBy("year", "month", "day")
  .parquet("output")

// 1. Spark UI 监控
// 访问 http://driver-node:4040

// 2. 自定义监控
val metrics = spark.sparkContext.statusTracker
val executorInfos = metrics.getExecutorInfos
executorInfos.foreach(executor => {
  println(s"Executor ${executor.executorId}: " +
    s"Memory ${executor.memoryUsed}/${executor.memoryTotal}")
})