代码之家 › 专栏 › 技术社区 › JQadrad

基于列值计算PySpark中以前的日期

pyspark pandas python

0

JQadrad · 技术社区 · 7 年前

我有以下数据:

client_id,transaction_id,start,end,amount
1,1,2018-12-09,2018-12-11,1000
1,2,2018-12-19,2018-12-21,2000
1,3,2018-12-19,2018-12-31,3000
2,4,2018-11-09,2018-12-20,4000
2,5,2018-12-19,2018-12-21,5000
2,6,2018-12-22,2018-12-31,6000

我试图使用PySpark添加一列,该列根据当前行的开始时间显示已完成事务的数量。我可以在Pandas中使用相当简单的代码来实现这一点,如下所示:

import pandas as pd
df = pd.read_csv('transactions.csv')
df['closed_transactions'] = df.apply(lambda row: len(df[ (df['end'] < 
row['start']) & (df['client_id'] == row['client_id'])]), axis=1)

产生数据帧

client_id   transaction_id  start   end amount  closed_transactions
0   1   1   2018-12-09  2018-12-11  1000    0
1   1   2   2018-12-19  2018-12-21  2000    1
2   1   3   2018-12-19  2018-12-31  3000    1
3   2   4   2018-11-09  2018-12-20  4000    0
4   2   5   2018-12-19  2018-12-21  5000    0
5   2   6   2018-12-22  2018-12-31  6000    2

然而,为了在PySpark中实现同样的目标,我努力让同样的东西发挥作用。我可以使用窗口函数为每个组添加一个简单的计数器,并且累计总和也可以工作,但是如果给定当前行的数据,我无法获得已结束交易的金额。

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext, Window
import pyspark.sql.functions as psf

config = SparkConf().setMaster('local')
spark = SparkContext.getOrCreate(conf=config)
sqlContext = SQLContext(spark)

spark_df = sqlContext.read.csv('transactions.csv', header=True)
window = Window.partitionBy('client_id').orderBy('start').rowsBetween(Window.unboundedPreceding, 0)

@psf.udf('string')
def get_number_of_transactions(curr_date):
    return spark_df[spark_df['end'] < curr_date].count()

spark_df \
    .withColumn('number_of_past_transactions', 
psf.row_number().over(window) - 1) \
    .withColumn('total_amount', psf.sum(psf.col('amount')).over(window)) \
   .withColumn('closed_transactions', 
get_number_of_transactions(psf.col('end'))) \
    .show()

非常感谢您的帮助!

1 回复 | 直到 7 年前

1

0

Rahul Chawla cbiqih 7 年前

client_id 并添加一个布尔列,其中 start_date<end_date start_date 取此布尔列的和。

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext, Window
import pyspark.sql.functions as psf

config = SparkConf().setMaster('local')
spark = SparkContext.getOrCreate(conf=config)
sqlContext = SQLContext(spark)

spark_df = sqlContext.read.csv('transactions.csv', header=True)

# Renaming columns for self join
df2 = spark_df
for c in df.columns:
    df2 = df2 if c == 'client_id' else df2.withColumnRenamed(c, 'x_{cl}'.format(cl=c))

# Joining with self on client ID
new_df = spark_df.join(df2, 'header')

# Creating the flag column and summing it by grouping on start_date
new_df = new_df.withColumn('valid_transaction', when(col('start_date')<col('x_end_date'),1).otherwise(0)).groupBy(['client_id', 'start_date']).agg(sum('valid_transaction'))