PySpark SparkContext

SparkContext是任何Spark功能的入口点。当我们运行任何Spark应用程序时，会启动一个驱动程序，该驱动程序具有主函数，并在此处初始化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 − Worker节点的环境变量。

• batchSize − 作为单个Java对象表示的Python对象的数量。设置为1以禁用批处理，设置为0以根据对象大小自动选择批处理大小，或设置为-1以使用无限批处理大小。

• Serializer − RDD序列化器。

• Conf − Spark属性的L{SparkConf}对象。

• Gateway − 使用现有的网关和JVM，否则初始化一个新的JVM。

• JSC − JavaSparkContext实例。

• profiler_cls − 用于进行剖析的自定义Profiler类（默认为pyspark.profiler.BasicProfiler）。

在上述参数中， master 和 appname 最常用。任何PySpark程序的前两行如下所示：

from pyspark import SparkContext
sc = SparkContext("local", "First App")

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: Cannot run multiple SparkContexts at once”。

<<< logFile = "file:///home/hadoop/spark-2.1.0-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

SparkContext示例 – Python程序

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

----------------------------------------firstapp.py---------------------------------------
from pyspark import SparkContext
logFile = "file:///home/hadoop/spark-2.1.0-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)
----------------------------------------firstapp.py---------------------------------------

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

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