Python大数据之PySpark(三)使用Python语言开发Spark程序代码

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Python大数据之PySpark(三)使用Python语言开发Spark程序代码

使用Python语言开发Spark程序代码

  • Spark Standalone的PySpark的搭建----bin/pyspark --master spark://node1:7077
  • Spark StandaloneHA的搭建—Master的单点故障(node1,node2),zk的leader选举机制,1-2min还原
  • 【scala版本的交互式界面】bin/spark-shell --master xxx
  • 【python版本交互式界面】bin/pyspark --master xxx
  • 【提交任务】bin/spark-submit --master xxxx

【学会配置】Windows的PySpark环境配置

  • 1-安装Andaconda
  • 2-在Anaconda Prompt中安装PySpark
  • 3-执行安装
  • 4-使用Pycharm构建Project(准备工作)
  • 需要配置anaconda的环境变量–参考课件
  • 需要配置hadoop3.3.0的安装包,里面有winutils,防止pycharm写代码的过程中报错

补充:

PyCharm构建Python project

  • 项目规划
  • 项目名称:Bigdata25-pyspark_3.1.2
  • 模块名称:PySpark-SparkBase_3.1.2,PySpark-SparkCore_3.1.2,PySpark-SparkSQL_3.1.2
  • 文件夹:
  • main pyspark的代码
  • data 数据文件
  • config 配置文件
  • test 常见python测试代码放在test中

应用入口:SparkContext

WordCount代码实战

  • 需求:给你一个文本文件,统计出单词的数量
  • 算子:rdd的api的操作,就是算子,flatMap扁平化算子,map转换算子
  • Transformation算子
  • Action算子
  • 步骤:
  • 1-首先创建SparkContext上下文环境
    2-从外部文件数据源读取数据
    3-执行flatmap执行扁平化操作
    4-执行map转化操作,得到(word,1)
    5-reduceByKey将相同Key的Value数据累加操作
    6-将结果输出到文件系统或打印
  • 代码:
# -*- coding: utf-8 -*-
# Program function: Spark的第一个程序
# 1-思考:sparkconf和sparkcontext从哪里导保
# 2-如何理解算子?Spark中算子有2种,
# 一种称之为Transformation算子(flatMapRDD-mapRDD-reduceBykeyRDD),
# 一种称之为Action算子(输出到控制台,或文件系统或hdfs),比如collect或saveAsTextFile都是Action算子
from pyspark import SparkConf,SparkContext
if __name__ == '__main__':
   # 1 - 首先创建SparkContext上下文环境
   conf = SparkConf().setAppName("FirstSpark").setMaster("local[*]")
   sc = SparkContext(conf=conf)
   sc.setLogLevel("WARN")#日志输出级别
   # 2 - 从外部文件数据源读取数据
   fileRDD = sc.textFile("D:\BigData\PyWorkspace\Bigdata25-pyspark_3.1.2\PySpark-SparkBase_3.1.2\data\words.txt")
   # print(type(fileRDD))#<class 'pyspark.rdd.RDD'>
   # all the data is loaded into the driver's memory.
   # print(fileRDD.collect())
   # ['hello you Spark Flink', 'hello me hello she Spark']
   # 3 - 执行flatmap执行扁平化操作
   flat_mapRDD = fileRDD.flatMap(lambda words: words.split(" "))
   # print(type(flat_mapRDD))
   # print(flat_mapRDD.collect())
   #['hello', 'you', 'Spark', 'Flink', 'hello', 'me', 'hello', 'she', 'Spark']
   # # 4 - 执行map转化操作,得到(word, 1)
   rdd_mapRDD = flat_mapRDD.map(lambda word: (word, 1))
   # print(type(rdd_mapRDD))#<class 'pyspark.rdd.PipelinedRDD'>
   # print(rdd_mapRDD.collect())
   # [('hello', 1), ('you', 1), ('Spark', 1), ('Flink', 1), ('hello', 1), ('me', 1), ('hello', 1), ('she', 1), ('Spark', 1)]
   # 5 - reduceByKey将相同Key的Value数据累加操作
   resultRDD = rdd_mapRDD.reduceByKey(lambda x, y: x + y)
   # print(type(resultRDD))
   # print(resultRDD.collect())
   # [('Spark', 2), ('Flink', 1), ('hello', 3), ('you', 1), ('me', 1), ('she', 1)]
   # 6 - 将结果输出到文件系统或打印
   resultRDD.saveAsTextFile("D:\BigData\PyWorkspace\Bigdata25-pyspark_3.1.2\PySpark-SparkBase_3.1.2\data\output\wordsAdd")
   # 7-停止SparkContext
   sc.stop()#Shut down the SparkContext.

  • 总结:

