如何创建sparkcontext，最近在学习pyspark？
Spark提供了一个Python_Shell，即pyspark，从而可以以交互的方式使用Python编写Spark程序。
有关Spark的基本架构介绍参考http://blog.csdn.net/cymy001/article/details/78483614；
有关Pyspark的环境配置参考http://blog.csdn.net/cymy001/article/details/78430892。
pyspark里最核心的模块是SparkContext（简称sc）,最重要的数据载体是RDD。RDD就像一个NumPy array或者一个Pandas Series，可以视作一个有序的item集合。只不过这些item并不存在driver端的内存里，而是被分割成很多个partitions，每个partition的数据存在集群的executor的内存中。
引入Python中pyspark工作模块
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
#任何Spark程序都是SparkContext开始的，SparkContext的初始化需要一个SparkConf对象，SparkConf包含了Spark集群配置的各种参数(比如主节点的URL)。初始化后，就可以使用SparkContext对象所包含的各种方法来创建和操作RDD和共享变量。Spark shell会自动初始化一个SparkContext(在Scala和Python下可以，但不支持Java)。
#getOrCreate表明可以视情况新建session或利用已有的session
1
2
3
4
5
6
7
SparkSession是Spark 2.0引入的新概念。SparkSession为用户提供了统一的切入点，来让用户学习spark的各项功能。 在spark的早期版本中，SparkContext是spark的主要切入点，由于RDD是主要的API，我们通过sparkcontext来创建和操作RDD。对于每个其他的API，我们需要使用不同的context。例如，对于Streming，我们需要使用StreamingContext；对于sql，使用sqlContext；对于hive，使用hiveContext。但是随着DataSet和DataFrame的API逐渐成为标准的API，就需要为他们建立接入点。所以在spark2.0中，引入SparkSession作为DataSet和DataFrame API的切入点。SparkSession实质上是SQLContext和HiveContext的组合(未来可能还会加上StreamingContext)，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了SparkContext，所以计算实际上是由SparkContext完成的。
初始化RDD的方法
（1）本地内存中已经有一份序列数据(比如python的list)，可以通过sc.parallelize去初始化一个RDD。当执行这个操作以后，list中的元素将被自动分块(partitioned)，并且把每一块送到集群上的不同机器上。
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
#（a）利用list创建一个RDD;使用sc.parallelize可以把Python list，NumPy array或者Pandas Series,Pandas DataFrame转成Spark RDD。
rdd = sc.parallelize([1,2,3,4,5])
rdd
#Output:ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:480
#（b）getNumPartitions()方法查看list被分成了几部分
rdd.getNumPartitions()
#Output:4
1
2
3
4
5
6
7
8
9
10
11
12
13
14
#（c）glom().collect()查看分区状况
rdd.glom().collect()
#Output:[[1], [2], [3], [4, 5]]
1
2
3
在这个例子中，是一个4-core的CPU笔记本;Spark创建了4个executor，然后把数据分成4个块。colloect()方法很危险，数据量上BT文件读入会爆掉内存……
（2）创建RDD的另一个方法是直接把文本读到RDD。文本的每一行都会被当做一个item，不过需要注意的一点是，Spark一般默认给定的路径是指向HDFS的，如果要从本地读取文件的话，给一个file://开头（windows下是以file:\\开头）的全局路径。
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
#（a）记录当前pyspark工作环境位置
import os
cwd=os.getcwd()
cwd
#Output:'C:\\Users\\Yu\\0JulyLearn\\5weekhadoopspark'
#（b）要读入的文件的全路径
rdd=sc.textFile("file:\\\\\\" + cwd + "\\names\yob1880.txt")
rdd
#Output:file:\\\C:\Users\Yu\0JulyLearn\5weekhadoopspark\names\yob1880.txt MapPartitionsRDD[3] at textFile at NativeMethodAccessorImpl.java:0
#（c）first()方法取读入的rdd数据第一个item
rdd.first()
#Output:'Mary,F,7065'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
甚至可以sc.wholeTextFiles读入整个文件夹的所有文件。但是要特别注意，这种读法，RDD中的每个item实际上是一个形如(文件名，文件所有内容)的元组。读入整个文件夹的所有文件。
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
#记录当前pyspark工作环境位置
import os
cwd=os.getcwd()
cwd
#Output:'C:\\Users\\Yu\\0JulyLearn\\5weekhadoopspark'
rdd = sc.wholeTextFiles("file:\\\\\\" + cwd + "\\names\yob1880.txt")
rdd
#Output:org.apache.spark.api.java.JavaPairRDD@12bcc15
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
rdd.first()
Output:
('file:/C:/Users/Yu/0JulyLearn/5weekhadoopspark/names/yob1880.txt',
1
2
3
4
5
其余初始化RDD的方法，包括：HDFS上的文件，Hive中的数据库与表，Spark SQL得到的结果。这里暂时不做介绍。
RDD Transformation
（1）RDDs可以进行一系列的变换得到新的RDD，有点类似列表推导式的操作，先给出一些RDD上最常用到的transformation：
map() 对RDD的每一个item都执行同一个操作
flatMap() 对RDD中的item执行同一个操作以后得到一个list，然后以平铺的方式把这些list里所有的结果组成新的list
filter() 筛选出来满足条件的item
distinct() 对RDD中的item去重
sample() 从RDD中的item中采样一部分出来，有放回或者无放回
sortBy() 对RDD中的item进行排序
1
2
3
4
5
6
如果想看操作后的结果，可以用一个叫做collect()的action把所有的item转成一个Python list。数据量大时，collect()很危险……
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
numbersRDD = sc.parallelize(range(1,10+1))
print(numbersRDD.collect())
#map()对RDD的每一个item都执行同一个操作
squaresRDD = numbersRDD.map(lambda x: x**2) # Square every number
print(squaresRDD.collect())
#filter()筛选出来满足条件的item
filteredRDD = numbersRDD.filter(lambda x: x % 2 == 0) # Only the evens
print(filteredRDD.collect())
#Output:
#[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
#[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
#[2, 4, 6, 8, 10]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
#flatMap() 对RDD中的item执行同一个操作以后得到一个list，然后以平铺的方式把这些list里所有的结果组成新的list
sentencesRDD=sc.parallelize(['Hello world','My name is Patrick'])
wordsRDD=sentencesRDD.flatMap(lambda sentence: sentence.split(" "))
print(wordsRDD.collect())
print(wordsRDD.count())
#Output:
#['Hello', 'world', 'My', 'name', 'is', 'Patrick']
#6
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
对比一下：
这里如果使用map的结果是[[‘Hello’, ‘world’], [‘My’, ‘name’, ‘is’, ‘Patrick’]]，
使用flatmap的结果是全部展开[‘Hello’, ‘world’, ‘My’, ‘name’, ‘is’, ‘Patrick’]。
flatmap即对应Python里的如下操作：
l = ['Hello world', 'My name is Patrick']
ll = []
for sentence in l:
ll = ll + sentence.split(" ") #+号作用,two list拼接
ll
1
2
3
4
5
（2）最开始列出的各个Transformation，可以一个接一个地串联使用，比如:
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
def doubleIfOdd(x):
if x % 2 == 1:
return 2 * x
else:
return x
numbersRDD = sc.parallelize(range(1,10+1))
resultRDD = (numbersRDD
.map(doubleIfOdd) #map,filter,distinct()
.filter(lambda x: x > 6)
.distinct()) #distinct()对RDD中的item去重
resultRDD.collect()
#Output:[8, 10, 18, 14]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
（3）当遇到更复杂的结构，比如被称作“pair RDDs”的以元组形式组织的k-v对（key, value），Spark中针对这种item结构的数据，定义了一些transform和action:
reduceByKey(): 对所有有着相同key的items执行reduce操作
groupByKey(): 返回类似(key, listOfValues)元组的RDD，后面的value List 是同一个key下面的
sortByKey(): 按照key排序
countByKey(): 按照key去对item个数进行统计
collectAsMap(): 和collect有些类似，但是返回的是k-v的字典
1
2
3
4
5
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
rdd=sc.parallelize(["Hello hello", "Hello New York", "York says hello"])
resultRDD=(rdd
.flatMap(lambda sentence:sentence.split(" "))
.map(lambda word:word.lower())
.map(lambda word:(word, 1)) #将word映射成(word,1)
.reduceByKey(lambda x, y: x + y)) #reduceByKey对所有有着相同key的items执行reduce操作
resultRDD.collect()
#Output:[('hello', 4), ('york', 2), ('says', 1), ('new', 1)]
result = resultRDD.collectAsMap() #collectAsMap类似collect,以k-v字典的形式返回
result
#Output:{'hello': 4, 'new': 1, 'says': 1, 'york': 2}
resultRDD.sortByKey(ascending=True).take(2) #sortByKey按键排序
#Output:[('hello', 4), ('new', 1)]
#取出现频次最高的2个词
print(resultRDD
.sortBy(lambda x: x[1], ascending=False)
.take(2))
#Output:[('hello', 4), ('york', 2)]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
RDD间的操作
（1）如果有2个RDD，可以通过下面这些操作，对它们进行集合运算得到1个新的RDD
rdd1.union(rdd2): 所有rdd1和rdd2中的item组合（并集）
rdd1.intersection(rdd2): rdd1 和 rdd2的交集
rdd1.substract(rdd2): 所有在rdd1中但不在rdd2中的item（差集）
rdd1.cartesian(rdd2): rdd1 和 rdd2中所有的元素笛卡尔乘积（正交和）
1
2
3
4
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
#初始化两个RDD
numbersRDD = sc.parallelize([1,2,3])
moreNumbersRDD = sc.parallelize([2,3,4])
1
2
3
4
5
6
7
8
9
numbersRDD.union(moreNumbersRDD).collect() #union()取并集
#Output:[1, 2, 3, 2, 3, 4]
numbersRDD.intersection(moreNumbersRDD).collect() #intersection()取交集
#Output:[2, 3]
numbersRDD.subtract(moreNumbersRDD).collect() #substract()取差集
#Output:[1]
numbersRDD.cartesian(moreNumbersRDD).collect() #cartesian()取笛卡尔积
#Output:[(1, 2), (1, 3), (1, 4), (2, 2), (2, 3), (2, 4), (3, 2), (3, 3), (3, 4)]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
（2）在给定2个RDD后，可以通过一个类似SQL的方式去join它们
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
# Home of different people
homesRDD = sc.parallelize([
('Brussels', 'John'),
('Brussels', 'Jack'),
('Leuven', 'Jane'),
('Antwerp', 'Jill'),
])
# Quality of life index for various cities
lifeQualityRDD = sc.parallelize([
('Brussels', 10),
('Antwerp', 7),
('RestOfFlanders', 5),
])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
homesRDD.join(lifeQualityRDD).collect() #join
#Output:
#[('Antwerp', ('Jill', 7)),
# ('Brussels', ('John', 10)),
# ('Brussels', ('Jack', 10))]
homesRDD.leftOuterJoin(lifeQualityRDD).collect() #leftOuterJoin
#Output:
#[('Antwerp', ('Jill', 7)),
# ('Leuven', ('Jane', None)),
# ('Brussels', ('John', 10)),
# ('Brussels', ('Jack', 10))]
homesRDD.rightOuterJoin(lifeQualityRDD).collect() #rightOuterJoin
#Output:
#[('Antwerp', ('Jill', 7)),
# ('RestOfFlanders', (None, 5)),
# ('Brussels', ('John', 10)),
# ('Brussels', ('Jack', 10))]
homesRDD.cogroup(lifeQualityRDD).collect() #cogroup
#Output:
#[('Antwerp',
# (<pyspark.resultiterable.ResultIterable at 0x73d2d68>,
# <pyspark.resultiterable.ResultIterable at 0x73d2940>)),
# ('RestOfFlanders',
# (<pyspark.resultiterable.ResultIterable at 0x73d2828>,
# <pyspark.resultiterable.ResultIterable at 0x73d2b70>)),
# ('Leuven',
# (<pyspark.resultiterable.ResultIterable at 0x73d26a0>,
# <pyspark.resultiterable.ResultIterable at 0x7410a58>)),
# ('Brussels',
# (<pyspark.resultiterable.ResultIterable at 0x73d2b38>,
# <pyspark.resultiterable.ResultIterable at 0x74106a0>))]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# Oops! Those <ResultIterable>s are Spark's way of returning a list
# that we can walk over, without materializing the list.
# Let's materialize the lists to make the above more readable:
(homesRDD
.cogroup(lifeQualityRDD)
.map(lambda x:(x[0], (list(x[1][0]), list(x[1][1]))))
.collect())
#Output:
#[('Antwerp', (['Jill'], [7])),
# ('RestOfFlanders', ([], [5])),
# ('Leuven', (['Jane'], [])),
# ('Brussels', (['John', 'Jack'], [10]))]
1
2
3
4
5
6
7
8
9
10
11
12
13
惰性计算，actions方法
特别注意：Spark的一个核心概念是惰性计算。当你把一个RDD转换成另一个的时候，这个转换不会立即生效执行！！！Spark会把它先记在心里，等到真的有actions需要取转换结果时，才会重新组织transformations(因为可能有一连串的变换)。这样可以避免不必要的中间结果存储和通信。
常见的action如下，当它们出现的时候，表明需要执行上面定义过的transform了:
collect(): 计算所有的items并返回所有的结果到driver端，接着 collect()会以Python list的形式返回结果
first(): 和上面是类似的，不过只返回第1个item
take(n): 类似，但是返回n个item
count(): 计算RDD中item的个数
top(n): 返回头n个items，按照自然结果排序
reduce(): 对RDD中的items做聚合
1
2
3
4
5
6
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
rdd = sc.parallelize(range(1,10+1))
rdd.reduce(lambda x, y: x + y) #reduce(): 对RDD中的items做聚合
#Output:55
1
2
3
4
5
6
7
8
9
10
reduce的原理：先在每个分区(partition)里完成reduce操作，然后再全局地进行reduce。
有时候需要重复用到某个transform序列得到的RDD结果。但是一遍遍重复计算显然是要开销的，所以我们可以通过一个叫做cache()的操作把它暂时地存储在内存中。缓存RDD结果对于重复迭代的操作非常有用，比如很多机器学习的算法，训练过程需要重复迭代。
import pyspark
from pyspark import SparkContext as sc
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)
import numpy as np
numbersRDD = sc.parallelize(np.linspace(1.0, 10.0, 10))
squaresRDD = numbersRDD.map(lambda x: x**2)
squaresRDD.cache() # Preserve the actual items of this RDD in memory
avg = squaresRDD.reduce(lambda x, y: x + y) / squaresRDD.count()
print(avg)
#Output:38.5
怎么样才能学好java编程？
我是一名从事Java领域10余年的工程师，首先很高兴能回答你的这个问题 。
一、付出努力是前提
前段时间996引起不小的风波，今日马云先生的669又再次带我开车兜了一回风。其实这背后折射出的道理都一样，不论干什么，首先都要付出超过常人的努力，才能收获幸福的果实，这是前提。除了努力，另外方向更加重要。
二、方向不对努力白费
走技术这条路线基本上薪资跟你的技术功底有很大的关系，这是发展的基础。如果你要在技术这条路上一直发展下去，可以提早可以思考一下如何构建好属于自己的一套基础技术体系。牢固的基础技术体系可以让你对技术理解更加透彻，不仅知其然也能知其所以然。新出来的功能，框架或者中间件，你也能比旁人理解更加透彻。反映到工作实践上，你能解决高难棘手的问题，或者提出非常优秀的解决方案，理论结合实战经验，不断提升你的整体水平，这些都会是非常宝贵的经验，当然这些积累都会成为你的核心竞争力。你发展的平台也更加广阔，薪水自然也能比同龄人超出好几倍。
在北上广深，如果技术能力不错，年薪60万上百万都没有问题。反之， 随着年龄增大，每天忙碌在无休止业务代码中，年复一年，你将在竞争高端岗位的时候非常吃亏。高端岗位对基础原理，理论以及重大项目的实践经验都是有很高的要求，如果你提早准备，这些都是可以逐渐积累，不至于后面追赶而感觉压力倍增。
我身边有一朋友工作将近10年，知识面其实非常广，什么高并发，分布式，中间件，性能优化等知识点了解非常多，概念性的知识娓娓道来，但是不够深入和系统性。后面想换高端岗位经常碰壁，被拒绝，非常尴尬。原因就是技能点了解很多，平时没有归纳总结，技能都是散列状的，不成体系，自问对那块领域最了解，都模棱两可，这个就非常尴尬了。
如何做到用正确的技术体系的学习呢？ 举一个简单的例子：
(1) 剥笋第一层:
public class TestInteger {
public static void main(String[] args) {
Integer n1 = 1;
Integer n2= new Integer(1);
System.out.println(n1 == n2);
}
}
有一点基础我们看出第一个就是自动装箱，其实调用的是Integer.valueOf()方法.第二个直接在堆里生成一个对象。
挺简单是吧，我们再深入一点看看下面这段程序。
(2) 剥笋第二层:
public class TestInteger {
public static void main(String[] args) {
Integer n1 = 127;
Integer n2 = 127;
Integer n3 = 128;
Integer n4 = 128;
System.out.println(n1 == n2);
System.out.println(n3 == n4);
}
}
你觉得输出结果会是什么呢？会不会是两个True?
默认情况下输出结果是：
Ture
False
因为Integer有一个内部静态类IntegerCache，默认情况会缓存-128到127之间整数，超过这个范围就会new Integer(n)，重新生成一个对象。 然后结束了吗？ 好戏还在后面。我们把知识再延伸下。
(3) 剥笋第三层：
class Product {
//库存数量
private Integer stock;
void subStock(Integer num){
synchronized(stock){
if(this.stock>num){
this.stock -=num;
}
}
}
}
在并发环境下会出现什么问题？
简单地说存在两个核心问题：
1. 锁是有可能变化的。
2. Integer做锁是不安全，也就是说不能用来做锁。
如果库存stock在-128到127默认范围之间，它是在缓存在静态常量池里面，如果被其他代码使用没有释放，你的程序永远得不到锁，无法进行减库存操作。如果超出-128到127默认范围，你的锁是变化的，它违反锁的互斥条件的原则，锁就没有意义，库存就会乱套。
这里面内容还能继续展开，涉及到的内容越来越多，越来越细，像剥笋一样，层层深入。篇幅有限，不再累赘。
以上希望对你有所启发，欢迎在评论区留言，发布自己不同的观点。
DataFrame和DataSet的区别？
官网解释：RDD：A Resilient Distributed DataSet (RDD), the basic abstraction in Spark.rdd是一个分布式的数据集，数据分散在分布式集群的各台机器上A DataFrame is equivalent to a relational table in Spark SQL, and can be created using various functions in SQLContextdataframe更像是一张关系型数据表，是一种spark独有的数据格式吧，这种格式的数据可以使用sqlcontext里面的函数
spark五大组件？
Master：管理集群和节点，不参与计算。
Worker：计算节点，进程本身不参与计算，和master汇报。
Driver：运行程序的main方法，创建sparkcontext对象。
Spark context：控制整个application的生命周期，包括DAGSchedular和TaskSchedular等组件。
Client：用户提交程序的入口
SparkSession可以替代SparkContext么？
SparkSession就是设计出来合并SparkContext和SQLContext的。
我建议能用SparkSession就尽量用。
如果发现有些API不在SparkSession中，你还是可以通过SparkSession来拿到SparkContext和SQLContex的。
