Spark引入与理解

Spark学习

Spark介绍

1、什么是spark
- 基于内存的计算引擎，它的计算速度非常快。但是仅仅只涉及到数据的计算，并没有涉及到数据的存储。
2、spark引入

MapReduce框架局限性
- 1，Map结果写磁盘，Reduce写HDFS，多个MR之间通过HDFS交换数据
- 2，任务调度和启动开销大
- 3，不适合迭代计算（如机器学习、图计算等等），交互式处理（数据挖掘）
- 4，不适合流式处理（点击日志分析）
- 5，MapReduce编程不够灵活，仅支持Map和Reduce两种操作
  
  Hadoop生态圈
  - 批处理：MapReduce、Hive、Pig
  - 流式计算：Storm
  - 交互式计算：Impala、presto
需要一种灵活的框架可同时进行批处理、流式计算、交互式计算
- 内存计算引擎，提供cache机制来支持需要反复迭代计算或者多次数据共享，减少数据读取的IO开销
- DAG引擎，较少多次计算之间中间结果写到HDFS的开销
- 使用多线程模型来减少task启动开销，shuffle过程中避免不必要的sort操作以及减少磁盘IO
spark的缺点是：吃内存，不太稳定，优点是速度快，spark中的job中间结果可以不落地，可以存放在内存中。

spark集群相关概念

spark集群架构(Standalone模式)
- Application
  
  用户自己写的Spark应用程序，批处理作业的集合。Application的main方法为应用程序的入口，用户通过Spark的API，定义了RDD和对RDD的操作。
- Master和Worker
  
  整个集群分为 Master 节点和 Worker 节点，相当于 Hadoop 的 Master 和 Slave 节点。
  - Master：Standalone模式中主控节点，负责接收Client提交的作业，管理Worker，并命令Worker启动Driver和Executor。
  - Worker：Standalone模式中slave节点上的守护进程，负责管理本节点的资源，定期向Master汇报心跳，接收Master的命令，启动Driver和Executor。
- Client：客户端进程，负责提交作业到Master。
- Driver：一个Spark作业运行时包括一个Driver进程，也是作业的主进程，负责作业的解析、生成Stage并调度Task到Executor上。包括DAGScheduler，TaskScheduler。
- Executor：即真正执行作业的地方，一个集群一般包含多个Executor，每个Executor接收Driver的命令Launch Task，一个Executor可以执行一到多个Task。
Spark作业相关概念
- Stage：一个Spark作业一般包含一到多个Stage。
- Task：一个Stage包含一到多个Task，通过多个Task实现并行运行的功能。
- DAGScheduler：实现将Spark作业分解成一到多个Stage，每个Stage根据RDD的Partition个数决定Task的个数，然后生成相应的Task set放到TaskScheduler中。
- TaskScheduler：实现Task分配到Executor上执行。

Spark下载与安装

下载

1	wget https://mirrors.tuna.tsinghua.edu.cn/apache/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz

解压

1	tar -zxvf spark-2.4.4-bin-hadoop2.7.tgz

环境变量配置

1
2
3

cd conf/  
cp spark-env.sh.template spark-env.sh  
vi spark-env.sh

添加如下两行（Java home 路径）：

1 2	JAVA_HOME=/usr/lib/jdk/jdk1.8.0_144 SPARK_WORKER_MEMORY=2g

vi ~/.bashrc

1 2	export SPARK_HOME=/home/parallels/app/spark-2.4.4-bin-hadoop2.7 export PATH=$SPARK_HOME/bin:$SPARK_HOME/sbin:$PATH

进入

命令行终端输入后，可以进行使用。

1	spark-shell

另外，参考Ubuntu Spark 环境搭建进行配置PATHHOME。

在PyCharm创建一个DataFrame如下

from pyspark.sql import SparkSession
from pyspark.sql import Row

spark = SparkSession.builder.appName('test').getOrCreate()
sc = spark.sparkContext
# spark.conf.set("spark.sql.shuffle.partitions", 6)
# ================直接创建==========================
l = [('Ankit',25),('Jalfaizy',22),('saurabh',20),('Bala',26)]
rdd = sc.parallelize(l)
#为数据添加列名
people = rdd.map(lambda x: Row(name=x[0], age=int(x[1])))
#创建DataFrame
schemaPeople = spark.createDataFrame(people)
schemaPeople.show()

通过csv建立

from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark import SparkConf

SPARK_APP_NAME = "dataframetest"

conf = SparkConf()
config = (
    ("spark.app.name", SPARK_APP_NAME),
    ("spark.executor.memory", "2g"),
    ("spark.executor.cores", "2"),
)

conf.setAll(config)

spark = SparkSession.builder.config(conf=conf).appName('test').getOrCreate()
sc = spark.sparkContext
# spark.conf.set("spark.sql.shuffle.partitions", 6)
# ================直接创建==========================
l = [('Ankit',25),('Jalfaizy',22),('saurabh',20),('Bala',26)]
rdd = sc.parallelize(l)
#为数据添加列名
people = rdd.map(lambda x: Row(name=x[0], age=int(x[1])))
#创建DataFrame
schemaPeople = spark.createDataFrame(people)
schemaPeople.show()

# ==================从csv读取======================
#加载csv类型的数据并转换为DataFrame
df = spark.read.format("csv").option("header", "true").load("/tmp/iris.csv")
#显示数据结构
df.printSchema()
#显示前10条数据
df.show(10)
#统计总量
print(df.count())
#列名
print(df.columns)

其中，需要先在HDFS上放置iris.csv数据。spark-env.sh、spark-dafault.sh、slave文件参考spark学习-spark安装和启动

如果出现

1	java.net.UnknownHostException

请参考启动Spark报Caused by: java.net.UnknownHostException: iotsparkmaster: unknown error

参考&致谢

https://blog.csdn.net/u010171031/article/details/51849562

https://blog.csdn.net/Enche/article/details/81142357

https://www.jianshu.com/p/4716899eba19