TopK需求

需求:[(‘Spark’, 2), (‘Flink’, 1), (‘hello’, 3), (‘you’, 1), (‘me’, 1), (‘she’, 1)]

排序:[ (‘hello’, 3),(‘Spark’, 2),]

共识:Spark核心或灵魂是rdd,spark的所有操作都是基于rdd的操作

代码:

# -*- coding: utf-8 -*-
# Program function: 针对于value单词统计计数的排序
# 1-思考:sparkconf和sparkcontext从哪里导保
# 2-如何理解算子?Spark中算子有2种,
# 一种称之为Transformation算子(flatMapRDD-mapRDD-reduceBykeyRDD),
# 一种称之为Action算子(输出到控制台,或文件系统或hdfs),比如collect或saveAsTextFile都是Action算子
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
# 1 - 首先创建SparkContext上下文环境
conf = SparkConf().setAppName("FirstSpark").setMaster("local[*]")
sc = SparkContext(conf=conf)
sc.setLogLevel("WARN")  # 日志输出级别
# 2 - 从外部文件数据源读取数据
fileRDD = sc.textFile("D:\BigData\PyWorkspace\Bigdata25-pyspark_3.1.2\PySpark-SparkBase_3.1.2\data\words.txt")
# print(type(fileRDD))#<class 'pyspark.rdd.RDD'>
# all the data is loaded into the driver's memory.
# print(fileRDD.collect())
# ['hello you Spark Flink', 'hello me hello she Spark']
# 3 - 执行flatmap执行扁平化操作
flat_mapRDD = fileRDD.flatMap(lambda words: words.split(" "))
# print(type(flat_mapRDD))
# print(flat_mapRDD.collect())
# ['hello', 'you', 'Spark', 'Flink', 'hello', 'me', 'hello', 'she', 'Spark']
# # 4 - 执行map转化操作,得到(word, 1)
rdd_mapRDD = flat_mapRDD.map(lambda word: (word, 1))
# print(type(rdd_mapRDD))#<class 'pyspark.rdd.PipelinedRDD'>
# print(rdd_mapRDD.collect())
# [('hello', 1), ('you', 1), ('Spark', 1), ('Flink', 1), ('hello', 1), ('me', 1), ('hello', 1), ('she', 1), ('Spark', 1)]
# 5 - reduceByKey将相同Key的Value数据累加操作
resultRDD = rdd_mapRDD.reduceByKey(lambda x, y: x + y)
# print(type(resultRDD))
print(resultRDD.collect())
# [('Spark', 2), ('Flink', 1), ('hello', 3), ('you', 1), ('me', 1), ('she', 1)]
# 6 针对于value单词统计计数的排序
print("==============================sortBY=============================")
print(resultRDD.sortBy(lambda x: x[1], ascending=False).take(3))
# [('hello', 3), ('Spark', 2), ('Flink', 1)]
print(resultRDD.sortBy(lambda x: x[1], ascending=False).top(3, lambda x: x[1]))
print("==============================sortBykey=============================")
print(resultRDD.map(lambda x: (x[1], x[0])).collect())
# [(2, 'Spark'), (1, 'Flink'), (3, 'hello'), (1, 'you'), (1, 'me'), (1, 'she')]
print(resultRDD.map(lambda x: (x[1], x[0])).sortByKey(False).take(3))
#[(3, 'hello'), (2, 'Spark'), (1, 'Flink')]
# 7-停止SparkContext
sc.stop()  # Shut down the SparkContext.
  • sortBy
  • sortByKey操作

