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_salary
  FROM my_table
  GROUP BY department
  ORDER 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()

注意事项

1. 临时视图只在当前 SparkSession 中有效
2. 对于大数据集，Spark SQL 会自动优化执行计划
3. 可以混合使用 DataFrame API 和 SQL 查询
4. 确保在集群模式下正确配置了资源

PySpark 的 SQL 支持非常完整，包括大多数标准 SQL 功能以及一些 Spark 特有的扩展功能。