2019-01-04

tools

Apache Spark - 统一分析引擎

项目简介

Apache Spark是一个开源的统一分析引擎，专为大规模数据处理而设计。由加州大学伯克利分校的AMPLab开发，后来成为Apache软件基金会的顶级项目。Spark相比传统的MapReduce框架，最大的优势是基于内存的计算，能够显著提高数据处理速度。

Spark的核心理念是提供一个统一的平台来处理各种大数据工作负载，包括批处理、实时流处理、交互式查询、机器学习和图计算。这种统一的架构使得开发者可以使用一套API来解决多种数据处理需求。

主要特性

内存计算：数据存储在内存中，比磁盘I/O快100倍
多语言支持：支持Scala、Java、Python和R
统一API：批处理和流处理使用相同的API
丰富的库：内置机器学习、图计算和SQL查询库
容错性：基于RDD的血缘关系实现故障恢复
易于使用：简洁的编程模型和交互式Shell

项目原理

核心架构

Spark采用主从架构模式：

Spark架构
├── Driver Program (驱动程序)
│   ├── SparkContext
│   └── DAG Scheduler
├── Cluster Manager (集群管理器)
│   ├── Standalone
│   ├── YARN
│   └── Mesos
└── Worker Nodes (工作节点)
    ├── Executor
    ├── Cache
    └── Tasks

RDD（弹性分布式数据集）

RDD是Spark的核心抽象，具有以下特点：

不可变性：RDD一旦创建就不能修改
分布式：数据分布在集群的多个节点上
容错性：通过血缘关系（Lineage）实现故障恢复
惰性求值：只有在执行Action操作时才会真正计算

Spark组件

Spark Core：

提供基础功能和RDD API
任务调度和内存管理
故障恢复机制

Spark SQL：

结构化数据处理
支持SQL查询和DataFrame API
与Hive兼容

Spark Streaming：

实时流数据处理
微批处理模型
与Kafka、Flume等集成

MLlib：

机器学习算法库
分类、回归、聚类算法
特征工程和模型评估

GraphX：

图计算引擎
图算法库
图数据并行计算

使用场景

1. 大数据批处理

处理大规模离线数据，如ETL作业、数据清洗和转换。

2. 实时流处理

处理实时数据流，如日志分析、实时监控和预警系统。

3. 交互式数据分析

通过Spark SQL进行快速的数据探索和分析。

4. 机器学习

利用MLlib进行大规模机器学习模型训练和预测。

5. 图计算

分析社交网络、推荐系统等图结构数据。

具体案例

案例1：WordCount批处理

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

object WordCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("WordCount")
    val sc = new SparkContext(conf)
    
    // 读取文本文件
    val textFile = sc.textFile("hdfs://namenode:9000/input/text")
    
    // WordCount处理
    val wordCounts = textFile
      .flatMap(line => line.split(" "))  // 分割单词
      .filter(_.nonEmpty)                // 过滤空字符串
      .map(word => (word, 1))           // 映射为(word, 1)
      .reduceByKey(_ + _)               // 按key聚合
      .sortBy(_._2, false)              // 按计数降序排列
    
    // 保存结果
    wordCounts.saveAsTextFile("hdfs://namenode:9000/output/wordcount")
    
    sc.stop()
  }
}

# Python版本
from pyspark import SparkConf, SparkContext

conf = SparkConf().setAppName("WordCount")
sc = SparkContext(conf=conf)

# 读取文本文件
text_file = sc.textFile("hdfs://namenode:9000/input/text")

# WordCount处理
word_counts = (text_file
    .flatMap(lambda line: line.split(" "))
    .filter(lambda word: word != "")
    .map(lambda word: (word, 1))
    .reduceByKey(lambda a, b: a + b)
    .sortBy(lambda x: x[1], False))

# 保存结果
word_counts.saveAsTextFile("hdfs://namenode:9000/output/wordcount")

sc.stop()

案例2：实时流处理

import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.serialization.StringDeserializer

object KafkaStreamingExample {
  def main(args: Array[String]): Unit = {
    val ssc = new StreamingContext(sparkConf, Seconds(10))
    
    // Kafka配置
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "spark_streaming_group",
      "auto.offset.reset" -> "latest"
    )
    
    val topics = Array("user_events")
    
    // 创建Kafka流
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    
    // 处理用户事件
    val userEvents = stream.map(record => record.value())
      .flatMap(_.split(","))
      .map(event => {
        val parts = event.split(":")
        (parts(0), parts(1).toInt) // (userId, eventCount)
      })
      .reduceByKeyAndWindow(
        (a: Int, b: Int) => a + b,
        (a: Int, b: Int) => a - b,
        Seconds(60),  // 窗口大小
        Seconds(10)   // 滑动间隔
      )
    
    // 输出结果
    userEvents.foreachRDD { rdd =>
      val topUsers = rdd.sortBy(_._2, false).take(10)
      topUsers.foreach { case (userId, count) =>
        println(s"User: $userId, Events: $count")
      }
    }
    
    ssc.start()
    ssc.awaitTermination()
  }
}

