# coding:utf8 from pyspark.sql import SparkSession from pyspark.sql.functions import avg if __name__ == '__main__': spark = SparkSession\ .builder\ .master("local[*]")\ .appName("Simple Application")\ .getOrCreate() bookDF = spark \ .createDataFrame([("spark", 2), ("hadoop", 6), ("hadoop", 4), ("spark", 6)]) \ .toDF("book", "amount") avgDF = bookDF.groupBy("book").agg(avg("amount")) avgDF.show() spark.stop()

视频信息

答案文本 复制

视频字幕 复制

PySpark是Apache Spark的Python API，它为Python开发者提供了强大的分布式数据处理能力。与传统的单机数据处理不同，PySpark可以在集群上并行处理大规模数据集。它的主要特点包括分布式计算、内存计算、易于使用和多语言支持。通过PySpark，我们可以轻松处理TB级别的数据，进行复杂的数据分析和机器学习任务。 这是一个简单的Spark应用程序示例。代码展示了如何使用PySpark进行基本的数据处理。首先导入必要的模块，然后创建SparkSession，接着创建DataFrame，对数据进行分组聚合计算，最后显示结果并停止Spark会话。 SparkSession的创建使用了Builder设计模式。首先调用SparkSession.builder开始构建过程，然后通过master方法设置运行模式。这里使用local[*]表示在本地模式运行，星号表示使用所有可用的CPU核心。接着用appName方法为应用程序设置一个名称，便于在Spark UI中识别。最后调用getOrCreate方法，如果已存在SparkSession则返回现有实例，否则创建新的实例。 接下来创建DataFrame并进行数据处理。首先使用spark.createDataFrame从Python列表创建DataFrame，然后用toDF方法指定列名为book和amount。接着使用groupBy方法按book列进行分组，并用agg方法结合avg函数计算每本书的平均数量。最后调用show方法显示结果，可以看到spark的平均值是4.0，hadoop的平均值是5.0。 程序执行后会输出一个表格，显示每本书的平均数量。hadoop的平均数量是5.0，spark的平均数量是4.0。这个简单的示例展示了Spark的基本用法：创建SparkSession、处理数据、执行计算和显示结果。最后不要忘记调用spark.stop()来释放资源。这就是一个完整的Spark应用程序的基本结构。 DataFrame的创建过程分为几个步骤。首先，我们有一个Python列表，包含四个元组，每个元组代表一行数据。调用createDataFrame方法时，Spark会自动推断数据类型，字符串被识别为string类型，数字被识别为long类型。此时创建的DataFrame使用默认列名_1和_2。接着调用toDF方法，将列名重命名为book和amount，形成最终的DataFrame结构，类似于关系型数据库中的表。 GroupBy聚合操作是数据分析中的核心功能。首先，groupBy方法按照book列对数据进行分组，将相同书名的记录归为一组。spark组包含数值2和6，hadoop组包含数值6和4。然后agg方法结合avg函数对每组的amount列计算平均值。spark组的平均值是2加6除以2等于4.0，hadoop组的平均值是6加4除以2等于5.0。最终生成一个新的DataFrame，包含每本书及其对应的平均数量。 程序执行后会输出一个格式化的表格，显示每本书的平均数量。hadoop的平均值是5.0，spark的平均值是4.0。整个程序的执行流程包括五个步骤：首先创建SparkSession，然后构建DataFrame，接着执行GroupBy聚合操作，显示计算结果，最后调用spark.stop()释放资源。资源管理是非常重要的最佳实践，及时释放SparkSession可以避免资源泄漏，提高系统整体性能。这个简单的示例展示了PySpark进行数据处理的完整流程。