第2天：核心概念之SparkContext

在今天的文章中，我们将会介绍PySpark中的一系列核心概念，包括SparkContext、RDD等。

SparkContext概念

SparkContext是所有Spark功能的入口。无论我们希望运行什么样的Spark应用，都需要初始化SparkContext来驱动程序执行，从而将任务分配至Spark的工作节点中执行。

在PySpark中SparkContext使用Py4J来启动一个JVM并创建一个JavaSparkContext。默认情况下，PySpark已经创建了一个名为sc的SparkContext，并且在一个JVM进程中可以创建多个SparkContext，但是只能有一个active级别的，因此，如果我们在创建一个新的SparkContext是不能正常使用的。

一个Spark的基础版数据流图如下：

下面的代码块描述了在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'="">
 
)
 

这些参数的含义如下：

1. master：Spark集群的入口url地址。
2. appName：任务名称。
3. sparkHome：Spark安装目录。
4. pyFiles：.zip 或 .py 文件可发送给集群或添加至环境变量中。
5. Environment：Spark Worker节点的环境变量。
6. batchSize：批处理数量。设置为1表示禁用批处理，设置0以根据对象大小自动选择批处理大小，设置为-1以使用无限批处理大小。
7. Serializer：RDD序列化器。
8. Conf：SparkConf对象，用于设置Spark集群的相关属性。
9. Gateway：选择使用现有网关和JVM或初始化新JVM。
10. JSC：JavaSparkContext实例。
11. profiler_cls：可用于进行性能分析的自定义Profiler（默认为pyspark.profiler.BasicProfiler）。

Ps：在上述所有参数中，master和appName是最常用的参数，几乎所有的应用都需要传入这两个参数。

SparkContext实战

在我们了解了什么是SparkContext后，接下来，我们希望可以通过一些简单的PySpark shell入门示例来加深对SparkContext的理解。

在这个例子中，我们将计算README.md文件中带有字符“a”或“b”的行数。例如，假设该文件中有5行，3行有’a’字符，那么输出将是 Line with a：3。

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

logFile = "file:///ssd1/spark-2.4.2-bin-hadoop2.7/README.md"
 
logData = sc.textFile(logFile).cache()
 
numAs = logData.filter(lambda s: 'a' in s).count()
 
numBs = logData.filter(lambda s: 'b' in s).count()
 
print "Lines with a: %i, lines with b: %i" % (numAs, numBs)
 
# Lines with a: 62, lines with b: 30
 

那么，如果我们不希望使用PySpark shell交互式命令行，而是希望一个Python文件中实现呢？ first_app.py文件如下：

from pyspark import SparkContext
 
logFile = "file:///ssd1/spark-2.4.2-bin-hadoop2.7/README.md" 
 
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 "Lines with a: %i, lines with b: %i" % (numAs, numBs)
 

此时，我们可以执行如下命令来触发该任务：

$SPARK_HOME/bin/spark-submit firstapp.py
 
Output: Lines with a: 62, lines with b: 30
 
 </class>