from pyspark.conf import SparkConf from pyspark.sql.session import SparkSession from pyspark import SparkContext from pyspark.sql.types import Row if __name__ == "__main__": sc = SparkContext("local","Simple App") peopleRDD = sc.textFile("file:///usr/local/spark/employee.txt") rowRDD = peopleRDD.map(lambda line : line.split(",")).map(lambda attributes : Row(int(attributes[0]),attributes[1],int(attributes[2]))).toDF() rowRDD.createOrReplaceTempView("employee") personsDF = spark.sql("select * from employee") personsDF.rdd.map(lambda t : "id:"+str(t[0])+","+"Name:"+t[1]+","+"age:"+str(t[2])).foreach(print)

这是一段 PySpark 的代码，用于读取本地文件系统中的数据并将其转换为 DataFrame 对象，然后使用 Spark SQL 对 DataFrame 进行查询操作。具体来说，代码中使用了 SparkConf 和 SparkSession 对象来创建 SparkContext，然后使用 SparkContext 对象读取本地文件系统中的数据，将其转换为 DataFrame 对象并进行查询操作。

首先，代码中使用 SparkConf 对象创建了一个 SparkConf 配置对象，用于配置 Spark 应用程序的运行参数。其中，"local" 表示在本地模式下运行 Spark 应用程序。然后，使用 SparkSession 对象创建了一个 SparkContext 对象，该对象用于与 Spark 集群进行通信，并启动 Spark 应用程序。需要注意的是，SparkSession 对象通常是在 PySpark 2.0 之后引入的，用于替代旧版的 SparkContext 对象。

接着，代码中使用了 SparkContext 对象的 textFile 方法读取了本地文件系统中的数据，并将其转换为 RDD 对象。在这里，我们将文件的路径作为参数传递给 textFile 方法。需要注意的是，文件路径必须是本地文件系统中的路径，如果要读取 HDFS 文件系统中的数据，则需要使用 hdfs:// 协议。

然后，代码中使用了 map 和 Row 对象对 RDD 中的每个元素进行转换，并使用 toDF 方法将转换后的 RDD 转换为 DataFrame 对象。具体来说，我们使用 map 方法对 RDD 中的每个元素进行转换，将其转换为一个 Row 对象。在这里，我们假设数据文件中每一行都包含三个字段，使用 split 方法对每一行进行分割，并将分割结果转换为一个 Row 对象。然后，使用 toDF 方法将转换后的 RDD 转换为 DataFrame 对象，并将其保存到 rowRDD 变量中。

接下来，代码中使用了 createOrReplaceTempView 方法将 DataFrame 对象注册为临时表。具体来说，我们使用 createOrReplaceTempView 方法将 rowRDD 中的数据保存到一个名为 "employee" 的临时表中，供后续查询使用。

最后，代码中使用了 Spark SQL 对 DataFrame 进行查询操作，并将查询结果转换为 RDD 对象并输出。具体来说，我们使用 spark.sql 方法对 "employee" 表进行查询操作，并将查询结果保存到 personsDF 变量中。然后，使用 rdd 方法将查询结果转换为 RDD 对象，使用 map 和 lambda 函数对 RDD 中的每个元素进行转换，并使用 foreach 和 print 方法将转换后的结果输出到控制台。在这里，我们将查询结果按照指定格式进行输出，包括 id、Name 和 age 三个字段。