Spark学习笔记（3）：Spark DataFrame

简介：
本篇为《Spark学习笔记》系列第三部分，主要探讨Spark DataFrame的概念、操作及应用场景，帮助读者深入理解数据处理框架。 系列博客是学习厦门大学林子雨老师Spark编程基础课程的笔记，方便回顾。 系列博客包括： - Spark学习笔记（一）：Spark概述与运行原理 - Spark学习笔记（二）：RDD编程基础 在Spark SQL中增加了DataFrame这一概念，即带有Schema信息的RDD。这使得用户可以在Spark SQL环境中执行SQL语句，并且可以使用多种数据源如Hive、HDFS、Cassandra等外部来源或JSON格式的数据。 目前，Spark SQL支持Scala、Java和Python三种语言，并遵循SQL-92规范。 DataFrame的引入让Spark能够处理大规模结构化数据，相比原有的功能提供了更强的能力。它不仅增强了类型安全性还增加了更多优化选项，简化了流程并提升了效率。 在Spark 2.0及以上版本中，管理DataFrame的任务由SparkSession接口接管，替代了早期的SQLContext和HiveContext。创建一个SparkSession示例如下： ```python from pyspark.sql import SparkSession spark = SparkSession.builder.config(conf=SparkConf()).getOrCreate() ``` 在Python环境中，默认会提供SparkContext对象（sc）和SparkSession对象（spark）。 DataFrame可以从多种数据源加载，包括文本段落件、JSON文件及Parquet文件。例如： ```python # 从文本段落件加载 df_text = spark.read.text(people.txt) # 从JSON文件加载 df_json = spark.read.json(people.json) # 从Parquet文件加载 df_parquet = spark.read.parquet(people.parquet) ``` 这些方法能够根据不同的数据格式自动推断列名和类型。 创建DataFrame有两种主要方式：一是通过反射机制推断RDD的模式，二是编程定义RDD模式。 1. 反射机制推断模式： 当已知RDD的数据结构时可以使用这种方法。首先定义一个Row类然后将RDD转换为Row类型的RDD，最后调用`createDataFrame()`方法创建DataFrame并注册临时视图： ```python from pyspark.sql import Row, SparkSession spark = SparkSession.builder.config(conf=SparkConf()).getOrCreate() people_rdd = spark.sparkContext.textFile(people.txt).map(lambda line: line.split(,)) people_rdd = people_rdd.map(lambda p: Row(name=p[0], age=int(p[1]))) schema_people = spark.createDataFrame(people_rdd) schema_people.createOrReplaceTempView(people) ``` 2. 编程方式定义模式： 当无法预知数据结构时，可以通过编程方式来定义DataFrame的模式。这通常涉及先创建一个包含所需字段的类然后将RDD转换为此类实例最后使用`createDataFrame()`方法。 一旦DataFrame被注册为临时视图就可以使用`sql()`执行SQL查询： ```python query_result = spark.sql(select name, age from people where age > 20) ``` 除了支持SQL之外，DataFrame还提供了丰富的API来进行数据转换和清洗如过滤、分组、聚合及连接等操作。这些功能使得处理大规模结构化数据更加高效且易于理解。 DataFrame在Spark内部通过Catalyst编译器进行优化可以执行列式存储、代码生成和计划优化从而提高查询性能。同时，DataFrame的API支持Scala、Java和Python语言供开发人员选择最合适的编程环境。 总结来说，使用DataFrame是处理大规模结构化数据的核心技能之一，在大数据分析中具有重要价值。