Spark 3.2.0实战：从HDFS文件读取到独立应用打包全流程（Ubuntu18.04环境）

大数据处理已成为现代技术栈的核心组成部分，而Apache Spark凭借其内存计算优势和丰富的API生态，成为分布式计算领域的标杆工具。本文将带您从零开始，在Ubuntu18.04环境下搭建完整的Spark开发环境，通过实际案例演示从基础文件操作到独立应用开发的完整工作流。无论您是刚接触大数据处理的开发者，还是需要快速验证原型的数据工程师，这套实践指南都能帮助您避开常见陷阱，快速掌握Spark核心开发技能。

1. 环境配置与基础准备

1.1 系统环境检查与依赖安装

在开始Spark之旅前，确保您的Ubuntu18.04系统已更新至最新状态：

sudo apt update && sudo apt upgrade -y

Spark运行需要Java环境支持，推荐安装OpenJDK8：

sudo apt install openjdk-8-jdk -y

验证Java安装是否成功：

java -version
# 期望输出类似：openjdk version "1.8.0_312"

1.2 Spark 3.2.0安装与配置

从Apache官网下载预编译的Spark包：

wget https://archive.apache.org/dist/spark/spark-3.2.0/spark-3.2.0-bin-hadoop3.2.tgz
tar -xzf spark-3.2.0-bin-hadoop3.2.tgz
sudo mv spark-3.2.0-bin-hadoop3.2 /usr/local/spark

配置环境变量使Spark命令全局可用：

echo 'export SPARK_HOME=/usr/local/spark' >> ~/.bashrc
echo 'export PATH=$PATH:$SPARK_HOME/bin' >> ~/.bashrc
source ~/.bashrc

验证Spark安装：

spark-shell --version
# 应显示：Spark version 3.2.0

1.3 HDFS环境准备

Spark可以与多种存储系统集成，HDFS是最常用的分布式文件系统。安装Hadoop 3.3.2：

wget https://archive.apache.org/dist/hadoop/common/hadoop-3.3.2/hadoop-3.3.2.tar.gz
tar -xzf hadoop-3.3.2.tar.gz
sudo mv hadoop-3.3.2 /usr/local/hadoop

配置Hadoop环境变量：

echo 'export HADOOP_HOME=/usr/local/hadoop' >> ~/.bashrc
echo 'export PATH=$PATH:$HADOOP_HOME/bin' >> ~/.bashrc
source ~/.bashrc

启动HDFS单节点集群：

cd /usr/local/hadoop
./bin/hdfs namenode -format
./sbin/start-dfs.sh

检查服务是否正常：

jps
# 应看到NameNode、DataNode和SecondaryNameNode进程

2. Spark基础文件操作实战

2.1 本地文件系统交互

创建测试文件并验证Spark读取：

echo -e "Spark\nHadoop\nBig Data" > ~/test.txt

启动Spark-shell进行交互式操作：

val localFile = sc.textFile("file:///home/ubuntu/test.txt")
println(s"文件行数: ${localFile.count()}")
println("文件内容:")
localFile.collect().foreach(println)

2.2 HDFS文件操作

将测试文件上传至HDFS：

hdfs dfs -mkdir -p /user/ubuntu
hdfs dfs -put ~/test.txt /user/ubuntu/

在Spark-shell中操作HDFS文件：

val hdfsFile = sc.textFile("hdfs://localhost:9000/user/ubuntu/test.txt")
val wordCounts = hdfsFile.flatMap(_.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)
wordCounts.collect().foreach(println)

2.3 性能优化技巧

对于大规模数据处理，以下配置可以显著提升性能：

// 在Spark-shell启动时添加参数
spark-shell --executor-memory 2G --driver-memory 1G

// 或在代码中配置
import org.apache.spark.SparkConf
val conf = new SparkConf()
  .set("spark.executor.memory", "2G")
  .set("spark.driver.memory", "1G")

