RDD的好处

1

data_from_file=sc.textFile('/Users/drabast/VS14MORT.txt.gz',4)

Schema

RDD是无schema的数据结构.因此在以下的代码片段中的并行数据集,通过Spark使用RDD非常适用:

1

data_heterogenous=sc.parallelize([('Ferrrari','fast'),{'Porsche':10000},['Spain','visited',4504]]).collect()

从文件读取

从文本文件读取数据时,文件中的每一行形成了RDD的一个元素.
data_from_file.take(1)命令输出一个元素

全局作用域和局部作用域

Spark有两种模式下运行:本地的和集群的.本地运行你的Spark代码时,和你目前的python没有什么不同:相比任何其他部分来说,变化最有可能是语法上的,只是加上了一个交织的部分,数据和代码在不同的工作者进程之间复制.

转换

转换可以调整数据集.包括映射、筛选、连接、转换数据集中的值.

.map(….)转换

该方法应用在每个RDD元素上:在data_from_file_conv的数据集情况里,你可以认为这是每一行的转换

1

data_2014=data_from_file_conv.map(lambda row:int(row[16]))

.filter(…)转换

另外一个经常需要适用的转换方法是.filter(….),该方法可以让你从数据集中选择元素,该元素符合特定的标准。

.flatMap(..)

.flatMap(…)方法和.Map()方法的工作类似.但是.flatMap()返回一个扁平的结果,而不是一个列表。

.distinct(…)转换

该方法返回指定列中不同的列表.如果逆向知道你的数据集或者验证这个数据集，.distinct(…)方法会非常有用

.sample(…)转换

.sample(….)方法返回数据集的随机样本.第一个参数指定采样是否应该替换,第二个参数定义返回数据的分数(指数学中的分数),第三个参数是伪随机数产生器的种子.

1
2

fraction=0.1
data_sample=data_from_file_conv.sample(False,fraction,666)

.leftOuterJoin(…)转换

.leftOuterJoin(….),就像在SQL中一样,根据两个数据集中都有的值来连接两个RDD,并返回左侧的RDD记录,而右边的记录附加在两个RDD匹配的地方:

1
2
3

rdd1=sc.parallelize([('a',1),('b',4),('c',10)])
rdd2=sc.parallelize([('a',4),('a',1),('b','6'),('d',15)])
rdd3=rdd1.leftOuterJoin(rdd2)

.repartition(…)转换

重新对数据集进行分区,改变了数据集分区的数量.此功能应该谨慎适用,因为他会重组数据,导致对性能方面产生巨大的影响.

操作

和转换不同,操作执行数据集上的计划任务.一旦完成数据转换,则可以执行相应转换.这可能不包含转换(例如,take(n)会仅仅返回n条来自RDD的记录,即使你没有对RDD做任何转换)或者直接执行一系列转换.

.take()方法

这可以说是最有用的方法(也是用的最多的方法,如.map(…)方法).该方法优于.collect(..),因为它只返回单个数据分区的前n行,对比之下,.collect(…)返回的是整个RDD.处理大数据集时,这个区别很重要.

1

data_take_sampled=data_from_file_conv.take(1)

1

data_take_sampled=data_from_file_conv.takeSample(False,1,667)

.collect()方法

将所有RDD的元素返回给驱动程序

.reduce方法

.reduce(..)方法使用指定的方法减少RDD中的元素.
你可以利用该方法计算RDD总的元素数量

1

rdd1.map(lambda row:row[1]).reduce(lambda x,y:x+y)

.count(..)方法

.count(…)统计出了RDD里得元素数量

.saveAsTextFile(…)方法

在RDD执行.saveAsTextFile(..)可以让RDD保存为文本文件:每个文件一个分区.

.foreach(…)方法

这个方法对RDD里得每个元素,用迭代得方式应用相同的函数;和.map(…)比较，.foreach(…)方法按照一个接一个的方式,对每一条记录应用一个定义好的函数。当您希望将数据保存到PySpark本身不支持的数据库时,该方法很有用.

总结