从HDFS读取数据

# -*- coding: utf-8 -*-
# Program function: 从HDFS读取文件
from pyspark import SparkConf, SparkContext
import time
if __name__ == '__main__':
 # 1 - 首先创建SparkContext上下文环境
 conf = SparkConf().setAppName("FromHDFS").setMaster("local[*]")
 sc = SparkContext(conf=conf)
 sc.setLogLevel("WARN")  # 日志输出级别
 # 2 - 从外部文件数据源读取数据
 fileRDD = sc.textFile("hdfs://node1:9820/pydata/input/hello.txt")
 # ['hello you Spark Flink', 'hello me hello she Spark']
 # 3 - 执行flatmap执行扁平化操作
 flat_mapRDD = fileRDD.flatMap(lambda words: words.split(" "))
 # ['hello', 'you', 'Spark', 'Flink', 'hello', 'me', 'hello', 'she', 'Spark']
 # # 4 - 执行map转化操作,得到(word, 1)
 rdd_mapRDD = flat_mapRDD.map(lambda word: (word, 1))
 # [('hello', 1), ('you', 1), ('Spark', 1), ('Flink', 1), ('hello', 1), ('me', 1), ('hello', 1), ('she', 1), ('Spark', 1)]
 # 5 - reduceByKey将相同Key的Value数据累加操作
 resultRDD = rdd_mapRDD.reduceByKey(lambda x, y: x + y)
 # print(type(resultRDD))
 print(resultRDD.collect())
 # 休息几分钟
 time.sleep(600)
 # 7-停止SparkContext
 sc.stop()  # Shut down the SparkContext.

         

         

提交代码到集群执行

  • 关键:sys.argv[1],
  • 代码:
# -*- coding: utf-8 -*-
# Program function: 提交任务执行
import sys
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
   # 1 - 首先创建SparkContext上下文环境
   conf = SparkConf().setAppName("FromHDFS").setMaster("local[*]")
   sc = SparkContext(conf=conf)
   sc.setLogLevel("WARN")  # 日志输出级别
   # 2 - 从外部文件数据源读取数据
   # hdfs://node1:9820/pydata/input/hello.txt
   fileRDD = sc.textFile(sys.argv[1])
   # ['hello you Spark Flink', 'hello me hello she Spark']
   # 3 - 执行flatmap执行扁平化操作
   flat_mapRDD = fileRDD.flatMap(lambda words: words.split(" "))
   # ['hello', 'you', 'Spark', 'Flink', 'hello', 'me', 'hello', 'she', 'Spark']
   # # 4 - 执行map转化操作,得到(word, 1)
   rdd_mapRDD = flat_mapRDD.map(lambda word: (word, 1))
   # [('hello', 1), ('you', 1), ('Spark', 1), ('Flink', 1), ('hello', 1), ('me', 1), ('hello', 1), ('she', 1), ('Spark', 1)]
   # 5 - reduceByKey将相同Key的Value数据累加操作
   resultRDD = rdd_mapRDD.reduceByKey(lambda x, y: x + y)
   # print(type(resultRDD))
   resultRDD.saveAsTextFile(sys.argv[2])
   # 7-停止SparkContext
   sc.stop()  # Shut down the SparkContext.
  • 结果:

