主页 / coding / 问题 / 79367947
Accepted
mabanalyst
Asked: 2025-01-19 04:45:15 +0800 CST

PySpark 过滤

  • 772

目前，我正在对包含借款人如何偿还贷款的信息的数据库进行计算。

从技术角度来看，我正在使用 PySpark，刚刚面临如何使用高级过滤操作的问题。

我的数据框如下所示：

ID     ContractDate LoanSum  ClosingDate Status Bank
ID3    2024-06-10   20                   Active A
ID3    2024-06-11   30                   Active A
ID3    2024-06-12   50                   Active A
ID3    2024-06-12   15                   Active B
ID3    2024-06-12   5        2024-06-18  Closed A
ID3    2024-06-13   40       2024-06-20  Closed A
ID3    2024-06-22   50                   Active A
ID4    2024-07-11   20                   Active A
ID4    2024-07-12   30                   Active B
ID4    2024-07-13   50                   Active B
ID4    2024-07-11   5        2024-08-20  Closed A

我的目标是通过“贷款金额”字段计算借款人的总和，这些借款人拥有同一家银行在首笔贷款发放之日起 3 天内发放的 3 笔或更多有效贷款。

在我的例子中，ID3 的总和为 20 + 30 + 50 = 100

我目前所做的：

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

df = spark.createDataFrame(data).toDF('ID','ContractDate','LoanSum','ClosingDate', 'Status', 'Bank')
df.show()

cols = df.columns
w = Window.partitionBy('ID').orderBy('ContractDate')

df.withColumn('PreviousContractDate', f.lag('ContractDate').over(w)) \
  .withColumn('Target', f.expr('datediff(ContractDate, PreviousContractDate) >= 0 & datediff(ContractDate, PreviousContractDate) <= 3')) \
  .withColumn('Target', f.col('Target') | f.lead('Target').over(w)) \
  .filter('Target == True')

此代码仅有助于根据 ContractDate 获取向一名借款人发放的贷款。

我如何添加更多条件？

pyspark

1 个回答

  1. Best Answer

    要解决您的问题，请按照以下代码操作。作为示例，我使用上述数据框。

    代码：

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

data1 = [
    ('ID3', '2024-06-10', 20, None, 'Active', 'A'),
    ('ID3', '2024-06-11', 30, None, 'Active', 'A'),
    ('ID3', '2024-06-12', 50, None, 'Active', 'A'),
    ('ID3', '2024-06-12', 15, None, 'Active', 'B'),
    ('ID3', '2024-06-12', 5, '2024-06-18', 'Closed', 'A'),
    ('ID3', '2024-06-13', 40, '2024-06-20', 'Closed', 'A'),
    ('ID3', '2024-06-22', 50, None, 'Active', 'A'),
    ('ID4', '2024-07-11', 20, None, 'Active', 'A'),
    ('ID4', '2024-07-12', 30, None, 'Active', 'B'),
    ('ID4', '2024-07-13', 50, None, 'Active', 'B'),
    ('ID4', '2024-07-11', 5, '2024-08-20', 'Closed', 'A'),
]


df12 = spark.createDataFrame(data1, ['ID', 'ContractDate', 'LoanSum', 'ClosingDate', 'Status', 'Bank']) \
    .withColumn('ContractDate', f.to_date('ContractDate')) \
    .filter(f.col('Status') == 'Active')

# Use  window function and calculate cumulative count
w = Window.partitionBy('ID', 'Bank').orderBy('ContractDate')
df = df12.withColumn('CumulativeCount', f.sum(
    f.when(f.datediff(f.col('ContractDate'), f.lag('ContractDate').over(w)).isNull(), 1)
    .when(f.datediff(f.col('ContractDate'), f.lag('ContractDate').over(w)) <= 3, 1)
    .otherwise(0)
).over(w))

df1 = df.filter(f.col('CumulativeCount') >= 3).groupBy('ID', 'Bank').agg(f.sum('LoanSum').alias('TotalLoanSum'))


display(df1)

    输出：

    在此处输入图片描述

    • 1

相关问题