转换的作用是从现有数据集创建新数据集.转换是惰性的,因为它们仅在动作需要将结果返回到驱动程序时才计算。

Transformation用于对RDD的创建，RDD只能使用Transformation创建，同时还提供大量操作方法，包括map，filter，groupBy，join等，RDD利用这些操作生成新的RDD，但是需要注意，无论多少次Transformation，在RDD中真正数据计算Action之前都不可能真正运行。

Action是数据执行部分，其通过执行count，reduce，collect等方法真正执行数据的计算部分。实际上，RDD中所有的操作都是Lazy模式进行，运行在编译中不会立即计算最终结果，而是记住所有操作步骤和方法，只有显示的遇到启动命令才执行。这样做的好处在于大部分前期工作在Transformation时已经完成，当Action工作时，只需要利用全部自由完成业务的核心工作。

Spark资源调度和任务调度过程介绍

参考文献： https://blog.csdn.net/helloxiaozhe/article/details/81533438

Spark资源调度和任务调度的流程:

1. 启动集群后,Worker节点会向Master节点汇报资源情况,Master掌握了集群资源情况
2. 当Spark提交一个Application后,根据RDD之间的依赖关系将Application形成一个DAG(有向无环图).任务提交后,Spark会在Driver端创建两个对象:DAGScheduler和TaskScheduler。
3. DAGScheduler是任务调度的高层调度器,是一个对象.DAGScheduler的主要作用就是将DAG根据RDD之间的宽窄依赖关系划分为一个个的Stage,然后将这些Stage以TaskSet的形式提交给TaskScheduler(TaskScheduler是任务调度的低层调度器,这里TaskSet其实就是一个集合,里面封装的就是一个个的task任务,也就是stage中的并行度task任务)
4. TaskSchedule会遍历TaskSet集合，拿到每个task后会将task发送到计算节点Executor中去执行（其实就是发送到Executor中的线程池ThreadPool去执行）。
5. task在Executor线程池中的运行情况会向TaskScheduler反馈，
6. 当task执行失败时，则由TaskScheduler负责重试，将task重新发送给Executor去执行，默认重试3次。如果重试3次依然失败，那么这个task所在的stage就失败了。
7. stage失败了则由DAGScheduler来负责重试，重新发送TaskSet到TaskSchdeuler，Stage默认重试4次。如果重试4次以后依然失败，那么这个job就失败了。job失败了，Application就失败了。
8. TaskScheduler不仅能重试失败的task,还会重试straggling（落后，缓慢）task（也就是执行速度比其他task慢太多的task）。如果有运行缓慢的task那么TaskScheduler会启动一个新的task来与这个运行缓慢的task执行相同的处理逻辑。两个task哪个先执行完，就以哪个task的执行结果为准。这就是Spark的推测执行机制。在Spark中推测执行默认是关闭的。推测执行可以通过spark.speculation属性来配置。
   总结:
9. 对于ETL类型要入数据库的业务要关闭推测执行机制，这样就不会有重复的数据入库。
10. 如果遇到数据倾斜的情况，开启推测执行则有可能导致一直会有task重新启动处理相同的逻辑，任务可能一直处于处理不完的状态。（所以一般关闭推测执行）
11. 一个job中多个action， 就会有多个job，一般一个action对应一个job,如果一个application中有多个job时，按照顺序一次执行，即使后面的失败了，前面的执行完了就完了，不会回滚。
12. 有SparkContext端就是Driver端。

粗粒度资源申请和细粒度资源申请