[掌握-扩展阅读]远程PySpark环境配置

  • 需求:需要将PyCharm连接服务器,同步本地写的代码到服务器上,使用服务器上的Python解析器执行
  • 步骤:
  • 1-准备PyCharm的连接

  • 2-需要了解服务器的地址,端口号,用户名,密码


  • 设置自动的上传,如果不太好使,重启pycharm

  • 3-pycharm读取的文件都需要上传到linux中,复制相对路径

  • 4-执行代码在远程服务器上
  • 5-执行代码

# -*- coding: utf-8 -*-
# Program function: Spark的第一个程序
# 1-思考:sparkconf和sparkcontext从哪里导保
# 2-如何理解算子?Spark中算子有2种,
# 一种称之为Transformation算子(flatMapRDD-mapRDD-reduceBykeyRDD),
# 一种称之为Action算子(输出到控制台,或文件系统或hdfs),比如collect或saveAsTextFile都是Action算子
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
 # 1 - 首先创建SparkContext上下文环境
 conf = SparkConf().setAppName("FirstSpark").setMaster("local[*]")
 sc = SparkContext(conf=conf)
 sc.setLogLevel("WARN")  # 日志输出级别
 # 2 - 从外部文件数据源读取数据
 fileRDD = sc.textFile("/export/data/pyspark_workspace/PySpark-SparkBase_3.1.2/data/words.txt")
 # fileRDD = sc.parallelize(["hello you", "hello me", "hello spark"])
 # 3 - 执行flatmap执行扁平化操作
 flat_mapRDD = fileRDD.flatMap(lambda words: words.split(" "))
 # print(type(flat_mapRDD))
 # print(flat_mapRDD.collect())
 # ['hello', 'you', 'Spark', 'Flink', 'hello', 'me', 'hello', 'she', 'Spark']
 # # 4 - 执行map转化操作,得到(word, 1)
 rdd_mapRDD = flat_mapRDD.map(lambda word: (word, 1))
 # print(type(rdd_mapRDD))#<class 'pyspark.rdd.PipelinedRDD'>
 # print(rdd_mapRDD.collect())
 # [('hello', 1), ('you', 1), ('Spark', 1), ('Flink', 1), ('hello', 1), ('me', 1), ('hello', 1), ('she', 1), ('Spark', 1)]
 # 5 - reduceByKey将相同Key的Value数据累加操作
 resultRDD = rdd_mapRDD.reduceByKey(lambda x, y: x + y)
 # print(type(resultRDD))
 print(resultRDD.collect())
 # [('Spark', 2), ('Flink', 1), ('hello', 3), ('you', 1), ('me', 1), ('she', 1)]
 # 6 - 将结果输出到文件系统或打印
 # resultRDD.saveAsTextFile("D:\BigData\PyWorkspace\Bigdata25-pyspark_3.1.2\PySpark-SparkBase_3.1.2\data\output\wordsAdd")
 # 7-停止SparkContext
 sc.stop()  # Shut down the SparkContext.
  • 切记忘记上传python的文件,直接执行
  • 注意1:自动上传设置

  • 注意2:增加如何使用standalone和HA的方式提交代码执行
  • 但是需要注意,尽可能使用hdfs的文件,不要使用单机版本的文件,因为standalone是集群模式
