how to use values of consequent rows of grouped data to decide value of current row using pyspark

user_10268554 0
Viewed 48

Of the dataset below I want to change the value of the won_offer column to either a 1 or a 0. The problem is that I need the consequent rows of the customer-code combination to decide the value of that column.

If one of the next rows within 30 days of the current row's date contains order and the price is lower than the current row's price, the 0 of the won_offer column of that row can become a 1.

sample dataset:

analysis = sqlContext.createDataFrame(
    [
        ('customer1', 'code1', 'date', 'order', 1.7, 0, 1),
        ('customer1', 'code2', 'date', 'offer', 1.5, 0, 2),
        ('customer1', 'code2', 'date', 'offer', 2.0, 0, 2),
        ('customer2', 'code1', 'date', 'offer', 1.2, 0,4),
        ('customer2', 'code1', 'date', 'order', 1.1, 0,4),
        ('customer2', 'code1', 'date', 'order', 2.0, 0,4),
        ('customer2', 'code1', 'date', 'offer', 1.2, 0,4)
    ],
    ('customer', 'code', 'order_date', 'type', 'price', 'final_offer', 'counter')
)

I tried something like this but it doesn't work, since I don't know how to pass multiple rows to my udf:

w = \
    Window.partitionBy('customer','code').orderBy('orderoffer_date')

@F.udf(returnType=IntegerType())
def logic_udf(counter, curr_date, next_dates, current_type, next_types, curr_price, next_prices) :
    for i in range(len(counter)):
        if (next_dates[i] < curr_date+30):
            if (next_types[i] == 'order') & (next_prices[i] < curr_price ):
                return 1
            else:
                return 0
        else:
            return 0

analysis = analysis.withColumn('won_offer', 
               logic(analysis.counter, analysis.order_date,lead(analysis.order_date, 
               analysis.n).over(w), analysis.type,lead(analysis.type, 
               analysis.n).over(w), analysis.price, lead(analysis.price, 
               analysis.n).over(w)))

desired output:

desired_result = sqlCtx.createDataFrame(
    [
        ('customer1', 'code1', 'date', 'order', 1.7, 0, 1),
        ('customer1', 'code2', 'date', 'offer', 1.5, 0, 2),
        ('customer1', 'code2', 'date', 'offer', 2.0, 0, 2),
        ('customer2', 'code1', 'date', 'offer', 1.2, 1, 4),
        ('customer2', 'code1', 'date', 'order', 1.1, 1, 4),
        ('customer2', 'code1', 'date', 'order', 1.0, 0, 4),
        ('customer2', 'code1', 'date', 'offer', 1.2, 0, 4)
    ],
    ('customer', 'code', 'order_date', 'type', 'price', 'final_offer', 'counter')
)

I realize my question is quite complex. If someone could just tell me how I could pass multiple rows of grouped data to a udf, I would already be helped a lot.

In short: the main goal is to decide the value of a column in a row by looking at multiple columns in the next rows (and still within their specific group).

Thanks in Advance! Charles

pyspark python
1 Answers

You can use window with sql functions to replace logic_udf. Since You are only use first row after the current row you can add first row after current row to current row.

from pyspark.sql import functions as F

analysis \
    .withColumn('next_order_date', F.first('order_date').over(w)) \
    .withColumn('next_type', F.first('type').over(w)) \
    .withColumn('next_price', F.first('price').over(w)) \
    .withColumn('won_offer', F.when(condition, 1).otherwise(0))
user_72800290
Related