13. 粗粒度资源申请(Spark）:在Application执行之前，将所有的资源申请完毕，当资源申请成功后，才会进行任务的调度，当所有的task执行完成后，才会释放这部分资源。
    优点:在Application执行之前，所有的资源都申请完毕，每一个task运行时直接使用资源就可以了，不需要task运行时在执行前自己去申请资源，task启动就快了，task执行快了，stage执行就快了，job就快了，application执行就快了。
    缺点:直到最后一个task执行完成才会释放资源，集群的资源无法充分利用。当数据倾斜时更严重。
14. 细粒度资源申请（MapReduce）
    Application执行之前不需要先去申请资源，而是直接执行，让job中的每一个task在执行前自己去申请资源，task执行完成就释放资源。
    优点：集群的资源可以充分利用。
    缺点：task自己去申请资源，task启动变慢，Application的运行就相应的变慢了。

DataFrame

DataFrame也是一种不可变的分布式数据集,这种数据集被组织成指定的列,类似于关系数据库的表。Spark DataFrame是一个近似的概念,即允许用户轻松地使用结构化数据(如数据表).不过也存在一些差异.
通过在分布式数据集上施加结构,让Spark用户利用Spark SQL来查询结构化的数据或使用Spark表达式方法(而不是lambda).

Python到RDD之间的通信

每当使用RDD执行PySpark程序时,潜在地需要巨大地开销来执行作业。在PySpark驱动器中,Spark Context通过Py4j启动一个使用JavaSparkContext的JVM。所有的RDD转换最初都映射到java中的PythonRDD对象.

一旦这些任务被推送到Spark工作节点,PythonRDD对象就使用管道(pipe)启动Python的子进程,发送代码和数据到Python中进行处理:
虽然该方法允许PySpark将数据处理分布到多个工作节点的多个Python子进程中,Python和JVM之间还是有很多上下文切换和通信开销。

利用DataFrame加速PySpark

引入DataFrame之前,Python查询速度普遍比使用RDD的Scala查询慢(后者快两倍)。通常情况,这种查询性能的降低源于Python和JVM之间的通信开销

常见DataFrame

通常情况下,通过使用SparkSession导入数据(或调用PySpark的shell脚本Spark)来创建DataFrame

生成自己的JSON数据

1	stringJSONRDD=sc.parallelize(("""{"id":"123","name":"Katie","age":19,"eyeColor":"brown"}""","""{"id":"234","name":"Michael","age":22,"eyeColor":"green"}"""))

现在已经创建了RDD,利用SparkSession read.json方法,RDD将会转换成一个DataFrame(即Spark.read.json(….)).还可以利用.createOrReplaceTempView方法创建一个临时表


1 2	swimmersJson=spark.read.json(stringJSONRDD) swimmersJson.createOrReplaceTempView("swimmersJson")#创建临时表

许多的RDD操作都是有相关转换的,直到行为操作执行,这些RDD操作都不会被执行
重要的是注意parallelize、map和mapPartitions都是RDD的转换而来.spark.read.json包裹在DataFrame中,不仅仅是RDD转换,还是RDD转换成DataFrame的行动.
注意:创建临时表是一次DataFrame转换,并且只有直到执行DataFrame动作时,创建临时表才会被执行.

简单的DataFrame查询

.show()

swimmersJson.show()
默认显示前10行

SQL查询

如果你愿意编写SQL语句,则可以编写以下查询


1	spark.sql("select * from swimmersJson").collect()

使用.collect()方法,返回行对象列表所有的记录.也可以使用.show()方法

RDD的交互操作

有两种从RDD变换到DataFrame(或者Dataset[T])的不同方法:使用反射推断模式或以编程方式指定模式.

使用反射来推断模式

来建立DataFrame和运行查询的过程中,我们略过了DataFrame的模式是自动定义的这一事实.最初,行对象通过传递一列键/值对作为行类的**kwargs来构造.然后,SparkSQL将行对象的RDD转变为一个DataFrame,在DataFrame中键就是列,数据类型通过采用数据来推断.

在开始创建swimmersJSON DataFrame之后,没有指定模式,你会注意到利用printSchema()方法的模式定义


1 2	#打印模式 swimmersJson.printSchema()

编写指定模式

在这种情况下,我们通过Spark SQL中引入数据类型(pyspark.sql.types),以编程方式来指定模式,并生成一些.csv数据


1 2 3 4 5 6

from pyspark.sql.types import * stringCSVRDD=sc.parallelize([ (123,'Katie',19,'brown'), (234,'Michael',22,'green'), (345,'Simone',23,'blue') ])

首先,我们根据以下的[schema]变量将模式编码成一个字符串.然后我们会利用StructType和StructField定义模式.


1 2 3 4 5 6

schema=StructType([ StructField("id",LongType(),True), StructField("name",StringType(),True), StructField("age",LongType(),True), StructField("eyeColor",StringType(),True) ])

注意,StructField被分解为以下方面:

15. name:该字段的名字
16. dataType:该字段的数据类型
17. nullable:指示此字段的值是否为空

1 2 3 4

#对RDD应用该模式并创建DataFrame swimmers=spark.createDataFrame(stringCSVRDD,scheam) #利用DataFrame创建一个临时视图 swimmers.createOrReplaceTempView("swimmers")

利用DataFrame API查询

行数

得到DataFrame中的行数


1	swimmers.count()

运行筛选语句

运行一个筛选语句,可以使用filter子句(筛选子句);在下面的代码段中,我们使用select子句来指定要返回的列


1 2 3 4

#获取age22的id swimmers.select("id","age").filter("age=22").show() #上述查询的另外一种方法如下: swimmers.select(swimmers.id,swimmers.age).filter(swimmers.age==22).show()

利用SQL查询

行数

1	spark.sql('select count(1) from swimmers').show()

利用where子句运行筛选语句

1	spark.sql("select id,age from swimmers where age=22").show()

和DataFrame API查询一样,如果指示想要取回眼睛颜色以字母b开头的游泳运动员的名字,我们还可以使用like语法:


1 2	spark.sql( "select name,eyeColor from swimmers where eyeColor like 'b%'").show()

1. 1. 初识Spark
   1. 1.1. Spark作业和API
      1. 1.1.1. 执行过程
      2. 1.1.2. Pyspark原理
         1. 1.1.2.1. 大致选择
   2. 1.2. 弹性分布式数据集(RDD)
      1. 1.2.1. RDD特性
      2. 1.2.2. RDD的好处
      3. 1.2.3. RDD编程接口
   3. 1.3. DataFrame
   4. 1.4. Dataset
   5. 1.5. Catalyst优化器
   6. 1.6. RDD、DataFrame和Dataset怎么选择还好
      1. 1.6.1. RDD
      2. 1.6.2. DataFrame
      3. 1.6.3. Dataset
      4. 1.6.4. 总结：Dataset/DataFrame/RDD的区别
2. 2. 弹性分布式数据集(RDD)
   1. 2.1. 创建RDD
      1. 2.1.1. Schema
      2. 2.1.2. 从文件读取
   2. 2.2. 全局作用域和局部作用域
   3. 2.3. 转换
      1. 2.3.1. .map(….)转换
      2. 2.3.2. .filter(…)转换
      3. 2.3.3. .flatMap(..)
      4. 2.3.4. .distinct(…)转换
      5. 2.3.5. .sample(…)转换
      6. 2.3.6. .leftOuterJoin(…)转换
      7. 2.3.7. .repartition(…)转换
   4. 2.4. 操作
      1. 2.4.1. .take()方法
      2. 2.4.2. .collect()方法
      3. 2.4.3. .reduce方法
      4. 2.4.4. .count(..)方法
      5. 2.4.5. .saveAsTextFile(…)方法
      6. 2.4.6. .foreach(…)方法
   5. 2.5. 总结
   6. 2.6. Spark资源调度和任务调度过程介绍
      1. 2.6.1. Spark资源调度和任务调度的流程:
3. 3. DataFrame
   1. 3.1. Python到RDD之间的通信
   2. 3.2. 利用DataFrame加速PySpark
   3. 3.3. 常见DataFrame
      1. 3.3.1. 生成自己的JSON数据
   4. 3.4. 简单的DataFrame查询
      1. 3.4.1. .show()
      2. 3.4.2. SQL查询
   5. 3.5. RDD的交互操作
      1. 3.5.1. 使用反射来推断模式
      2. 3.5.2. 编写指定模式
   6. 3.6. 利用DataFrame API查询
      1. 3.6.1. 行数
      2. 3.6.2. 运行筛选语句
   7. 3.7. 利用SQL查询
      1. 3.7.1. 行数
      2. 3.7.2. 利用where子句运行筛选语句