大数据 – Spark系列《一》- 从Hadoop到Spark：大数据计算引擎的演进

一站式

一个平台或工具提供了多种功能或服务，可以满足用户的多种需求，而不需要依赖其他额外的工具或服务

MapReduce不是一站式

是因为它在处理结构化数据时需要借助第三方工具hive来进行数据管理和查询

Spark是一站式

针对结构化数据处理：由spark-sql组件处理

针对机器学习：提供了MLlib（Machine Learning Library）组件，MLlib支持高数公式和迭代式处理海量数据，实现猜你喜欢、精准广告推送

针对图关系处理需求：使用Spark GraphX图库，可以处理大规模的图数据，支持图算法的应用，例如社交网络分析、网络流量分析等。

针对实时统计需求： Spark提供了Spark Streaming组件，可以处理实时数据流，支持实时数据的处理和统计分析，例如实时监控、实时报警等应用场景。

1.3 🐶Spark特性

1. 🥙通用性

Spark针对不同的处理场景有不同的解决方，包括Spark SQL（处理结构化数据）、Spark ML（机器学习）、Spark GraphX（图处理）、Spark Streaming（实时数据流处理）等，使其适用于各种不同类型的数据处理需求。

2. 🥙简洁灵活

Spark的上层API丰富多样，包括map、flatMap、groupBy、reduce等，这些API设计简洁灵活，使得用户可以方便地进行数据转换和计算，同时也支持函数式编程的思想，提高了代码的可读性和易维护性。

3. 🥙多语言

Spark支持多种编程语言，包括Scala、Java、Python和R等，这使得用户可以使用自己熟悉的编程语言来开发Spark应用程序，提高了开发效率并降低了学习成本。

1.4 🐶Spark Core编程体验

1.4.1 spark开发工程搭建

1. 🥙开发语言选择：

Spark提供多种语言的开发API，包括Java、Scala、Python等。
Scala是Spark自身用于开发的语言，其语法与Spark提供的数据计算API非常相似，因此在开发Spark数据运算程序时，通常优先选择Scala语言。

2. 🥙依赖管理工具：

目前主流的依赖管理工具包括Maven、Sbt等。
Maven是最流行和通用的依赖管理工具之一，因此在开发Spark项目时常使用Maven进行依赖管理、编译、打包等处理。

<!-- 在pom.xml里面添加如下依赖 -->

   <!-- 属性列表 -->
    <properties>
        <maven.compiler.source>8</maven.compiler.source> <!-- Java编译器源代码版本为8 -->
        <maven.compiler.target>8</maven.compiler.target> <!-- Java编译器目标字节码版本为8 -->
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spark.version>3.2.3</spark.version> <!-- Spark框架版本为3.2.3 -->
        <scala.version>2.12.12</scala.version> <!-- Scala版本为2.12.12 -->
    </properties>

    <dependencies>
        <!-- scala的依赖 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.10</version>
        </dependency>
        <!-- spark core 即为spark内核 ，其他高级组件都要依赖spark core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>${spark.version}</version>
        </dependency>
    </dependencies>

    <!-- 依赖下载国内镜像库，如果之前在maven里面配置了则不需要复制-->
    <repositories>
        <repository>
            <id>nexus-aliyun</id>
            <name>Nexus aliyun</name>
            <layout>default</layout>
            <url>http://maven.aliyun.com/nexus/content/groups/public</url>
            <snapshots>
                <enabled>false</enabled>
                <updatePolicy>never</updatePolicy>
            </snapshots>
            <releases>
                <enabled>true</enabled>
                <updatePolicy>never</updatePolicy>
            </releases>
        </repository>
    </repositories>

    <!-- maven插件下载国内镜像库，如果之前在maven里面配置了则不需要复制 -->
    <pluginRepositories>
        <pluginRepository>
            <id>ali-plugin</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
            <snapshots>
                <enabled>false</enabled>
                <updatePolicy>never</updatePolicy>
            </snapshots>
            <releases>
                <enabled>true</enabled>
                <updatePolicy>never</updatePolicy>
            </releases>
        </pluginRepository>
    </pluginRepositories>

    <build>
        <pluginManagement>
            <plugins>
                <!-- 编译scala的插件 -->
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                </plugin>
                <!-- 编译java的插件 -->
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.5.1</version>
                </plugin>
            </plugins>
        </pluginManagement>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>compile</phase>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <!-- 打jar插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

