分享

全面学习Spark包括:基础解析、概述、集群安装、开发等

问题导读:


1、什么是 Spark?
2、Spark 的用户和用途有哪些?
3、Spark 如何进行集群安装?
4、如何再Spark shell 中编写 WordCount 程序?

5、如何在 IDEA 中远程调试 WordCount 程序?



第1章 Spark 概述
1.1 什么是 Spark
  官网:http://spark.apache.org
   EkKeeg.png
  Spark 的产生背景
   EkKVOS.png
  Spark 是一种快速、通用、可扩展的大数据分析引擎,2009 年诞生于加州大学伯克利分校 AMPLab,2010 年开源,2013 年 6 月成为 Apache 孵化项目,2014 年 2 月成为 Apache 顶级项目。项目是用 Scala 进行编写。
  目前,Spark生态系统已经发展成为一个包含多个子项目的集合,其中包含 SparkSQL、Spark Streaming、GraphX、MLib、SparkR 等子项目,Spark 是基于内存计算的大数据并行计算框架。除了扩展了广泛使用的 MapReduce 计算模型,而且高效地支持更多计算模式,包括交互式查询和流处理。Spark 适用于各种各样原先需要多种不同的分布式平台的场景,包括批处理、迭代算法、交互式查询、流处理。通过在一个统一的框架下支持这些不同的计算,Spark 使我们可以简单而低耗地把各种处理流程整合在一起。而这样的组合,在实际的数据分析过程中是很有意义的。不仅如此,Spark 的这种特性还大大减轻了原先需要对各种平台分别管理的负担。
  大一统的软件栈,各个组件关系密切并且可以相互调用,这种设计有几个好处:
  1、软件栈中所有的程序库和高级组件都可以从下层的改进中获益。
  2、运行整个软件栈的代价变小了。不需要运行 5 到 10 套独立的软件系统了,一个机构只需要运行一套软件系统即可。系统的部署、维护、测试、支持等大大缩减。
  3、能够构建出无缝整合不同处理模型的应用。

  Spark 的内置项目如下:
   EkKEy8.png
  Spark Core:实现了 Spark 的基本功能,包含任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark Core 中还包含了对弹性分布式数据集(resilient distributed dataset,简称RDD)的 API 定义。
  Spark SQL:是 Spark 用来操作结构化数据的程序包。通过 Spark SQL,我们可以使用 SQL 或者 Apache Hive 版本的 SQL 方言(HQL)来查询数据。Spark SQL 支持多种数据源,比 如 Hive 表、Parquet 以及 JSON 等。
  Spark Streaming:是 Spark 提供的对实时数据进行流式计算的组件。提供了用来操作数据流的 API,并且与 Spark Core 中的 RDD API 高度对应。
  Spark MLlib:提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等,还提供了模型评估、数据导入等额外的支持功能。
  集群管理器:Spark 设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求,同时获得最大灵活性,Spark 支持在各种集群管理器(cluster manager)上运行,包括 Hadoop YARN、Apache Mesos,以及 Spark 自带的一个简易调度器,叫作独立调度器。
  Spark 得到了众多大数据公司的支持,这些公司包括 Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿里、腾讯、京东、携程、优酷土豆。当前百度的 Spark 已应用于凤巢、大搜索、直达号、百度大数据等业务;阿里利用 GraphX 构建了大规模的图计算和图挖掘系统,实现了很多生产系统的推荐算法;腾讯 Spark 集群达到 8000 台的规模,是当前已知的世界上最大的 Spark 集群。

1.2 Spark 特点



与 Hadoop 的 MapReduce 相比,Spark 基于内存的运算要快 100 倍以上,基于硬盘的运算也要快 10 倍以上。Spark 实现了高效的 DAG 执行引擎,可以通过基于内存来高效处理数据流。计算的中间结果是存在于内存中的。
EkKAQf.png

  • 易用

Spark 支持 Java、Python、R 和 Scala 的 API,还支持超过 80 种高级算法,使用户可以快速构建不同的应用。而且 Spark 支持交互式的 Python、R 和 Scala 的 shell,可以非常方便地在这些 shell 中使用 Spark 集群来验证解决问题的方法。
EkKiWt.png

  • 通用

