构建新申请致旧申请驳回的指标字段技术问询

需求：创建new_application_causes_rejection字段，识别同一personal_id下，某条记录的rejected_time出现在其他记录的creation_timestamp之后5分钟内的场景，满足则标记为1，否则为0。数据规模为数十万personal_id，多数对应多个application_id，各application_id下行数不一致。

示例数据：

解决方案 #

1. SQL实现（适用于BigQuery、Snowflake等大数据数据库） #

核心思路：先提取每个personal_id下唯一的申请创建时间，再关联原表判断每条驳回记录是否落在其他申请的5分钟窗口内。

WITH unique_applications AS (
    SELECT DISTINCT
        personal_id,
        application_id,
        creation_timestamp
    FROM your_table
),
rejection_check AS (
    SELECT
        t.personal_id,
        t.application_id,
        t.rejected_time,
        CASE
            WHEN EXISTS (
                SELECT 1
                FROM unique_applications ua
                WHERE ua.personal_id = t.personal_id
                  AND ua.application_id != t.application_id
                  AND t.rejected_time BETWEEN ua.creation_timestamp AND DATE_ADD(ua.creation_timestamp, INTERVAL 5 MINUTE)
            ) THEN 1
            ELSE 0
        END AS new_application_causes_rejection
    FROM your_table t
)
SELECT
    t.*,
    rc.new_application_causes_rejection
FROM your_table t
JOIN rejection_check rc
    ON t.personal_id = rc.personal_id
    AND t.application_id = rc.application_id
    AND t.rejected_time = rc.rejected_time;

2. PySpark实现（适用于超大规模分布式数据集） #

利用分组收集其他申请的创建时间，通过内置函数判断时间窗口匹配情况，兼顾性能与逻辑清晰。

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

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

# 读取原始数据
df = spark.read.table("your_table")

# 分组收集每个用户下所有其他申请的创建时间（去重）
other_app_times = df.select("personal_id", "application_id", "creation_timestamp").distinct() \
    .groupBy("personal_id") \
    .agg(collect_set("creation_timestamp").alias("other_creation_times"))

# 关联并计算标记字段
result_df = df.join(other_app_times, on="personal_id", how="left") \
    .withColumn(
        "new_application_causes_rejection",
        when(
            expr("exists(other_creation_times, x -> rejected_time between x and x + interval 5 minutes)"),
            1
        ).otherwise(0)
    ) \
    .drop("other_creation_times")

# 输出结果
result_df.show()

关键注意事项 #

• 确保时间字段为时间戳类型，若为字符串需先用to_timestamp转换。
• 大数据量场景下，SQL可按personal_id分区优化查询；PySpark可调整并行度或广播小表提升效率。
• 同一application_id下的多条驳回记录会独立判断，符合需求逻辑。

内容的提问来源于stack exchange，提问作者Alex Günsberg