# -*- coding: utf-8 -*-
# Program function: Spark的第一个程序
# 1-思考:sparkconf和sparkcontext从哪里导保
# 2-如何理解算子?Spark中算子有2种,
# 一种称之为Transformation算子(flatMapRDD-mapRDD-reduceBykeyRDD),
# 一种称之为Action算子(输出到控制台,或文件系统或hdfs),比如collect或saveAsTextFile都是Action算子
>from pyspark import SparkConf, SparkContext
>
>if __name__ == '__main__':
>
># 1 - 首先创建SparkContext上下文环境
>
>conf = SparkConf().setAppName("FirstSpark").setMaster("spark://node1:7077,node2:7077")
>sc = SparkContext(conf=conf)
>sc.setLogLevel("WARN")  # 日志输出级别
>
># 2 - 从外部文件数据源读取数据
>
>fileRDD = sc.textFile("hdfs://node1:9820/pydata/input/hello.txt")
>
># fileRDD = sc.parallelize(["hello you", "hello me", "hello spark"])
>
># 3 - 执行flatmap执行扁平化操作
>
>flat_mapRDD = fileRDD.flatMap(lambda words: words.split(" "))
>
># print(type(flat_mapRDD))
>
># print(flat_mapRDD.collect())
>
># ['hello', 'you', 'Spark', 'Flink', 'hello', 'me', 'hello', 'she', 'Spark']
>
># # 4 - 执行map转化操作,得到(word, 1)
>
>rdd_mapRDD = flat_mapRDD.map(lambda word: (word, 1))
>
># print(type(rdd_mapRDD))#<class 'pyspark.rdd.PipelinedRDD'>
>
># print(rdd_mapRDD.collect())
>
># [('hello', 1), ('you', 1), ('Spark', 1), ('Flink', 1), ('hello', 1), ('me', 1), ('hello', 1), ('she', 1), ('Spark', 1)]
>
># 5 - reduceByKey将相同Key的Value数据累加操作
>
>resultRDD = rdd_mapRDD.reduceByKey(lambda x, y: x + y)
>
># print(type(resultRDD))
>
>print(resultRDD.collect())
>
># [('Spark', 2), ('Flink', 1), ('hello', 3), ('you', 1), ('me', 1), ('she', 1)]
>
># 6 - 将结果输出到文件系统或打印
>
># resultRDD.saveAsTextFile("D:\BigData\PyWorkspace\Bigdata25-pyspark_3.1.2\PySpark-SparkBase_3.1.2\data\output\wordsAdd")
>
># 7-停止SparkContext
>
>sc.stop()  # Shut down the SparkContext.

总结

  • 函数式编程
#Python中的函数式编程
#1-map(func, *iterables) --> map object
def fun(x):
    return x*x
#x=[1,2,3,4,5] y=map(fun,[1,2,3,4,5]) #[1, 4, 9, 16, 25]
print(list(map(fun, [1, 2, 3, 4, 5])))
#2-lambda 匿名函数  java: x=>x*x 表达式  Scala:x->x*x
g=lambda x:x*x
print(g(10))
print(list(map(lambda x:x*x, [1, 2, 3, 4, 5])))
def add(x,y):
    return x+y
print(list(map(add, range(5), range(5, 10))))
print(list(map(lambda x,y:x+y,range(5),range(5,10))))
#3- [add(x,y) for x,y in zip(range(5),range(5,10))]
# print(list(zip([1, 2, 3], [4, 5, 6])))#[1,4],[2,5]
# print(list(zip([1, 2, 3], [4, 5, 6,7])))#[1,4],[2,5]
# print(list(zip([1, 2, 3,6], [4, 5, 6])))#[1,4],[2,5]
# 语法 lambda表达式语言:【lambda 变量:表达式】
# 列表表达式 [表达式 for 变量 in 可迭代的序列中 if 条件]
print([add(x, y) for x, y in zip(range(5), range(5))])
#[0, 2, 4, 6, 8]
#3-reduce
from functools import  reduce
# ((((1+2)+3)+4)+5)
print(reduce(lambda x, y: x + y, [1, 2, 3, 4, 5]))
# 4-filter
seq1=['foo','x41','?1','***']
def func(x):
    #Return True if the string is an alpha-numeric string
    return x.isalnum()
print(list(filter(func,seq1))) #返回 filter 对象
# sorted()
# 最后我们可以看到,函数式编程有如下好处:
# 1)代码更简单了。
# 2)数据集,操作,返回值都放到了一起。
# 3)你在读代码的时候,没有了循环体,于是就可以少了些临时变量,以及变量倒来倒去逻辑。
# 4)你的代码变成了在描述你要干什么,而不是怎么去干。


