写在前面

系统为ubuntu, spark为pyspark

一. 简单配置和读取txt，并打印

这里我们定义一个任务：

从txt中读取文件，并打印文件的每一行

from pyspark import SparkConf, SparkContext
import os
# 这里配置spark对用的python版本，如果版本不一致就会报错
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"# 配置单机模式
conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)# 文件路径，这个文件里的内容如下
"""
a b c
a c
d a
"""
path = 'file:///home/wgq/learn_spark/test.txt'# 读取文件
data = sc.textFile(path, 2)# 如果是从Python列表里读取
"""
arr = ['a b c', 'a c', 'd a']
# 读取文件
data = sc.parallelize(arr)
"""# 打印每一行
data.foreach(print)"""
输出：
a b c
a c
d a
"""

二. filter操作

定义任务：

从txt中读取文件，并且过滤掉没有c字母的行

from pyspark import SparkConf, SparkContext
import os
# 这里配置spark对用的python版本，如果版本不一致就会报错
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"# 配置单机模式
conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)# 文件路径，这个文件里的内容如下
"""
a b c
a c
d a
"""
path = 'file:///home/wgq/learn_spark/test.txt'# 读取文件
data = sc.textFile(path, 2)
data = data.filter(lambda line: 'c' in line)
# 打印每一行
data.foreach(print)"""
输出：
a b c
a c
"""

三. map操作

定义任务：

将[1, 2, 3, 4, 5]里的数据，奇数取反，偶数+1

from pyspark import SparkConf, SparkContext
import os
# 这里配置spark对用的python版本，如果版本不一致就会报错
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"# 配置单机模式
conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)arr = [1, 2, 3, 4, 5]
data = sc.parallelize(arr)
def change(x):if x % 2 == 1:return -xelse:return x + 1
# 打印每一行
data = data.map(change)
data.foreach(print)
"""
输出：
-1
3
-3
5
-5
"""

四. flatMap操作

将map后的所有元素打平

定义任务：

输出文件中的所有字母

from pyspark import SparkConf, SparkContext
import os
# 这里配置spark对用的python版本，如果版本不一致就会报错
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"# 配置单机模式
conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)arr = ['a b c', 'b c', 'd a']
data = sc.parallelize(arr)# 打印每一行
data = data.flatMap(lambda x: x.split(' '))
data.foreach(print)
"""
输出：
a
b
c
b
c
d
a
"""

五. groupByKey操作

这个操作会对key进行groupby，然后将value存在一个resultIterable里

from pyspark import SparkConf, SparkContext
import os
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)arr = [('a', 1), ('b', 1), ('c', 1),('b', 1), ('c', 1), ('d', 1), ('a', 1)
]
data = sc.parallelize(arr)# 打印每一行
data = data.groupByKey()
data.foreach(print)
"""
输出：
('a', <pyspark.resultiterable.ResultIterable object at 0x7f3c1764f7d0>)
('b', <pyspark.resultiterable.ResultIterable object at 0x7f3c1764f790>)
('c', <pyspark.resultiterable.ResultIterable object at 0x7f3c1764f7d0>)
('d', <pyspark.resultiterable.ResultIterable object at 0x7f3c1764f790>)
"""

六. reduceByKey操作

按照key对value执行reduce

from pyspark import SparkConf, SparkContext
import os
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)arr = [('a', 1), ('b', 1), ('c', 1),('b', 1), ('c', 1), ('d', 1), ('a', 1)
]
data = sc.parallelize(arr)# 打印每一行
data = data.reduceByKey(lambda a, b: a + b)
data.foreach(print)
"""
输出：
('a', 2)
('b', 2)
('c', 2)
('d', 1)
"""

七. 词频统计例子

from pyspark import SparkConf, SparkContext
import os
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)path = 'file:///home/wgq/learn_spark/test.txt'# 读取文件
data = sc.textFile(path)data = data.flatMap(lambda x: x.split(' ')). \map(lambda x: (x, 1)). \reduceByKey(lambda a, b: a + b). \foreach(print)
"""
输出：
('a', 3)
('b', 1)
('c', 2)
('d', 1)
"""

八. 计算平均值例子

定义任务：

每个字母代表一本书，每个键值对表示某一天某本书被借用几次，计算每本书平均每天被借用多少次

