专栏首页程序生活大数据入门与实战-PySpark的使用教程

大数据入门与实战-PySpark的使用教程

1 PySpark简介

Apache Spark是用Scala编程语言编写的。为了用Spark支持Python,Apache Spark社区发布了一个工具PySpark。使用PySpark,您也可以使用Python编程语言处理RDD。正是由于一个名为Py4j的库,他们才能实现这一目标。 这里不介绍PySpark的环境设置,主要介绍一些实例,以便快速上手。

2 PySpark - SparkContext

SparkContext是任何spark功能的入口点。当我们运行任何Spark应用程序时,会启动一个驱动程序,它具有main函数,并且此处启动了SparkContext。然后,驱动程序在工作节点上的执行程序内运行操作。

SparkContext使用Py4J启动JVM并创建JavaSparkContext。默认情况下,PySpark将SparkContext作为'sc'提供,因此创建新的SparkContext将不起作用。

以下代码块包含PySpark类的详细信息以及SparkContext可以采用的参数。

class pyspark.SparkContext (
   master = None,
   appName = None, 
   sparkHome = None, 
   pyFiles = None, 
   environment = None, 
   batchSize = 0, 
   serializer = PickleSerializer(), 
   conf = None, 
   gateway = None, 
   jsc = None, 
   profiler_cls = <class 'pyspark.profiler.BasicProfiler'>
)

以下是SparkContext的参数具体含义:

  • Master- 它是连接到的集群的URL。
  • appName- 您的工作名称。
  • sparkHome - Spark安装目录。
  • pyFiles - 要发送到集群并添加到PYTHONPATH的.zip或.py文件。
  • environment - 工作节点环境变量。
  • batchSize - 表示为单个Java对象的Python对象的数量。设置1以禁用批处理,设置0以根据对象大小自动选择批处理大小,或设置为-1以使用无限批处理大小。
  • serializer- RDD序列化器。
  • Conf - L {SparkConf}的一个对象,用于设置所有Spark属性。
  • gateway - 使用现有网关和JVM,否则初始化新JVM。
  • JSC - JavaSparkContext实例。
  • profiler_cls - 用于进行性能分析的一类自定义Profiler(默认为pyspark.profiler.BasicProfiler)。 在上述参数中,主要使用master和appname。任何PySpark程序的会使用以下两行:
from pyspark import SparkContext
sc = SparkContext("local", "First App")

2.1 SparkContext示例 - PySpark Shell

现在你对SparkContext有了足够的了解,让我们在PySpark shell上运行一个简单的例子。在这个例子中,我们将计算README.md文件中带有字符“a”或“b”的行数。那么,让我们说如果一个文件中有5行,3行有字符'a',那么输出将是→ Line with a:3。字符'b'也是如此。

注 - 我们不会在以下示例中创建任何SparkContext对象,因为默认情况下,当PySpark shell启动时,Spark会自动创建名为sc的SparkContext对象。如果您尝试创建另一个SparkContext对象,您将收到以下错误 - “ValueError:无法一次运行多个SparkContexts”。

在终端输入pyspark 启动PySpark Shell:

>>> logFile="file:////opt/modules/hadoop-2.8.5/README.txt"
>>> logData=sc.textFile(logFile).cache()
>>> numAs=logData.filter(lambda s:'a' in s).count()
>>> numBs=logData.filter(lambda s:'b' in s).count()                             
>>> print("Line with a:%i,line with b:%i" % (numAs,numBs))
Line with a:25,line with b:7

2.2 SparkContext示例 - Python程序

让我们使用Python程序运行相同的示例。创建一个名为demo.py的Python文件,并在该文件中输入以下代码。

from pyspark import SparkContext
logFile = "file:////opt/modules/hadoop-2.8.5/README.txt"
sc = SparkContext("local", "first app")
logData = sc.textFile(logFile).cache()
numAs = logData.filter(lambda s: 'a' in s).count()
numBs = logData.filter(lambda s: 'b' in s).count()
print("Line with a:%i,lines with b :%i" % (numAs, numBs))

然后我们将在终端中执行以下命令来运行此Python文件。我们将得到与上面相同的输出。

