PySpark 中使用 SQL 语句和表进行计算
PySpark 中使用 SQL 语句和表进行计算
PySpark 完全支持使用 SQL 语句和表进行 Spark 计算。以下是几种常见的使用方式：
1. 使用 Spark SQL
from pyspark.sql import SparkSession# 创建 SparkSession
spark = SparkSession.builder.appName("SQLExample").getOrCreate()# 创建 DataFrame
data = [("Alice", 34), ("Bob", 45), ("Cathy", 29)]
df = spark.createDataFrame(data, ["name", "age"])# 将 DataFrame 注册为临时视图
df.createOrReplaceTempView("people")# 执行 SQL 查询
result = spark.sql("SELECT name, age FROM people WHERE age > 30")
result.show()
2. 直接读取数据源为表
# 读取 CSV 文件并注册为表
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)
df.createOrReplaceTempView("my_table")# 执行复杂 SQL 查询
spark.sql("""SELECT department, AVG(salary) as avg_salaryFROM my_tableGROUP BY departmentORDER BY avg_salary DESC
""").show()
3. 使用全局临时视图（跨会话）
# 创建全局临时视图
df.createOrReplaceGlobalTempView("global_people")# 在其他 SparkSession 中访问（需要指定 global_temp 数据库）
spark.sql("SELECT * FROM global_temp.global_people").show()
4. 与 Hive 表集成
如果配置了 Hive 支持，可以直接查询 Hive 表：
# 查询已存在的 Hive 表
spark.sql("SELECT * FROM hive_database.hive_table").show()
注意事项
- 临时视图只在当前 SparkSession 中有效
- 对于大数据集，Spark SQL 会自动优化执行计划
- 可以混合使用 DataFrame API 和 SQL 查询
- 确保在集群模式下正确配置了资源
PySpark 的 SQL 支持非常完整，包括大多数标准 SQL 功能以及一些 Spark 特有的扩展功能。