相关实践学习
简单用户画像分析
本场景主要介绍基于海量日志数据进行简单用户画像分析为背景,如何通过使用DataWorks完成数据采集 、加工数据、配置数据质量监控和数据可视化展现等任务。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
8天前
|
监控 Python
Python中的装饰器:提升代码灵活性与可读性
在Python编程中,装饰器是一种强大的工具,能够提升代码的灵活性和可读性。本文将介绍装饰器的基本概念、使用方法以及实际应用场景,帮助读者更好地理解和利用这一功能。
|
9天前
|
算法 测试技术 开发者
性能优化与代码审查:提升Python开发效率
【4月更文挑战第9天】本文强调了Python开发中性能优化和代码审查的重要性。性能优化包括选择合适数据结构、使用生成器和避免全局变量,而代码审查涉及遵循编码规范、使用静态代码分析工具和编写单元测试。这些实践能提升代码效率和可维护性,促进团队协作。
|
6天前
|
分布式计算 Hadoop 大数据
大数据技术与Python:结合Spark和Hadoop进行分布式计算
【4月更文挑战第12天】本文介绍了大数据技术及其4V特性,阐述了Hadoop和Spark在大数据处理中的作用。Hadoop提供分布式文件系统和MapReduce,Spark则为内存计算提供快速处理能力。通过Python结合Spark和Hadoop,可在分布式环境中进行数据处理和分析。文章详细讲解了如何配置Python环境、安装Spark和Hadoop,以及使用Python编写和提交代码到集群进行计算。掌握这些技能有助于应对大数据挑战。
|
9天前
|
缓存 监控 算法
优化Python代码性能的10个技巧
提高Python代码性能是每个开发者都需要关注的重要问题。本文将介绍10个实用的技巧,帮助你优化Python代码,提升程序的运行效率和性能表现。无论是避免内存泄漏、减少函数调用次数,还是使用适当的数据结构,都能在不同场景下发挥作用,使你的Python应用更加高效稳定。
|
1天前
|
数据安全/隐私保护 Python
Python中的装饰器:提升代码可读性与灵活性
Python中的装饰器是一种强大的工具,可以在不改变函数原有逻辑的情况下,为函数添加额外的功能。本文将介绍装饰器的基本概念和用法,并通过实例演示如何利用装饰器提升代码的可读性和灵活性,使代码更加简洁、易于维护。
|
1天前
|
BI 开发者 数据格式
Python代码填充数据到word模板中
【4月更文挑战第16天】
|
3天前
|
前端开发 数据挖掘 API
使用Python中的Flask框架进行Web应用开发
【4月更文挑战第15天】在Python的Web开发领域,Flask是一个备受欢迎的轻量级Web框架。它简洁、灵活且易于扩展,使得开发者能够快速地构建出高质量的Web应用。本文将深入探讨Flask框架的核心特性、使用方法以及在实际开发中的应用。
|
3天前
|
缓存 算法 Python
优化Python代码的十大技巧
本文介绍了十种优化Python代码的技巧,涵盖了从代码结构到性能调优的方方面面。通过学习和应用这些技巧,你可以提高Python程序的执行效率,提升代码质量,以及更好地应对复杂的编程任务。
|
3天前
|
程序员 Python
Python中的装饰器:提升代码可读性与灵活性
在Python编程中,装饰器是一种强大的工具,可以在不修改原始代码的情况下,动态地添加功能。本文将深入探讨Python中装饰器的原理、用法和实际应用,以及如何利用装饰器提升代码的可读性和灵活性。
|
5天前
|
缓存 开发者 Python
深入探讨Python中的装饰器:提升代码可读性与灵活性
在Python编程中,装饰器是一种强大的工具,可以在不修改原始函数代码的情况下,对其行为进行扩展或修改。本文将深入探讨装饰器的原理和用法,以及如何利用装饰器提升代码的可读性和灵活性,为Python开发者提供更加优雅和高效的编程方式。

热门文章

最新文章