3. 独立应用开发全流程

3.1 项目结构规划

标准的Spark Scala项目应遵循Maven目录结构：

myproject/
├── build.sbt
├── project/
│   └── build.properties
└── src/
    └── main/
        └── scala/
            └── com/
                └── example/
                    └── MyApp.scala

3.2 使用sbt构建工具

安装sbt构建工具：

echo "deb https://repo.scala-sbt.org/scalasbt/debian all main" | sudo tee /etc/apt/sources.list.d/sbt.list
echo "deb https://repo.scala-sbt.org/scalasbt/debian /" | sudo tee /etc/apt/sources.list.d/sbt_old.list
curl -sL "https://keyserver.ubuntu.com/pks/lookup?op=get&search=0x2EE0EA64E40A89B84B2DF73499E82A75642AC823" | sudo apt-key add
sudo apt update
sudo apt install sbt

创建基础的build.sbt文件：

name := "SparkAppDemo"
version := "1.0.0"
scalaVersion := "2.12.15"
libraryDependencies += "org.apache.spark" %% "spark-core" % "3.2.0"

3.3 开发WordCount应用

创建完整的单词计数应用：

// src/main/scala/com/example/WordCount.scala
package com.example

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

object WordCount {
  def main(args: Array[String]): Unit = {
    require(args.length == 2, "用法: WordCount <输入路径> <输出路径>")
    
    val conf = new SparkConf()
      .setAppName("WordCount Application")
      .setIfMissing("spark.master", "local[*]")
      
    val sc = new SparkContext(conf)
    
    try {
      val textFile = sc.textFile(args(0))
      val counts = textFile
        .flatMap(_.split("\\s+"))
        .filter(_.nonEmpty)
        .map(word => (word.toLowerCase, 1))
        .reduceByKey(_ + _)
        .sortBy(_._2, ascending = false)
      
      counts.saveAsTextFile(args(1))
      println(s"成功处理 ${counts.count()} 个不同单词")
    } finally {
      sc.stop()
    }
  }
}

3.4 打包与提交应用

使用sbt创建可部署的jar包：

sbt package

提交应用到Spark集群运行：

spark-submit \
  --class "com.example.WordCount" \
  --master local[4] \
  target/scala-2.12/sparkappdemo_2.12-1.0.0.jar \
  hdfs://localhost:9000/user/ubuntu/input.txt \
  hdfs://localhost:9000/user/ubuntu/wordcount_output

4. 高级数据处理案例

4.1 数据去重实现

开发能处理大规模数据集去重的生产级应用：

// src/main/scala/com/example/Deduplicator.scala
package com.example

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

object Deduplicator {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Data Deduplication")
    val sc = new SparkContext(conf)
    
    val inputPaths = args.init
    val outputPath = args.last
    
    val combinedRDD = inputPaths.foldLeft(sc.emptyRDD[String]) { (acc, path) =>
      acc.union(sc.textFile(path).map(_.trim).filter(_.nonEmpty))
    }
    
    val distinctRDD = combinedRDD.distinct().coalesce(1)
    
    distinctRDD.saveAsTextFile(outputPath)
    
    sc.stop()
  }
}

4.2 学生成绩分析系统

构建完整的学生成绩分析流水线：

// src/main/scala/com/example/GradeAnalyzer.scala
package com.example

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

case class GradeRecord(student: String, subject: String, score: Double)

object GradeAnalyzer {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("Grade Analysis")
      .getOrCreate()
      
    import spark.implicits._
    
    // 从多个文件读取成绩数据
    val gradeDS = spark.read
      .option("header", "false")
      .option("delimiter", " ")
      .csv(args: _*)
      .toDF("student", "score")
      .withColumn("score", $"score".cast("double"))
      .as[GradeRecord]
    
    // 计算各科平均分
    val subjectAverages = gradeDS
      .groupBy("subject")
      .avg("score")
      .withColumnRenamed("avg(score)", "average_score")
    