spark-submit demo.py 

3 PySpark - RDD

在介绍PySpark处理RDD操作之前,我们先了解下RDD的基本概念:

RDD代表Resilient Distributed Dataset,它们是在多个节点上运行和操作以在集群上进行并行处理的元素。RDD是不可变元素,这意味着一旦创建了RDD,就无法对其进行更改。RDD也具有容错能力,因此在发生任何故障时,它们会自动恢复。您可以对这些RDD应用多个操作来完成某项任务

要对这些RDD进行操作,有两种方法 :

  • Transformation
  • Action

转换 - 这些操作应用于RDD以创建新的RDD。Filter,groupBy和map是转换的示例。

操作 - 这些是应用于RDD的操作,它指示Spark执行计算并将结果发送回驱动程序。

要在PySpark中应用任何操作,我们首先需要创建一个PySpark RDD。以下代码块具有PySpark RDD类的详细信息 :

class pyspark.RDD (
   jrdd, 
   ctx, 
   jrdd_deserializer = AutoBatchedSerializer(PickleSerializer())
)

接下来让我们看看如何使用PySpark运行一些基本操作,用以下代码创建存储一组单词的RDD(spark使用parallelize方法创建RDD),我们现在将对单词进行一些操作。

3.1 count()

返回RDD中的元素个数

----------------------------------------count.py---------------------------------------
from pyspark import SparkContext
sc = SparkContext("local", "count app")
words = sc.parallelize(
    ["scala",
     "java",
     "hadoop",
     "spark",
     "akka",
     "spark vs hadoop",
     "pyspark",
     "pyspark and spark"
     ])
counts = words.count()
print("Number of elements in RDD -> %i" % counts)

执行spark-submit count.py,将会输出以下结果

Number of elements in RDD → 8

3.2 collect()

返回RDD中的所有元素

----------------------------------------collect.py - --------------------------------------
from pyspark import SparkContext
sc = SparkContext("local", "collect app")
words = sc.parallelize(
    ["scala",
     "java",
     "hadoop",
     "spark",
     "akka",
     "spark vs hadoop",
     "pyspark",
     "pyspark and spark"
     ])
coll = words.collect()
print("Elements in RDD -> %s" % coll)

执行spark-submit collect.py 输出以下结果

Elements in RDD -> ['scala', 'java', 'hadoop', 'spark', 'akka', 'spark vs hadoop', 'pyspark', 'pyspark and spark']

3.3 foreach(func)

仅返回满足foreach内函数条件的元素。在下面的示例中,我们在foreach中调用print函数,该函数打印RDD中的所有元素。

# foreach.py
from pyspark import SparkContext
sc = SparkContext("local", "ForEach app")
words = sc.parallelize (
   ["scala", 
   "java", 
   "hadoop", 
   "spark", 
   "akka",
   "spark vs hadoop", 
   "pyspark",
   "pyspark and spark"]
)
def f(x): print(x)
fore = words.foreach(f)

执行spark-submit foreach.py,然后输出:

scala
java
hadoop
spark
akka
spark vs hadoop
pyspark
pyspark and spark

3.4 filter(f)

返回一个包含元素的新RDD,它满足过滤器内部的功能。在下面的示例中,我们过滤掉包含''spark'的字符串。

# filter.py
from pyspark import SparkContext
sc = SparkContext("local", "Filter app")
words = sc.parallelize(
    ["scala",
     "java",
     "hadoop",
     "spark",
     "akka",
     "spark vs hadoop",
     "pyspark",
     "pyspark and spark"]
)
words_filter = words.filter(lambda x: 'spark' in x)
filtered = words_filter.collect()
print("Fitered RDD -> %s" % (filtered))

执行spark-submit filter.py:

Fitered RDD -> ['spark', 'spark vs hadoop', 'pyspark', 'pyspark and spark']

3.5 map(f, preservesPartitioning = False)

通过将该函数应用于RDD中的每个元素来返回新的RDD。在下面的示例中,我们形成一个键值对,并将每个字符串映射为值1