添加依赖后记得刷新项目

1.4.2 Spark编程流程

在使用Spark进行大数据处理时，通常会遵循以下编程流程，这一流程包括数据加载、处理转换以及输出结果等步骤。

1. 🥙获取sparkcontext对象

在Spark应用程序中，首先需要获取一个SparkContext对象，它是与集群连接的主要入口点，用于创建RDD以及在集群上执行操作。

使用setAppName方法为我们的Spark应用程序设置一个有意义的名称，以便于在Spark UI或日志中识别
可以使用setMaster方法指定定Spark应用程序的运行环境. Spark支持多种运行模式，包括本地模式(local)、YARN模式和自带的集群模式（Standalone）。

val conf = new SparkConf().setAppName("SparkExample").setMaster("local")
val sc = new SparkContext(conf)

2. 🥙加载数据

在Spark中，数据通常从外部数据源加载，比如文本文件、数据库或者HDFS中的文件。通过textFile等方法，可以将数据加载并映射成RDD分布式集合。

// 在Spark中加载数据并创建RDD
val dataRDD = sc.textFile("data/a.txt")

3. 🥙处理转换数据

接下来，我们在RDD上根据需求，调用一系列的转换算子（transformation），对数据进行处理和转换，以得到结果RDD。转换算子可以对RDD进行各种操作，如过滤、映射、聚合等。

// 在RDD上调用转换算子进行数据处理
val transformedRDD = dataRDD
  .flatMap(line => line.split(" ")) // 将每行数据按空格分割成单词
  .map(word => (word, 1)) // 将单词映射成键值对 (word, 1)
  .reduceByKey(_ + _) // 根据单词进行聚合统计

4. 🥙输出结果,释放资源

最后，在结果RDD上调用动作算子（action），触发计算执行，并输出最终结果。动作算子会触发Spark作业的执行，将计算结果返回给驱动程序或者保存到外部存储中。

// 在结果RDD上调用动作算子输出结果
transformedRDD.saveAsTextFile("data/wordCount_output")

同时，在程序结束时，需要释放资源，关闭SparkContext对象。

sc.stop()

1.4.3 简单代码实现-wordCount

//数据：a.txt
a a a a a a a a a a
b b b b b
c c c c c c
d d d d  e e e f f f f f g  g g g h
a a a a a a a a a a
b b b b b
c c c c c c
d d d d  e e e f f f f f g  g g g h
a a a a a a a a a a
b b b b b
c c c c c c
d d d d  e e e f f f f f g  g g g h
a a a a a a a a a a
b b b b b
c c c c c c
d d d d  e e e f f f f f g  g g g h
a a a a a a a a a a
b b b b b
c c c c c c
d d d d  e e e f f f f f g  g g g h

package com.doit.day0130

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

/**
 * @日期: 2024/1/31
 * @Author: Wang NaPao
 * @Blog: https://blog.csdn.net/weixin_40968325?spm=1018.2226.3001.5343
 * @Tips: 和我一起学习吧
 * @Description:
 */


object WordCountExample {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf对象，并设置应用程序名称和运行模式
    val conf = new SparkConf()
      .setAppName("Starting...")
      .setMaster("local")
    // 创建SparkContext对象，并传入SparkConf对象
    val sc = new SparkContext(conf)

    //使用SparkContext的textFile方法读取输入文件，得到一个RDD。
    val rdd1 = sc.textFile("data/a.txt")

    //对RDD进行转换操作，将每一行按空格分割成单词，并将每个单词映射为(key, value)对，其中value为1。
    val rdd2: RDD[(String, Int)] = rdd1.flatMap(line => {
      val arr = line.split("\s+")
      val tuples = arr.map(tp => (tp, 1))
      tuples
    })
    //对转换后的RDD进行分组操作，按照单词进行分组
    val rdd3 = rdd2.groupBy(_._1)

    //对分组后的RDD进行映射操作，将每个单词和对应的出现次数映射为(key, value)对。
    val rdd4 = rdd3.map(tp => {
      (tp._1, tp._2.size)
    })

    //使用saveAsTextFile方法将结果保存到输出文件中。
    rdd4.saveAsTextFile("data/wordCount_output")

    //停止SparkContext对象。
    sc.stop()
  }
}

结果：