Spark 提供了统一的解决方案。Spark 可以用于批处理、交互式查询(Spark SQL)、实时流处理(Spark Streaming)、机器学习(Spark MLlib)和图计算(GraphX)。这些不同类型的处理都可以在同一个应用中无缝使用。Spark 统一的解决方案非常具有吸引力,毕竟任何公司都想用统一的平台去处理遇到的问题,减少开发和维护的人力成本和部署平台的物力成本。
EkKkSP.png

  • 兼容性

Spark 可以非常方便地与其他的开源产品进行融合。比如,Spark 可以使用 Hadoop 的 YARN 和 Apache Mesos 作为它的资源管理和调度器器,并且可以处理所有 Hadoop 支持的数据,包括 HDFS、HBase 和 Cassandra 等。这对于已经部署 Hadoop 集群的用户特别重要,因为不需要做任何数据迁移就可以使用 Spark 的强大处理能力。Spark 也可以不依赖于第三方的资源管理和调度器,它实现了 Standalone 作为其内置的资源管理和调度框架,这样进一步降低了 Spark 的使用门槛,使得所有人都可以非常容易地部署和使用 Spark。此外,Spark 还提供了在 EC2 上部署 Standalone 的 Spark 集群的工具。
EkKmwQ.png

1.3 Spark 的用户和用途
  我们大致把 Spark 的用例分为两类:数据科学应用和数据处理应用。也就对应的有两种人群:数据科学家和工程师。
  数据科学任务
  主要是数据分析领域,数据科学家要负责分析数据并建模,具备 SQL、统计、预测建模(机器学习)等方面的经验,以及一定的使用 Python、Matlab 或 R 语言进行编程的能力。
  数据处理应用
  工程师定义为使用 Spark 开发生产环境中的数据处理应用的软件开发者,通过对接 Spark 的 API 实现对处理的处理和转换等任务。

第2章 Spark 集群安装
2.1 集群角色
EkKKFs.png

  从物理部署层面上来看,Spark 主要分为两种类型的节点,Master 节点和 Worker 节点,Master 节点主要运行集群管理器的中心化部分,所承载的作用是分配 Application 到 Worker 节点,维护 Worker 节点 的 Driver、Application 的状态。Worker 节点负责具体的业务运行。
  从 Spark 程序运行的层面来看,Spark 主要分为驱动器节点和执行器节点。

2.2 机器准备
  准备两台以上 Linux 服务器,安装好 JDK1.8。
2.3 下载 Spark 安装包

EkKQWq.png

Step0、使用下载命令
[mw_shl_code=shell,true]wget 下载地址
[/mw_shl_code]
Step1、上传 spark-2.1.1-bin-hadoop2.7.tgz 安装包到 Linux 对应的目录上,本人是上传至 /opt/software 目录下
Step2、解压安装包到指定位置
[mw_shl_code=shell,true]tar -zxf /opt/software/spark-2.1.1-bin-hadoop2.7.tgz -C /opt/module
[/mw_shl_code]
如下图所示:
EkKMYn.png


2.4 配置 Spark Standalone 模式
  Spark 的部署模式有Local、Local-Cluster、Standalone、Yarn、Mesos,我们选择最具代表性的 Standalone 集群部署模式。

Step1、进入到 Spark 安装目录中的配置目录 conf
[mw_shl_code=shell,true]cd /opt/module/spark-2.1.1-bin-hadoop2.7/conf
[/mw_shl_code]
如下图所示:
EkK3lV.png

Step2、将 slaves.template 复制为 slaves
Step3、将 spark-env.sh.template 复制为 spark-env.sh
EkKGOU.png

Step4、修改 slaves 文件,将 Worker 的 hostname 输入:
EkK1S0.png

Step5、修改 spark-env.sh 文件,添加如下配置:
SPARK_MASTER_HOST=hadoop102
SPARK_MASTER_PORT=7077

EkK8yT.png

Step6、将配置好的 Spark 文件拷贝到其他节点上 或者 使用配置分发的脚本
[mw_shl_code=shell,true]scp -r /opt/module/spark-2.1.1-bin-hadoop2.7/ atguigu@hadoop103:/opt/module/
scp -r /opt/module/spark-2.1.1-bin-hadoop2.7/ atguigu@hadoop104:/opt/module/