    // 计算学生总分排名
    val studentTotals = gradeDS
      .groupBy("student")
      .sum("score")
      .withColumnRenamed("sum(score)", "total_score")
      .orderBy($"total_score".desc)
    
    // 输出结果到不同目录
    subjectAverages.write.json(args(0) + "_subject_analysis")
    studentTotals.write.json(args(0) + "_student_ranking")
    
    spark.stop()
  }
}

4.3 性能调优实战

通过以下技术提升Spark应用性能：

配置参数优化表：

参数名推荐值说明
spark.executor.memory4G-8G根据集群资源调整
spark.driver.memory2G-4G驱动节点内存
spark.default.parallelism集群核心数×2-3控制RDD分区数量
spark.sql.shuffle.partitions200-400SQL操作的分区数

代码级优化技巧：

// 1. 合理使用缓存
val cachedRDD = expensiveTransformationRDD.persist(
  StorageLevel.MEMORY_AND_DISK_SER)

// 2. 减少shuffle操作
// 不良实践：引起两次shuffle
rdd.map(...).reduceByKey(...).groupByKey(...)

// 优化方案：组合操作
rdd.map(...).aggregateByKey(...)

// 3. 使用广播变量代替join
val smallLookupTable = spark.sparkContext.broadcast(smallDataset.collectAsMap())
largeRDD.map { case (k, v) =>
  (k, v + smallLookupTable.value.getOrElse(k, 0))
}

5. 生产环境最佳实践

5.1 日志与监控配置

配置log4j.properties文件控制日志级别：

log4j.rootCategory=WARN, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n

# 设置Spark相关日志级别
log4j.logger.org.apache.spark=WARN
log4j.logger.org.apache.hadoop=WARN
log4j.logger.io.netty=ERROR

集成Prometheus监控：

// 在SparkConf中添加监控配置
conf.set("spark.metrics.conf.*.sink.prometheus.class", 
  "org.apache.spark.metrics.sink.PrometheusSink")
conf.set("spark.metrics.conf.*.sink.prometheus.port", "9090")

5.2 错误处理与容错

健壮的生产代码应包含完善的错误处理：

object SafeSparkJob {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SafeSparkJob")
    val sc = new SparkContext(conf)
    
    try {
      val data = sc.textFile(args(0)).map { line =>
        try {
          // 解析逻辑可能抛出异常
          Some(parseLine(line))
        } catch {
          case e: Exception =>
            // 记录错误但不中断作业
            println(s"解析失败: $line, 错误: ${e.getMessage}")
            None
        }
      }.filter(_.isDefined).map(_.get)
      
      // 主处理逻辑
      processData(data)
      
    } catch {
      case e: SparkException =>
        println(s"Spark作业失败: ${e.getMessage}")
        sys.exit(1)
      case e: Exception =>
        println(s"未知错误: ${e.getMessage}")
        sys.exit(2)
    } finally {
      sc.stop()
    }
  }
  
  def parseLine(line: String): ParsedData = {...}
  def processData(data: RDD[ParsedData]): Unit = {...}
}

5.3 持续集成与部署

示例Jenkinsfile配置：

pipeline {
  agent any
  
  environment {
    SPARK_HOME = '/usr/local/spark'
    PATH = "$SPARK_HOME/bin:$PATH"
  }
  
  stages {
    stage('Build') {
      steps {
        sh 'sbt clean package'
      }
    }
    
    stage('Test') {
      steps {
        sh 'sbt test'
      }
    }
    
    stage('Deploy') {
      when {
        branch 'main'
      }
      steps {
        sh """
          spark-submit \
            --class ${params.MAIN_CLASS} \
            --master yarn \
            --deploy-mode cluster \
            target/scala-2.12/*.jar \
            ${params.JOB_ARGS}
        """
      }
    }
  }
}
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区