案例3：机器学习示例

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
import org.apache.spark.sql.SparkSession

object TextClassification {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("TextClassification")
      .getOrCreate()
    
    import spark.implicits._
    
    // 创建训练数据
    val training = spark.createDataFrame(Seq(
      (0L, "spark is great", 1.0),
      (1L, "hadoop is good", 1.0),
      (2L, "this is bad", 0.0),
      (3L, "terrible experience", 0.0)
    )).toDF("id", "text", "label")
    
    // 创建机器学习管道
    val tokenizer = new Tokenizer()
      .setInputCol("text")
      .setOutputCol("words")
    
    val hashingTF = new HashingTF()
      .setNumFeatures(1000)
      .setInputCol(tokenizer.getOutputCol)
      .setOutputCol("features")
    
    val lr = new LogisticRegression()
      .setMaxIter(10)
      .setRegParam(0.001)
    
    val pipeline = new Pipeline()
      .setStages(Array(tokenizer, hashingTF, lr))
    
    // 训练模型
    val model = pipeline.fit(training)
    
    // 创建测试数据
    val test = spark.createDataFrame(Seq(
      (4L, "spark ml is awesome"),
      (5L, "this is horrible"),
      (6L, "great data processing")
    )).toDF("id", "text")
    
    // 进行预测
    val predictions = model.transform(test)
    predictions.select("id", "text", "probability", "prediction")
      .show(false)
    
    spark.stop()
  }
}

案例4：Spark SQL数据分析

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

object SalesAnalysis {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("SalesAnalysis")
      .enableHiveSupport()
      .getOrCreate()
    
    import spark.implicits._
    
    // 读取销售数据
    val salesDF = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("hdfs://namenode:9000/data/sales.csv")
    
    // 创建临时视图
    salesDF.createOrReplaceTempView("sales")
    
    // SQL查询：按产品类别统计销售额
    val categoryStats = spark.sql("""
      SELECT 
        category,
        COUNT(*) as order_count,
        SUM(amount) as total_sales,
        AVG(amount) as avg_order_value,
        MAX(amount) as max_order
      FROM sales
      WHERE date >= '2023-01-01'
      GROUP BY category
      ORDER BY total_sales DESC
    """)
    
    categoryStats.show()
    
    // DataFrame API：计算月度趋势
    val monthlyTrend = salesDF
      .withColumn("month", date_format($"date", "yyyy-MM"))
      .groupBy("month")
      .agg(
        sum("amount").alias("monthly_sales"),
        count("*").alias("order_count"),
        countDistinct("customer_id").alias("unique_customers")
      )
      .orderBy("month")
    
    monthlyTrend.show()
    
    // 保存结果
    categoryStats
      .coalesce(1)
      .write
      .mode("overwrite")
      .parquet("hdfs://namenode:9000/output/category_stats")
    
    spark.stop()
  }
}

案例5：性能调优配置

# spark-submit提交参数
spark-submit \
  --class com.example.SparkApp \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 4g \
  --driver-cores 2 \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 10 \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.adaptive.coalescePartitions.enabled=true \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.adaptive.skewJoin.enabled=true \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.minExecutors=5 \
  --conf spark.dynamicAllocation.maxExecutors=20 \
  app.jar

// 代码中的性能优化
val spark = SparkSession.builder()
  .appName("OptimizedApp")
  .config("spark.sql.adaptive.enabled", "true")
  .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
  .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

// 缓存热点数据
val cachedDF = largeDF.cache()

// 广播小表
val broadcastDF = broadcast(smallDF)
val result = largeDF.join(broadcastDF, "key")

// 合理设置分区数
val repartitionedDF = df.repartition(200)

// 使用列式存储格式
df.write
  .mode("overwrite")
  .option("compression", "snappy")
  .parquet("output_path")

性能优化建议

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. 序列化优化

1
2
3

// 使用Kryo序列化器
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
spark.conf.set("spark.kryo.registrator", "com.example.MyKryoRegistrator")

3. 数据倾斜处理

// 加盐处理数据倾斜
val saltedDF = df.withColumn("salted_key", 
  concat($"key", lit("_"), (rand() * 100).cast("int")))

// 两阶段聚合
val stage1 = saltedDF.groupBy("salted_key").agg(sum("value").alias("partial_sum"))
val stage2 = stage1
  .withColumn("original_key", split($"salted_key", "_").getItem(0))
  .groupBy("original_key")
  .agg(sum("partial_sum").alias("final_sum"))

4. 资源配置建议

Driver配置：

CPU: 2-4核心
内存: 4-8GB
适合小到中等规模作业

Executor配置：

CPU: 4-8核心
内存: 8-16GB
数量: 根据数据量和集群资源确定

Apache Spark作为新一代大数据处理引擎，其统一的编程模型和高性能的内存计算能力使其成为现代数据处理平台的核心组件。通过合理的配置和优化，Spark可以为企业提供从批处理到流处理、从数据分析到机器学习的全方位大数据解决方案。

编外计划 - 日志

To be or not to be,--that is question.