如何用SQL/Python/PySpark统计薪资涨幅≥20%的员工ID及涨薪次数

首先约定员工薪资流水表核心字段（对齐通用薪资表结构）：

• emp_id：员工唯一标识ID
• salary：当期实际发放薪资金额
• pay_date：薪资发放日期，用于判定薪资调整的先后顺序

核心逻辑：按员工维度分组，按发薪时间排序取上一期薪资作为涨薪计算基数，当期涨幅 =（当期薪资 - 上期薪资）/ 上期薪资，再基于涨幅阈值做筛选统计。

PySpark 实现代码 #

from pyspark.sql import Window
import pyspark.sql.functions as F

# 读取薪资表数据，替换为实际数据源读取逻辑即可（支持读Hive表、CSV、Parquet等格式）
salary_df = spark.table("employee_salary")

# 定义窗口规则：同一员工按发薪日期从早到晚排序
emp_window = Window.partitionBy("emp_id").orderBy("pay_date")

# 计算每条薪资记录对应的上期薪资、当期涨薪幅度
salary_calc_df = salary_df.withColumn(
    "prev_salary",
    F.lag("salary", 1).over(emp_window)
).withColumn(
    "growth_rate",
    (F.col("salary") - F.col("prev_salary")) / F.col("prev_salary")
)

# 需求1：筛选出单次薪资涨幅≥20%的员工ID（去重）
qualified_emp_df = salary_calc_df.filter(
    F.col("growth_rate") >= 0.2
).select("emp_id").distinct()

# 需求2：统计符合条件的员工累计获得≥20%幅度涨薪的总次数
rise_count_df = salary_calc_df.filter(
    F.col("growth_rate") >= 0.2
).groupBy("emp_id").agg(
    F.count("*").alias("high_rise_total_times")
)

# 输出结果，可根据需要替换为写入表/文件的逻辑
# qualified_emp_df.show()
# rise_count_df.show()

SQL 实现参考 #

逻辑和PySpark版本完全一致，基于窗口函数实现：

-- 需求1：查询单次涨薪≥20%的员工ID
SELECT DISTINCT emp_id
FROM (
    SELECT
        emp_id,
        (salary - LAG(salary, 1) OVER(PARTITION BY emp_id ORDER BY pay_date)) 
        / LAG(salary, 1) OVER(PARTITION BY emp_id ORDER BY pay_date) AS growth_rate
    FROM employee_salary
) t
WHERE growth_rate >= 0.2;

-- 需求2：统计每位符合条件员工累计≥20%涨薪的次数
SELECT
    emp_id,
    COUNT(1) AS high_rise_total_times
FROM (
    SELECT
        emp_id,
        (salary - LAG(salary, 1) OVER(PARTITION BY emp_id ORDER BY pay_date)) 
        / LAG(salary, 1) OVER(PARTITION BY emp_id ORDER BY pay_date) AS growth_rate
    FROM employee_salary
) t
WHERE growth_rate >= 0.2
GROUP BY emp_id;

说明：上述逻辑自动过滤了每个员工的第一条薪资记录（无上期薪资，不存在涨薪对比基准），如果业务规则需要将入职定薪纳入特殊涨薪统计，可自行调整判断逻辑。

内容的提问来源于stack exchange，提问作者Surya Pratap