# map.py
from pyspark import SparkContext
sc = SparkContext("local", "Map app")
words = sc.parallelize(
    ["scala",
     "java",
     "hadoop",
     "spark",
     "akka",
     "spark vs hadoop",
     "pyspark",
     "pyspark and spark"]
)
words_map = words.map(lambda x: (x, 1))
mapping = words_map.collect()
print("Key value pair -> %s" % (mapping))

执行spark-submit map.py

Key value pair -> [('scala', 1), ('java', 1), ('hadoop', 1), ('spark', 1), ('akka', 1), ('spark vs hadoop', 1), ('pyspark', 1), ('pyspark and spark', 1)]

3.6 reduce(f)

执行指定的可交换和关联二元操作后,将返回RDD中的元素。在下面的示例中,我们从运算符导入add包并将其应用于'num'以执行简单的加法运算。说白了和Python的reduce一样:假如有一组整数[x1,x2,x3],利用reduce执行加法操作add,对第一个元素执行add后,结果为sum=x1,然后再将sum和x2执行add,sum=x1+x2,最后再将x2和sum执行add,此时sum=x1+x2+x3。

# reduce.py
from pyspark import SparkContext
from operator import add
sc = SparkContext("local", "Reduce app")
nums = sc.parallelize([1, 2, 3, 4, 5])
adding = nums.reduce(add)
print("Adding all the elements -> %i" % (adding))

执行spark-submit reduce.py:

Adding all the elements -> 15

3.7 join(other, numPartitions = None)

它返回RDD,其中包含一对带有匹配键的元素以及该特定键的所有值。

from pyspark import SparkContext
sc = SparkContext("local", "Join app")
x = sc.parallelize([("spark", 1), ("hadoop", 4)])
y = sc.parallelize([("spark", 2), ("hadoop", 5)])
joined = x.join(y)
final = joined.collect()
print( "Join RDD -> %s" % (final))

执行spark-submit join.py:

Join RDD -> [
   ('spark', (1, 2)),  
   ('hadoop', (4, 5))
]

reference:https://www.tutorialspoint.com/pyspark/index.htm

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • docker构建自定义镜像

    要构建一个镜像,第一步准备所需要的文件,第二步编写Dockerfile文件,比如我现在构建一个java web镜像

    用户2409797
  • 2018年10月1日编译型语言和解释型语言的区别

    计算机不能直接理解高级语言,只能直接理解机器语言(即计算机识别的二进制文件,因为不同的操作系统计算机识别的二进制文件是不同的,因为不同系统的指令用不同的二进制代...

    武军超
  • Java核心技术 卷I 基础知识 学习笔记(8)

    在Java SE7及以后的版本中,构造函数中可以省略泛型类型,省略的类型可以从变量的类型推断得出。

  • Java虚拟机(JVM)介绍和垃圾回收机制

    原理:通过一系列被称为“GC Roots‘’的对象作为起始点,从这些节点向下搜索,搜索所走过的路径叫做引用链,当一个节点到GC Roots没有任何引用链时,证明...

    飞狗
  • Java设计模式之【单例模式】

    在懒加载模式中,每次获取实例都要加锁,这会使程序运行速度变慢,通过双重判断加锁机制可以解决这个问题

    用户2409797
  • boost::algorithm(字符串算法库)

     trim_left_copy_if() 将字符串开头的符合我们提供的“谓词”的特定字符去掉,并且赋值给另一个字符串

    racaljk
  • 使用idea创建一个maven工程

    用户2409797
  • Java继承概述以及Java继承案例和继承的好处

    1.多个类中存在相同属性和行为时,将这些内容抽取到单独一个类中,那么多个类无需再定义这些相同属性和行为,只要继承那个类即可。

    黑泽君
  • Java核心技术 卷I 基础知识 学习笔记(1)

    简单性、面向对象、分布式、健壮性、安全性、体系结构中立、可移植性、解释型、高性能、多线程、动态性。

  • Springboot使用Filter以及踩过的坑

    1、不加@ServletComponentScan注解无法注入Filter,这个注解的作用就是去扫描@WebServlet、@WebFilter、@WebLis...

    用户2409797

扫码关注云+社区

领取腾讯云代金券