# 安裝PySpark
!pip install pyspark
# 導(dǎo)入必要的庫(kù)
from pyspark.sql import SparkSession
import pyspark.sql.functions as F  # 常用函數(shù)
import pyspark.sql.types as T      # 數(shù)據(jù)類(lèi)型

spark = SparkSession.builder \
    .appName("MySparkApp") \    # 設(shè)置應(yīng)用名稱(chēng)
    .config("spark.driver.memory", "2g") \  # 配置參數(shù)（可選）
    .getOrCreate()

# 檢查Spark版本
print(spark.version)
# 查看當(dāng)前配置
print(spark.conf.getAll())
# 列出數(shù)據(jù)庫(kù)和表
spark.sql("SHOW DATABASES").show()
spark.sql("USE your_database")  # 切換數(shù)據(jù)庫(kù)
spark.sql("SHOW TABLES").show()

# 方式1：從Python列表創(chuàng)建
data = [("Alice", 29), ("Bob", 35)]
columns = ["name", "age"]
df = spark.createDataFrame(data, schema=columns)
# 方式2：指定詳細(xì)模式（Schema）
schema = T.StructType([
    T.StructField("name", T.StringType(), True),
    T.StructField("age", T.IntegerType(), True)
])
df = spark.createDataFrame(data, schema=schema)

# 讀取CSV文件（常用）
df = spark.read.csv(
    "path/to/file.csv",
    header=True,        # 第一行作為列名
    inferSchema=True    # 自動(dòng)推斷列類(lèi)型
)
# 讀取Parquet文件（列式存儲(chǔ)，高效）
df = spark.read.parquet("path/to/file.parquet")
# 讀取JSON文件
df = spark.read.json("path/to/file.json")

# 保存為Parquet格式（推薦，壓縮率高）
df.write.mode("overwrite").parquet("output_path.parquet")
# 保存為CSV格式
df.write.mode("overwrite") \
    .option("header", True) \
    .csv("output_path.csv")
# 保存為Spark表（可在集群中持久化）
df.write.saveAsTable("table_name")

# 選擇特定列
df.select("name", "age").show()
# 創(chuàng)建新列（示例：年齡加1）
df = df.withColumn("age_plus_one", F.col("age") + 1)
# 重命名列
df = df.withColumnRenamed("old_name", "new_name")
# 更改列類(lèi)型（示例：整型轉(zhuǎn)字符串）
df = df.withColumn("age_str", F.col("age").cast(T.StringType()))
# 刪除列
df = df.drop("column_to_remove")

# 過(guò)濾行（示例：年齡大于30）
df_filtered = df.filter(F.col("age") > 30)
# 等價(jià)寫(xiě)法
df_filtered = df.where(df["age"] > 30)
# 排序
df_sorted = df.orderBy(F.col("age").desc())  # 按年齡降序

# 刪除包含任何空值的行
df_clean = df.dropna()
# 填充缺失值（示例：用0填充特定列）
df_filled = df.fillna({"age": 0, "name": "Unknown"})

# 基礎(chǔ)分組聚合（示例：按部門(mén)計(jì)算平均工資）
df.groupBy("department").agg(
    F.avg("salary").alias("avg_salary"),
    F.count("*").alias("employee_count")
).show()

# 假設(shè)有另一個(gè)部門(mén)信息表df_dept
df_joined = df.join(
    df_dept, 
    on="department_id",  # 連接鍵
    how="inner"          # 連接方式：inner, left, right, outer等
)

from pyspark.sql.window import Window
# 定義窗口：按部門(mén)分區(qū)，按工資降序
window_spec = Window.partitionBy("department").orderBy(F.col("salary").desc())
# 計(jì)算部門(mén)內(nèi)工資排名
df.withColumn("salary_rank", F.rank().over(window_spec)).show()

from pyspark.sql.functions import udf
# 定義Python函數(shù)
def categorize_age(age):
    return "Young" if age < 30 else "Senior"
# 注冊(cè)為UDF（需指定返回類(lèi)型）
categorize_udf = udf(categorize_age, T.StringType())
# 應(yīng)用UDF
df.withColumn("age_group", categorize_udf(F.col("age"))).show()

# 將DataFrame注冊(cè)為臨時(shí)視圖
df.createOrReplaceTempView("people")
# 執(zhí)行SQL查詢(xún)
result = spark.sql("""
    SELECT department, AVG(salary) as avg_sal
    FROM people 
    WHERE age > 25
    GROUP BY department
    ORDER BY avg_sal DESC
""")
result.show()

df.cache()  # 將DataFrame緩存到內(nèi)存
df.unpersist()  # 使用后釋放緩存

spark.stop()

from pyspark.sql import SparkSession
import pyspark.sql.functions as F
# 1. 初始化
spark = SparkSession.builder.appName("Example").getOrCreate()
# 2. 讀取數(shù)據(jù)
df = spark.read.csv("sales.csv", header=True, inferSchema=True)
# 3. 數(shù)據(jù)處理
result = (df
    .filter(F.col("amount") > 100)           # 篩選大額交易
    .groupBy("region", "product")            # 按地區(qū)和產(chǎn)品分組
    .agg(F.sum("amount").alias("total_sales")) # 計(jì)算總銷(xiāo)售額
    .orderBy(F.col("total_sales").desc())    # 按銷(xiāo)售額降序排序
)
# 4. 輸出
result.show()
result.write.mode("overwrite").parquet("sales_summary.parquet")
# 5. 清理
spark.stop()