或者
xsync /opt/module/spark-2.1.1-bin-hadoop2.7/
[/mw_shl_code]
Step7、Spark 集群配置完毕,目前是 1 个 Master,2 个 Work,hadoop102 上启动 Spark 集群
[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/sbin/start-all.sh
[/mw_shl_code]
如下图所示:
EkKwf1.png

启动后执行 jps 命令,主节点上有 Master 进程,其他子节点上有 Worker 进行
登录 Spark 管理界面查看集群状态(主节点):http://hadoop102:8080/ 或者 http://192.168.25.102:8080/
EkKYmF.png

到此为止,Spark 集群安装完毕。

问题1:如果遇到 “JAVA_HOME not set” 异常,如下图所示:
EkKNTJ.png

解决方案:可以在 sbin 目录下的 spark-config.sh 文件中加入如下配置,然后配置分发到其他机器:
[mw_shl_code=shell,true]export JAVA_HOME=/opt/module/jdk1.8.0_144
[/mw_shl_code]
如下图所示:
EkKtw4.png


问题2:如果遇到 Hadoop HDFS 的写入权限异常:
[mw_shl_code=text,true]org.apache.hadoop.security.AccessControlException
[/mw_shl_code]
解决方案: 在 hdfs-site.xml 中添加如下配置,关闭权限验证,然后配置分发到其他机器:

[mw_shl_code=xml,true]    <property>
        <name>dfs.permissions</name>
        <value>false</value>
    </property>  
[/mw_shl_code]
2.5 配置 Spark History Server
Step1、进入到 Spark 安装目录
[mw_shl_code=shell,true]cd /opt/module/spark-2.1.1-bin-hadoop2.7/conf
[/mw_shl_code]
Step2、将 spark-default.conf.template 复制为 spark-default.conf
[mw_shl_code=shell,true]$ cp spark-defaults.conf.template spark-defaults.conf
[/mw_shl_code]
Step3、修改 spark-default.conf 文件,开启 Log:
[mw_shl_code=text,true]spark.master                     spark://hadoop102:7077
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://hadoop102:9000/directory
[/mw_shl_code]
如下图所示:
EkKak9.png

Step4、修改 spark-env.sh 文件,添加如下配置:
[mw_shl_code=shell,true]export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=4000
-Dspark.history.retainedApplications=3
-Dspark.history.fs.logDirectory=hdfs://hadoop102:9000/directory"
[/mw_shl_code]
如下图所示:
EkKdYR.png

Step5、启动 HDFS 集群,在 HDFS 上创建好你所指定的 eventLog 日志目录。
[mw_shl_code=shell,true]
$ /opt/module/hadoop-2.7.2/bin/hdfs dfs -mkdir -p /directory
[/mw_shl_code]
参数描述:
[mw_shl_code=text,true]spark.eventLog.dir      Application 在运行过程中所有的信息均记录在该属性指定的路径下

spark.history.ui.port=4000      调整 WEBUI 访问的端口号为 4000
spark.history.retainedApplications=3        指定保存 Application 历史记录的个数,如果超过这个值,旧的应用程序信息将被删除,这个是内存中的应用数,而不是页面上显示的应用数
spark.history.fs.logDirectory=hdfs://hadoop102:9000/directory       配置了该属性后,在 start-history-server.sh 时就无需再显式的指定路径,Spark History Server 页面只展示该指定路径下的信息
[/mw_shl_code]
Step6、将配置好的 Spark 文件拷贝到其他节点上或者配置分发。
Step7、重启 Spark 集群。
[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/sbin/start-all.sh
[/mw_shl_code]
Step8、启动后执行历史服务器。
[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/sbin/start-history-server.sh
[/mw_shl_code]
网页上查看
EkKDl6.png

到此为止,Spark History Server 安装完毕。

2.6 配置 Spark HA
集群部署完了,但是有一个很大的问题,那就是 Master 节点存在单点故障,要解决此问题,就要借助 zookeeper,并且启动至少两个 Master 节点来实现高可靠,配置方式比较简单:
EkKr6K.png

Step1、Spark 集群规划:hadoop102,hadoop103 是 Master;hadoop103,hadoop104 是 Worker。
Step2、安装配置 Zookeeper 集群,并启动 Zookeeper 集群。
Step3、停止 spark 所有服务,在 hadoop102 节点上修改配置文件 spark-env.sh,在该配置文件中删掉 SPARK_MASTER_IP(即 SPARK_MASTER_HOST) 并添加如下配置:
[mw_shl_code=shell,true]export SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.url=hadoop102:2181,hadoop103:2181,hadoop104:2181
-Dspark.deploy.zookeeper.dir=/spark"
[/mw_shl_code]
如下图所示:
EkKBSx.png

Step4、在 hadoop102 节点上修改 slaves 配置文件内容指定 worker 节点。
[mw_shl_code=text,true]hadoop103
hadoop104
[/mw_shl_code]
Step5、将配置文件同步到所有节点。
Step6、在 hadoop102 上执行 sbin/start-all.sh 脚本,启动集群并启动第一个 master 节点,然后在 hadoop103 上执行 sbin/start-master.sh 启动第二个 master 节点。
Step7、程序中 spark 集群的访问地址需要改成:
[mw_shl_code=text,true]--master spark://hadoop102:7077,hadoop103:7077
[/mw_shl_code]
我们干掉 hadoop102 上的 Master 进程,然后再次执行 WordCount 程序,看是否能够执行成功:
EkKsOO.png

由上图可知,程序依旧可以运行。
同理:我们再干掉 hadoop103 上的 Master 进程,然后再次执行 WordCount 程序,看是否能够执行成功,经过测试,程序依旧可以执行成功,到此为止,Spark 的高可用完成!

Step8、我们想知道 Zookeeper 中保存了什么?

[mw_shl_code=shell,true][atguigu@hadoop102 zookeeper-3.4.10]$ pwd
/opt/module/zookeeper-3.4.10
[atguigu@hadoop102 zookeeper-3.4.10]$ bin/zkCli.sh -server hadoop102:2181,hadoop103:2181,hadoop104:2181
Connecting to hadoop102:2181,hadoop103:2181,hadoop104:2181
......
......
[zk: hadoop102:2181,hadoop103:2181,hadoop104:2181(CONNECTED) 0] ls /spark
[leader_election, master_status]
[zk: hadoop102:2181,hadoop103:2181,hadoop104:2181(CONNECTED) 1] get /spark/master_status
192.168.25.102
cZxid = 0x4000000059
ctime = Mon Apr 22 10:10:11 CST 2019
mZxid = 0x4000000059
mtime = Mon Apr 22 10:10:11 CST 2019
pZxid = 0x4000000063
cversion = 3
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 14
numChildren = 3
[zk: hadoop102:2181,hadoop103:2181,hadoop104:2181(CONNECTED) 2]
[/mw_shl_code]
2.7 配置 Spark Yarn 模式
Step1、修改 hadoop 配置下的 /opt/module/hadoop-2.7.2/etc/hadoop/yarn-site.xml 文件,然后分发到其他节点。
yarn-site.xml

[mw_shl_code=xml,true]<?xml version="1.0"?>
<configuration>
    <!-- Reducer获取数据的方式 -->
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
    </property>

    <!-- 指定YARN的ResourceManager的地址 -->
    <property>
        <name>yarn.resourcemanager.hostname</name>
        <value>hadoop103</value>
    </property>

    <!-- 日志聚集功能使能 -->
    <property>
        <name>yarn.log-aggregation-enable</name>
        <value>true</value>
    </property>

    <!-- 日志保留时间设置7天 -->
    <property>
        <name>yarn.log-aggregation.retain-seconds</name>
        <value>604800</value>
    </property>

    <!-- 任务历史服务器 -->
    <property>
        <name>yarn.log.server.url</name>
        <value>http://hadoop102:19888/jobhistory/logs/</value>
    </property>

    <!-- 指定yarn在启动的时候的内存大小 -->
    <property>
        <name>yarn.scheduler.maximum-allocation-mb</name>
        <value>2048</value>
    </property>
    <property>
        <name>yarn.scheduler.minimum-allocation-mb</name>
        <value>2048</value>
    </property>
    <property>
        <name>yarn.nodemanager.vmem-pmem-ratio</name>
        <value>2.1</value>
    </property>
    <property>
        <name>mapred.child.java.opts</name>
        <value>-Xmx1024m</value>
    </property>

    <!--是否启动一个线程检查每个任务正使用的物理内存量,如果任务超出分配值,则直接将其杀掉,默认是 true,实际开发中设置成 true,学习阶段设置成 false -->
    <property>
        <name>yarn.nodemanager.pmem-check-enabled</name>
        <value>false</value>
    </property>
    <!--是否启动一个线程检查每个任务正使用的虚拟内存量,如果任务超出分配值,则直接将其杀掉,默认是 true,实际开发中设置成 true,学习阶段设置成 false -->
    <property>
        <name>yarn.nodemanager.vmem-check-enabled</name>
        <value>false</value>
    </property>
</configuration>
[/mw_shl_code]
Step2、修改 /opt/module/spark-2.1.1-bin-hadoop2.7/conf/spark-env.sh,添加以下内容,然后分发到其他节点。
spark-env.sh

[mw_shl_code=bash,true]# 让 spark 能够发现 hadoop 的配置文件
HADOOP_CONF_DIR=/opt/module/hadoop-2.7.2/etc/hadoop
YARN_CONF_DIR=/opt/module/hadoop-2.7.2/etc/hadoop
[/mw_shl_code]
如下图所示:

EkK6mD.png


Step3、提交应用进行测即可
[mw_shl_code=bash,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \
--class com.atguigu.sparkdemo.WordCountDemo \
--master yarn \
--deploy-mode client \
--executor-memory 1G \
--total-executor-cores 2 \
/opt/software/sparkdemo-1.0-SNAPSHOT-jar-with-dependencies.jar \
hdfs://hadoop102:9000/RELEASE \
hdfs://hadoop102:9000/out

或者

$ /opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \
--class com.atguigu.sparkdemo.WordCountDemo \
--master yarn-client \
--executor-memory 1G \
--total-executor-cores 2 \
/opt/software/sparkdemo-1.0-SNAPSHOT-jar-with-dependencies.jar \
hdfs://hadoop102:9000/RELEASE \
hdfs://hadoop102:9000/out
[/mw_shl_code]
第3章 执行 Spark 程序
3.1 执行第一个 spark 程序

[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
/opt/module/spark-2.1.1-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.1.1.jar \
100
[/mw_shl_code]
参数说明:

[mw_shl_code=text,true]--master spark://hadoop102:7077     指定 Master 的地址
--executor-memory 1G                指定每个 executor 可用内存为 1G
--total-executor-cores 2            指定每个 executor 使用的 cup 核数为 2 个
[/mw_shl_code]
该算法是利用蒙特·卡罗算法求 PI,结果如下图:
EkKc0e.png

网页上查看 History Server
EkKgTH.png


3.2 Spark 应用提交
一旦打包好,就可以使用 bin/spark-submit 脚本启动应用了。 这个脚本负责设置 spark 使用的 classpath 和依赖,支持不同类型的集群管理器和发布模式:

[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \
--class <main-class> \
--master <master-url> \
--deploy-mode <deploy-mode> \
--conf <key>=<value> \
... # other options
<application-jar> \
[application-arguments]
[/mw_shl_code]
一些常用选项:
[mw_shl_code=text,true]--class: 你的应用的启动类 (如 org.apache.spark.examples.SparkPi)。
--master: 集群的 master URL (如 spark://192.168.25.102:7077)。
--deploy-mode: 是否发布你的驱动到 Worker 节点(cluster) 或者作为一个本地客户端 client)(默认是 client)。
--conf: 任意的 Spark 配置属性, 格式 key=value,如果值包含空格,可以加引号 "key=value",缺省的 Spark 配置。
application-jar: 打包好的应用 jar,包含依赖,这个 URL 在集群中全局可见。 比如 hdfs://共享存储系统, 如果是 file://path, 那么所有的节点的 path 都包含同样的 jar。
application-arguments: 传给 main() 方法的参数。[/mw_shl_code]
–master 后面的 URL 可以是以下格式:
EkK4pt.png

查看 Spark-submit 全部参数:
EkKRkd.png


3.3 Spark shell
&#8195;&#8195;spark-shell 是 Spark 自带的交互式 Shell 程序,方便用户进行交互式编程,用户可以在该命令行下用 scala 编写 spark 程序。

3.3.1 启动 Spark shell
启动 spark shell 时没有指定 master 地址
[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-shell
[/mw_shl_code]
启动 spark shell 时指定 master 地址
[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-shell \
--master spark://hadoop102:7077 \
--executor-memory 2G \
--total-executor-cores 2
[/mw_shl_code]
注意1:如果启动 spark shell 时没有指定 master 地址,但是也可以正常启动 spark shell 和执行 spark shell 中的程序,其实是启动了 spark 的 cluster 模式,如果 spark 是单节点,并且没有指定 slave 文件,这个时候如果打开 spark-shell 默认是 local 模式。
&#8195;&#8195;Local 模式是 master 和 worker 在同同一进程内。
&#8195;&#8195;Cluster 模式是 master 和 worker 在不同进程内。
注意2:Spark Shell 中已经默认将 SparkContext 类初始化为对象 sc。用户代码如果需要用到,则直接应用 sc 即可。
EkKffI.png


3.3.2 在 Spark shell 中编写 WordCount 程序
Step1、首先启动 HDFS,在 HDFS 上创建一个 /RELEASE 目录
[mw_shl_code=shell,true]$ /opt/module/hadoop-2.7.2/bin/hdfs dfs -mkdir -p /RELEASE
[/mw_shl_code]
Step2、将 Spark 目录下的 RELEASE 文件上传一个文件到:hdfs://hadoop102:9000/RELEASE 上
[mw_shl_code=shell,true]$ /opt/module/hadoop-2.7.2/bin/hdfs dfs -put /opt/module/spark-2.1.1-bin-hadoop2.7/RELEASE /RELEASE
[/mw_shl_code]
如下图所示:
EkKWtA.png

Step3、在 Spark shell 中用 scala 语言编写 spark 程序
[mw_shl_code=shell,true]scala> sc.textFile("hdfs://hadoop102:9000/RELEASE/RELEASE").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://hadoop102:9000/out")
[/mw_shl_code]
如下图所示:
EkK51P.png

Step4、使用 hdfs 命令查看结果
[mw_shl_code=shell,true]$ /opt/module/hadoop-2.7.2/bin/hdfs dfs -cat hdfs://hadoop102:9000/out/p*
[/mw_shl_code]
如下图所示:
EkKI6f.png

说明:
[mw_shl_code=text,true]sc 是 SparkContext 对象,该对象是提交 spark 程序的入口。
textFile(hdfs://hadoop102:9000/RELEASE/RELEASE)     是 hdfs 中读取数据
flatMap(_.split(" "))   先 map 在压平
map((_,1))              将单词和1构成元组
reduceByKey(_+_)        按照 key 进行 reduce,并将 value 累加
saveAsTextFile("hdfs://hadoop102:9000/out")         将结果写入到 hdfs 中
[/mw_shl_code]
如下图所示:
EkKoX8.png


3.4 在 IDEA 中编写 WordCount 程序
spark shell 仅在测试和验证我们的程序时使用的较多,在生产环境中,通常会在 IDE 中编制程序,然后打成 jar 包,然后提交到集群,最常用的是创建一个 Maven 项目,利用 Maven 来管理 jar 包的依赖。
Step1、创建一个项目
Step2、选择 Maven 项目,然后点击 next
Step3、填写 maven 的 GAV,然后点击 next
Step4、填写项目名称,然后点击 finish
Step5、创建好 maven 项目后,点击 Enable Auto-Import
Step6、配置 Maven 的 pom.xml

[mw_shl_code=xml,true]<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>sparkdemo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <scala.version>2.11.8</scala.version>
        <spark.version>2.1.1</spark.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.sparkdemo.WordCountDemo</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>
[/mw_shl_code]
Step7、将 src/main/scala 设置成源代码目录。
Step8、添加 IDEA Scala(执行此操作后,pom 文件中不用添加 scala 依赖,因为已经以 lib 库的方式加入)
EkKH0g.png

选择要添加的模块
EkKLkj.png

Step9、新建一个 Scala class,类型为 Object
EkK7nS.png

Step10、编写 spark 程序
示例代码如下:
[mw_shl_code=scala,true]package com.atguigu.sparkdemo

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

object WordCountDemo {
  val logger = LoggerFactory.getLogger(WordCountDemo.getClass)

  def main(args: Array[String]): Unit = {
    // 创建 SparkConf() 并设置 App 名称
    val sparkConf = new SparkConf().setAppName("WC")
    // 创建 SparkContext,该对象是提交 Spark App 的入口
    val sc = new SparkContext(sparkConf)
    // 使用 sc 创建 RDD 并执行相应的 transformation 和 action
    sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _, 1).sortBy(_._2, false).saveAsTextFile(args(1))
    // 停止 sc,结束该任务
    logger.info("complete!")
    sc.stop()
  }
}
[/mw_shl_code]
Step11、使用 Maven 打包:首先修改 pom.xml 中的 main class
EkKb7Q.png

Step12、点击 idea 右侧的 Maven Project 选项,点击 “闪电”图表,表示跳过测试,然后点击 Lifecycle,再分别双击 clean 和 package
EkKvpq.png

Step13、选择编译成功的 jar 包,并将该 jar 上传到 Spark 集群中的某个节点上
EkKOts.png

Step14、首先启动 hdfs 和 Spark 集群
启动 hdfs
[mw_shl_code=shell,true]/opt/module/hadoop-2.7.3/sbin/start-dfs.sh
[/mw_shl_code]
启动 spark
[mw_shl_code=shell,true]/opt/module/spark-2.1.1-bin-hadoop2.7/sbin/start-all.sh
[/mw_shl_code]
Step15、使用 spark-submit 命令提交 Spark 应用(注意参数的顺序)
[mw_shl_code=shell,true]$ /opt/module/spark-2.1.1-bin-hadoop2.7/bin/spark-submit \
--class com.atguigu.sparkdemo.WordCountDemo \
--master spark://hadoop102:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
/opt/software/sparkdemo-1.0-SNAPSHOT-jar-with-dependencies.jar \
hdfs://hadoop102:9000/RELEASE \
hdfs://hadoop102:9000/out1
[/mw_shl_code]
Step16、查看程序执行结果
[mw_shl_code=shell,true]$ /opt/module/hadoop-2.7.2/bin/hdfs dfs -cat hdfs://hadoop102:9000/out1/p*
[/mw_shl_code]
如下图所示:
EkKx10.png

–master 后面跟的参数小结:
EkKXhn.png

传入参数说明:
[mw_shl_code=text,true]hdfs://hadoop102:9000/RELEASE           输入文件路径
hdfs://hadoop102:9000/out1              输出文件路径

如果在 spark 程序中写死了这两处路径,则这两个参数就不需要了。
[/mw_shl_code]
3.5 在 IDEA 中本地调试 WordCount 程序
本地 Spark 程序调试需要使用 local 提交模式,即将本机当做运行环境,Master 和 Worker 都为本机。运行时直接加断点调试即可。如下:
EkMSXT.png

如果本机操作系统是 windows,如果在程序中使用了 hadoop 相关的东西,比如写入文件到 HDFS,则会遇到如下异常:
EkKzcV.png

出现这个问题的原因,并不是程序的错误。在 windows 下调试 spark 的时候,用到了 hadoop 相关的服务。

解决办法1:本项目生效,是将一个 hadoop 相关的服务 zip 包(hadoop-common-bin-2.7.3-x64.zip)解压到任意目录。
点击 Run -> Run Configurations
然后在 IDEA 中配置 Run Configuration,添加 HADOOP_HOME 变量即可:
EkMCBF.png


解决办法2:所有项目生效,windows 系统中配置 hadoop 的环境变量,如下图所示:
EkM9nU.png


3.6 在 IDEA 中远程调试 WordCount 程序
通过 IDEA 进行远程调试,主要是将 IDEA 作为 Driver 来提交应用程序,配置过程如下:
修改 sparkConf,添加最终需要运行的 Jar 包、Driver 程序的地址,并设置 Master 的提交地址:
EkMP74.png


3.7 Spark 核心概念
&#8195;&#8195;每个 Spark 应用都由一个驱动器程序(driver program)来发起集群上的各种并行操作。驱动器程序包含应用的 main 函数,并且定义了集群上的分布式数据集,还对这些分布式数据集应用了相关操作。
&#8195;&#8195;驱动器程序通过一个 SparkContext 对象来访问 Spark。这个对象代表对计算集群的一个连接。shell 启动时已经自动创建了一个 SparkContext 对象,是一个叫作 sc 的变量。
&#8195;&#8195;驱动器程序一般要管理多个执行器(executor)节点。
&#8195;&#8195; EkMFAJ.png








最新经典文章,欢迎关注公众号

来源:CSDN

作者:Lan&Jun

原文:《大数据技术之_19_Spark学习_01_Spark 基础解析 + Spark 概述 + Spark 集群安装 + 执行 Spark 程序》

https://blog.csdn.net/u012990179/article/details/89434309


已有(2)人评论

跳转到指定楼层
Robert_V1Q99 发表于 2019-4-26 09:54:12
谢谢分享知识
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条