from pyspark import SparkConf, SparkContext
import os
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)arr = [('a', 2), ('a', 3), ('a', 7),('b', 1), ('b', 2),('c', 1), ('c', 2), ('c', 3)
]data = sc.parallelize(arr)data = data.map(lambda x: (x[0], (x[1], 1))). \reduceByKey(lambda a, b: (a[0] + b[0], a[1] + b[1])). \mapValues(lambda x: x[0] / x[1]). \foreach(print)
"""
输出：
('a', 4.0)
('b', 1.5)
('c', 2.0)
"""

九. 二次排序

定义任务：

一个元素包含两个数字，按第一个升序，如果第一个相同，第二个降序排列

from pyspark import SparkConf, SparkContext
import os
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)arr = [(1, 2), (1, 1), (1, 7),(3, 3), (3, 1),(2, 1), (2, 2)
]data = sc.parallelize(arr)data = data.sortBy(lambda x: (x[0], -x[1])). \foreach(print)
"""
输入：
(1, 7)
(1, 2)
(1, 1)
(2, 2)
(2, 1)
(3, 3)
(3, 1)
"""

十. 文件排序

任务定义

有多个输入文件，文件中每一行为一个数字，要求读取所有文件，排序，第一个数字为rank值，第二个为整数

注意，读取多个文件以后，需要repartition在一起，否则是每个文件分开计算的

from pyspark import SparkConf, SparkContext
import os
os.environ["PYSPARK_PYTHON"] = "/home/wgq/anaconda3/envs/spark/bin/python3.7"conf = SparkConf().setMaster('local').setAppName('my_app')
sc = SparkContext(conf=conf)path = 'file:///home/wgq/learn_spark/nums*'
data = sc.textFile(path)# arr = [
#     (1, 2), (1, 1), (1, 7),
#     (3, 3), (3, 1),
#     (2, 1), (2, 2)
# ]
idx = -1
def getIndex():global idxidx += 1return idxdata = data.map(lambda x: int(x.strip())). \repartition(1). \sortBy(lambda x: x). \map(lambda x: (getIndex(), x)). \foreach(print)
"""
输出：
(0, 1)
(1, 2)
(2, 4)
(3, 5)
(4, 7)
(5, 9)
(6, 10)
"""

pyspark中RDD基本操作相关推荐

PySpark之RDD基本操作
PySpark之RDD基本操作 Spark是基于内存的计算引擎,它的计算速度非常快.但是仅仅只涉及到数据的计算,并没有涉及到数据的存储,但是,spark的缺点是:吃内存,不太稳定总体而言,Spark ...
pyspark rdd 基本操作
pyspark rdd 基本操作原文链接 #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Created on F ...
[大数据]PySpark原理与基本操作
一 PySpark Spark运行时架构首先我们先回顾下Spark的基本运行时架构,如下图所示,其中橙色部分表示为JVM,Spark应用程序运行时主要分为Driver和Executor,Driver ...
Spark RDD/Core 编程 API入门系列之rdd实战（rdd基本操作实战及transformation和action流程图）（源码）（三）...
本博文的主要内容是: 1.rdd基本操作实战 2.transformation和action流程图 3.典型的transformation和action RDD有3种操作: 1. Trandform ...
pyspark：RDD和DataFrame
作为数据挖掘工程师,以后必不可免要用到并行计算,pyspark是python操作spark的API,本人因此入了坑. 1 pyspark的安装见我另一篇博客:https://blog.csdn.ne ...
Nilearn中的基本操作和查看
目录 Nilearn简介 Nilearn操作第一步:查看数据第二步:平滑操作第三步:保存结果到文件中本分享为脑机学习者Rose整理发表于公众号:脑机接口社区 .QQ交流群:903290195 ...
Spark中 RDD之coalesce与repartition区别
Spark中 RDD之coalesce与repartition区别 coalesce def coalesce(numPartitions: Int, shuffle: Boolean = false ...
pyspark 条件_删除pyspark中特定条件下的特定行
我是火花的新手 . 我想删除一行使用spark sql.due来删除temptable中的不兼容性到目前为止我已经读过,操作删除像sql查询我需要永久保存pyspark中的表,这是hive表我猜 . ...
ORCAD 原理图中的基本操作
ORCAD 原理图中的基本操作 ================================================================ ORCAD支持单快捷键操作,这是太方便 ...

pyspark中RDD基本操作

写在前面

一. 简单配置和读取txt，并打印

二. filter操作

三. map操作

四. flatMap操作

五. groupByKey操作

六. reduceByKey操作

七. 词频统计例子

八. 计算平均值例子

九. 二次排序

十. 文件排序

pyspark中RDD基本操作相关推荐

最新文章

热门文章