初识spark

◆ 注意：
? textFile 方法底层封装的是读取 MR 读取文件的方式，读取文件之前
先 split，默认 split 大小是一个 block 大小。
? RDD 实际上不存储数据，这里方便理解，暂时理解为存储数据。
? 什么是 K,V 格式的 RDD?
? 如果 RDD 里面存储的数据都是二元组对象，那么这个 RDD 我们
就叫做 K,V 格式的 RDD。
? 哪里体现 RDD 的弹性（容错）？
? partition 数量，大小没有限制，体现了 RDD 的弹性。
? RDD 之间依赖关系，可以基于上一个 RDD 重新计算出 RDD。
? 哪里体现 RDD 的分布式？
? RDD 是由 Partition 组成，partition 是分布在不同节点上的。
? RDD 提供计算最佳位置，体现了数据本地化。体现了大数据中“计
算移动数据不移动”的理念。

1. 创建 SparkConf 对象
? 可以设置 Application name。
? 可以设置运行模式及资源需求。
2. 创建 SparkContext 对象
3. 基于 Spark 的上下文创建一个 RDD，对 RDD 进行处理。
4. 应用程序中要有 Action 类算子来触发 Transformation 类算子执行。
5. 关闭 Spark 上下文对象 SparkContext。

from pyspark import SparkContext
from pyspark import SparkConf

conf = SparkConf().setAppName(‘count‘).setMaster(‘local[*]‘)
sc = SparkContext(conf=conf)
data = [‘hehe  yi  kkk nys os‘,‘ni shi  kak‘,‘na laen is lie‘]
datardd = sc.parallelize(data)
# result = datardd.map(lambda x:x.split(‘ ‘)).collect()
# print(result)
result =datardd.flatMap(lambda  x:x.split(‘ ‘)).map(lambda w:(w,1)).collect()
print(result)

? map
将一个 RDD 中的每个数据项，通过 map 中的函数映射变为一个
新的元素。
特点：输入一条，输出一条数据。

? flatMap
先 map 后 flat。与 map 类似，每个输入项可以映射为 0 到多个
输出项。

? reduceByKey
将相同的 Key 根据相应的逻辑进行处理。

? sortby

? sortByKey
作用在 K,V 格式的 RDD 上，对 key 进行升序或者降序排序。

? collect
将计算结果回收到 Driver 端。

? foreach
可以循环迭代取出各节点服务器上的内容

from pyspark import SparkContext
from pyspark import SparkConf

‘‘‘
count
‘‘‘
# 系统设置conf
conf =SparkConf().setAppName(‘count‘).setMaster(‘local‘)
context = SparkContext(conf=conf)

data = [‘zhang san  is a student ‘,‘lisi like  play basketball‘,‘wanger like durking‘,‘zl sick of study‘,‘hehe is a boy‘]

# map(lambda x:x.split(" "))


#  对数据进行序列化
datardd = context.parallelize(data)
# result = datardd.flatMap(lambda x:x.split(‘ ‘)).map(lambda w: (w,1)).reduceByKey(lambda a,b:a+b).collect()
# print(result)

# 这里用map把原来的data数据按引号划分成五个部分,通过reduceByKey来统计各部分出现的次数.
# result = datardd.map(lambda w: (w,1)).reduceByKey(lambda a,b:a+b).collect()
# print(result)

# 这里的map ,它返回的是由一个个元素组成的小列表,最终由这些小列表组成一个大列表
# result = datardd.map(lambda x:x.split(‘ ‘)).collect()
# print(result)

#这里用的是flatmp 它返回的是由一个个打散的元素组成的列表
# result = datardd.flatMap(lambda x:x.split(‘ ‘)).collect()
# print(result)


#  这里最终的打印结果用foreach来处理,它最大的好处在于:不像collect把所有的数据全部放在列表里
# 这样很容易导致内存被撑爆,所以在不知道数据量大小的情况下最好少用collect,这里就用到了foreach
# 通过遍历循环将数据结果挨个挨个打印出来.
result = datardd.flatMap(lambda x:x.split(‘ ‘))

def  f(x):
    print(x)

re = result.foreach(f)
print(re)

from pyspark import SparkContext
from pyspark import SparkConf
import pickle

conf = SparkConf().setAppName(‘count‘).setMaster(‘local[*]‘)
sc = SparkContext(conf=conf)
text_file =sc.textFile(r‘E:\Hbase\api\text_file‘)
counts = text_file.flatMap(lambda x: x.split(" "))             .map(lambda word: (word, 1))              .reduceByKey(lambda a, b: a + b).collect()
print(counts)

f = open(‘test.txt‘,‘wb‘)

pickle.dump(counts,f)

上述图的运行步骤：

Driver 与集群节点之间有频繁的通信。
Driver 负责任务(tasks)的分发和结果的回收、任务的调度。如果 task的计算结果非常大就不要回收了，会造成 OOM(内存溢出)。
Worker 是 Standalone 资源调度框架里面资源管理的从节点，也是JVM 进程。
Master 是 Standalone 资源调度框架里面资源管理的主节点。也是JVM 进程。

eOffHeap, _deserialized, _replication(默认值为1)

5.MEMORY_ONLY_SER

SER做序列化。会消耗CPU。

参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication(默认值为1)

6.MEMORY_ONLY_SER_2

参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication(默认值为1)

7.MEMORY_AND_DISK

内存中若放不下，则多出的部分放在机器的本地磁盘上，区别于MEMORY_ONLY(内存中若放不下，则多出的部分原来在哪就还在哪)

参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication(默认值为1)

8.MEMORY_AND_DISK_2

参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication(默认值为1)

9.MEMORY_AND_DISK_SER

参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication(默认值为1)

10.MEMORY_AND_DISK_SER_2

参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication(默认值为1)

11.OFF_HEAP(不使用堆，比如可以使用Tachyon)

参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication(默认值为1)

SPARK_MASTER_IP:master 的 ip
SPARK_MASTER_PORT:提交任务的端口，默认是 7077
SPARK_WORKER_CORES：每个 worker 从节点能够支配的 core 的个
数
SPARK_WORKER_MEMORY:每个 worker 从节点能够支配的内存数