Spark SQL 基本操作

创建SparkSession是使用Spark SQL的第一步，需要导入相关依赖并初始化：

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("SparkSQLDemo").getOrCreate()

创建DataFrame

从Python列表创建DataFrame：

data = [("Alice", 34), ("Bob", 45), ("Cathy", 29)]
columns = ["name", "age"]
df = spark.createDataFrame(data, columns)
df.show()

从外部数据源读取数据：

df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)
df = spark.read.json("path/to/file.json")
df = spark.read.parquet("path/to/file.parquet")

DataFrame基本操作

查看数据结构和内容：

df.printSchema()
df.show(5)

选择特定列：

df.select("name", "age").show()

过滤数据：

df.filter(df["age"] > 30).show()

注册临时视图执行SQL查询

将DataFrame注册为临时视图：

df.createOrReplaceTempView("people")

执行SQL查询：

result = spark.sql("SELECT name, age FROM people WHERE age > 30")
result.show()

聚合操作

使用DataFrame API进行聚合：

df.groupBy("department").agg({"salary": "avg", "age": "max"}).show()

使用SQL进行聚合：

spark.sql("""
    SELECT department, AVG(salary), MAX(age)
    FROM employees
    GROUP BY department
""").show()

数据写入操作

将处理结果写入不同格式文件：

df.write.csv("output_path.csv")
df.write.json("output_path.json")
df.write.parquet("output_path.parquet")

性能优化技巧

缓存常用DataFrame：

df.cache()

分区处理大数据：

df.repartition(4).write.parquet("partitioned_data.parquet")

广播小表提高连接性能：

from pyspark.sql.functions import broadcast
df1.join(broadcast(df2), "join_key")