在PySpark中操作复杂的数据帧

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, expr, when

spark = SparkSession.builder.appName("DataFrameOperations").getOrCreate()

df = spark.read.format("csv").option("header", "true").load("data.csv")

# 选择特定的列
selected_columns = df.select("column1", "column2")

# 过滤数据
filtered_data = df.filter(col("column1") > 10)

# 添加新列
new_df = df.withColumn("new_column", expr("column1 + column2"))

# 修改列值
updated_df = df.withColumn("column1", when(col("column1") > 10, 1).otherwise(0))

# 计算平均值
avg_value = df.selectExpr("avg(column1)").collect()[0][0]

# 求和
sum_value = df.selectExpr("sum(column1)").collect()[0][0]

# 最大值
max_value = df.selectExpr("max(column1)").collect()[0][0]

# 内连接
inner_join_df = df1.join(df2, on="column")

# 左连接
left_join_df = df1.join(df2, on="column", how="left")

# 右连接
right_join_df = df1.join(df2, on="column", how="right")

sorted_df = df.orderBy("column1", ascending=False)

grouped_df = df.groupBy("column1").agg({"column2": "sum", "column3": "avg"})

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

window_spec = Window.partitionBy("column1").orderBy("column2")
ranked_df = df.withColumn("rank", row_number().over(window_spec))