<jupyter_start><jupyter_text>Transcript from pandas to SparkThe Notebook transcripts the methods applied in pandas for the featureengineering generation into a Spark Dataframe and methods applied to resultinto the same, or similar outcome.At the end of the notebook the original df result will be compared with thespark functions outcome to compare and accept if the resulting functions aresufficient to accept them as a valid result.  Import dependencies and read data<jupyter_code>import pandas as pd

from pyspark.sql import functions as f
from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import rank
from pyspark.sql.functions import monotonically_increasing_id, current_date, \
    date_add, when, lit, col, months_between, floor

spark = SparkSession \
    .builder \
    .appName("Credit Features Processing") \
    .getOrCreate()
df = pd.read_csv('../data/raw_data/dataset_credit_risk.csv')
df_func = spark.read.csv("../data/raw_data/dataset_credit_risk.csv",
                         header=True,
                         sep=',',
                         inferSchema=False)<jupyter_output><jupyter_text>Sort and cast<jupyter_code>df = df.sort_values(by=["id", "loan_date"])
df = df.reset_index(drop=True)
df["loan_date"] = pd.to_datetime(df.loan_date)
# spark
def order_by_id_and_loan_date(data_frame):
    data_frame = data_frame.orderBy("id", "loan_date")
    # Index only to compare with pandas
    data_frame = data_frame.withColumn("index",
                                       monotonically_increasing_id())
    return data_frame

def cast_dates_to_datetime(data_frame):
    # loan_date
    data_frame = data_frame.withColumn('loan_date',
                                       data_frame['loan_date'].cast('date'))
    data_frame = data_frame.withColumn('loan_date',
                                       when(col("loan_date")
                                            <= date_add(current_date(), 1),
                                            col("loan_date")))
    #birthday
    data_frame = data_frame.withColumn('birthday',
                                       data_frame['birthday'].cast('date'))
    data_frame = data_frame.withColumn('birthday',
                                       when(col("birthday")
                                            <= date_add(current_date(), 1),
                                            col("birthday")))
    # job_start_date
    data_frame = data_frame.withColumn('job_start_date',
                                       data_frame['job_start_date'].cast('date'))
    data_frame = data_frame.withColumn('job_start_date',
                                       when(col("job_start_date")
                                            <= date_add(current_date(), 1),
                                            col("job_start_date")))
    return data_frame

df_func = order_by_id_and_loan_date(df_func)
df_func = cast_dates_to_datetime(df_func)<jupyter_output><empty_output><jupyter_text>Feature nb_previous_loans<jupyter_code>df_grouped = df.groupby("id")
df["nb_previous_loans"] = df_grouped["loan_date"].rank(method="first") - 1
df["nb_previous_loans"] = df["nb_previous_loans"].astype('int')
def generate_nb_previous_loans(data_frame):
    window = Window.partitionBy("id").orderBy("loan_date")
    data_frame = data_frame.withColumn("nb_previous_loans",
                                       rank().over(window) -1)
    return data_frame
df_func = generate_nb_previous_loans(df_func)<jupyter_output><empty_output><jupyter_text>Feature avg_amount_loans_previous<jupyter_code>def avg_amount_loans_prev(df):
    avg = pd.Series(index=df.index)
    for i in df.index:
        df_aux = df.loc[df.loan_date < df.loan_date.loc[i], :]
        avg.at[i] = df_aux.loan_amount.mean()
    return avg
%%time  
avg_amount_loans_previous = pd.Series()
# the following cycle is the one that takes forever if we try to compute it for the whole dataset
for user in df.id.unique():
    df_user = df.loc[df.id == user, :]
    avg_amount_loans_previous = avg_amount_loans_previous.append(avg_amount_loans_prev(df_user))

df["avg_amount_loans_previous"] = avg_amount_loans_previous

%%time
def generate_avg_amount_loans_previous(data_frame):
    window = Window.partitionBy("id"). \
        rowsBetween(Window.unboundedPreceding, -1)

    data_frame = data_frame.sort(f.asc('loan_date')) \
        .withColumn("avg_amount_loans_previous",
                    f.mean('loan_amount').over(window))
    return data_frame

df_func = generate_avg_amount_loans_previous(df_func)<jupyter_output>CPU times: user 3.49 ms, sys: 4.01 ms, total: 7.5 ms
Wall time: 120 ms<jupyter_text>Feature age<jupyter_code>df['birthday'] = pd.to_datetime(df['birthday'], errors='coerce')
df['age'] = (pd.to_datetime('today').normalize() -
             df['birthday']).dt.days // 365

def compute_age_features(data_frame):
    data_frame = data_frame.withColumn("age",
                          floor(months_between(current_date(),
                                               col("birthday"))/lit(12))
                          .cast('integer'))
    return data_frame

df_func = compute_age_features(df_func)<jupyter_output><empty_output><jupyter_text>Feature years_on_the_job<jupyter_code>df['job_start_date'] = pd.to_datetime(df['job_start_date'], errors='coerce')
df['years_on_the_job'] = (pd.to_datetime('today').normalize() -
                          df['job_start_date']).dt.days // 365
def compute_years_on_the_job(data_frame):
    data_frame = data_frame.withColumn("years_on_the_job",
                          floor(months_between(current_date(),
                                               col("job_start_date"))/lit(12))
                          .cast('integer'))
    return data_frame

df_func = compute_years_on_the_job(df_func)<jupyter_output><empty_output><jupyter_text>Feature flag_own_car<jupyter_code>df['flag_own_car'] = df.flag_own_car.apply(lambda x : 0 if x == 'N' else 1)
def add_feature_own_car_flag(data_frame):
    data_frame = data_frame.withColumn('flag_own_car',
                                       when(col("flag_own_car") == 'N',
                                            lit(0)).otherwise(1))
    return data_frame

df_func = add_feature_own_car_flag(df_func)<jupyter_output><empty_output><jupyter_text>Select subset of columns<jupyter_code>df = df[['id',
         'age',
         'years_on_the_job',
         'nb_previous_loans',
         'avg_amount_loans_previous',
         'flag_own_car', 'status']]
def credit_fe_columns_subset(data_frame):
    data_frame = data_frame.select('id',
                             'age',
                             'years_on_the_job',
                             'nb_previous_loans',
                             'avg_amount_loans_previous',
                             'flag_own_car',
                             'status')
    return data_frame

df_func = credit_fe_columns_subset(df_func)<jupyter_output><empty_output><jupyter_text>Compare Results<jupyter_code>df_copy = df.copy()
# df = df_copy.copy()<jupyter_output><empty_output><jupyter_text>Order both dataset to do a 1:1 comparinson<jupyter_code>df_func = df_func.orderBy('index') # Index only to compare with pd.Dataframe
df = df.sort_index() # pd.Dataframe
df_test = df_func.toPandas() # Transform PySpark dataframe into pd.Dataframe
df_test.info(verbose=True) # PySpark convert to pandas Resulting schema
df.info(verbose=True) # pd.Dataframe resulting schema<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 777715 entries, 0 to 777714
Data columns (total 7 columns):
 #   Column                     Non-Null Count   Dtype  
---  ------                     --------------   -----  
 0   id                         777715 non-null  int64  
 1   age                        777715 non-null  int64  
 2   years_on_the_job           649743 non-null  float64
 3   nb_previous_loans          777715 non-null  int64  
 4   avg_amount_loans_previous  740750 non-null  float64
 5   flag_own_car               777715 non-null  int64  
 6   status                     777715 non-null  int64  
dtypes: float64(2), int64(5)
memory usage: 41.5 MB<jupyter_text>Due to the differences presented in the schemas, the pyspark dataframe will becasted into the schema of the pd.Dataframe in order to reproduce the resultsand make them comparable 1:1. This is decided after seeing a comparinsonwith no nulls in both outcomes.<jupyter_code>df_test["id"] = df_test["id"].astype(int)
df_test["age"] = df_test["age"].astype(int)
df_test["flag_own_car"] = df_test["flag_own_car"].astype(int)
df_test["nb_previous_loans"] = df_test["nb_previous_loans"].astype(int)
df_test["status"] = df_test["status"].astype(int)<jupyter_output><empty_output><jupyter_text>Cast float64 fields and round to same data type and same number of decimalsto avoid resolution misscomparison.<jupyter_code># Round years to zero decimals
df_test["years_on_the_job"] = \
    df_test["years_on_the_job"].astype('float64').round(0)
df["years_on_the_job"] = df["years_on_the_job"].round(0)

# Round float fields to 2 decimals
df_test["avg_amount_loans_previous"] = \
    df_test["avg_amount_loans_previous"].astype('float64').round(2)
df["avg_amount_loans_previous"] = df["avg_amount_loans_previous"].round(2)

df_test.info(verbose=True)<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 777715 entries, 0 to 777714
Data columns (total 7 columns):
 #   Column                     Non-Null Count   Dtype  
---  ------                     --------------   -----  
 0   id                         777715 non-null  int64  
 1   age                        777715 non-null  int64  
 2   years_on_the_job           649743 non-null  float64
 3   nb_previous_loans          777715 non-null  int64  
 4   avg_amount_loans_previous  741258 non-null  float64
 5   flag_own_car               777715 non-null  int64  
 6   status                     777715 non-null  int64  
dtypes: float64(2), int64(5)
memory usage: 41.5 MB<jupyter_text>Assertion of a 1:1 comparison<jupyter_code>print("The outcomes are different than expected: ", df_test.equals(df))
print("Are the shapes of both datasets the same?:", (df.shape==df_test.shape))<jupyter_output>The outcomes are different than expected:  False
Are the shapes of both datasets the same?: True<jupyter_text>Because of side effects of the python floor division and traditional roundfunction in the years calculated in pandas the fields with those values willbe dropped after seing how many records have differences between bothdataframes and adress those fields are the discrepancy causes, just to followthe asumption that the round vs floor libraries are the cause.Disclaimer: The floor function was denoted bevause the age of a person has tobe quantizied after the birthday and not after 6 months or any other arbitraryvalue for the months.<jupyter_code>import numpy as np
def test_differences(df1, df2):
    different_records_df=pd.concat([df1,df2]).drop_duplicates(keep=False)
    print("Head of df of not matching records")
    print(different_records_df.sort_index().head(100))

    # Both different records are kept for the same row
    different_records = different_records_df.shape[0]/2
    total_expected_records = df.shape[0]
    difference_percentage = \
        np.round(different_records/total_expected_records*100,2)
    print(f"Difference percentage between DF: {difference_percentage}%")
test_differences(df, df_test)<jupyter_output>Head of df of not matching records
           id  age  years_on_the_job  nb_previous_loans  \
125   5008811   52               8.0                 27   
125   5008811   52               8.0                 26   
130   5008811   52               8.0                 32   
130   5008811   52               8.0                 31   
212   5008821   48               3.0                  1   
...       ...  ...               ...                ...   
2632  5008980   61               NaN                  9   
2662  5008980   61               NaN                 39   
2662  5008980   61               NaN                 38   
2723  5008984   57               6.0                 26   
2723  5008984   57               6.0                 25   

      avg_amount_loans_previous  flag_own_car  status  
125                      131.11             0       0  
125                      130.47             0       0  
130                      127.70             0       0  
130                      128.85 [...]<jupyter_text>Following the hypothesis of the Age difference, we will drop this column andrepeat the previous test of finding different records and calculate thedifferences.<jupyter_code>df_no_head = df.drop("age", axis=1)
df_test_no_head = df_test.drop("age", axis=1)
test_differences(df_no_head , df_test_no_head )<jupyter_output>Head of df of not matching records
           id  years_on_the_job  nb_previous_loans  avg_amount_loans_previous  \
125   5008811               8.0                 27                     131.11   
125   5008811               8.0                 26                     130.47   
130   5008811               8.0                 32                     127.70   
130   5008811               8.0                 31                     128.85   
212   5008821               3.0                  1                        NaN   
...       ...               ...                ...                        ...   
2632  5008980               NaN                  9                     128.43   
2662  5008980               NaN                 39                     123.78   
2662  5008980               NaN                 38                     123.17   
2723  5008984               6.0                 26                     136.27   
2723  5008984               6.0                 25                     135[...]<jupyter_text>The difference percentage reduced dramatically, proposing the years computingis the causing of the differences potentially. We will remove the years onjob field and repeat the test.<jupyter_code>df_no_years_job= df_no_head.drop("years_on_the_job", axis=1)
df_test_no_years_job= df_test_no_head.drop("years_on_the_job", axis=1)
test_differences(df_no_years_job , df_test_no_years_job)<jupyter_output>Head of df of not matching records
           id  nb_previous_loans  avg_amount_loans_previous  flag_own_car  \
125   5008811                 27                     131.11             0   
125   5008811                 26                     130.47             0   
130   5008811                 31                     128.85             0   
130   5008811                 32                     127.70             0   
212   5008821                  0                     100.02             1   
...       ...                ...                        ...           ...   
2632  5008980                  9                     128.43             0   
2662  5008980                 39                     123.78             0   
2662  5008980                 38                     123.17             0   
2723  5008984                 26                     136.27             1   
2723  5008984                 25                     135.74             1   

      status  
125        0  
125       [...]
{'fraction_non_alphanumeric': 0.08564480084133036, 'fraction_numerical': 0.05284606283686079, 'mean_word_length': 1.9612689762553523, 'pattern_counts': {'":': 0, '<': 63, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 71, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '598214', 'license': 'MIT', 'n_tokens_mistral': 5045, 'n_tokens_neox': 4386, 'n_words': 1007, 'path': 'notebooks/Features engineering by functions.ipynb', 'repo_name': 'samuelrojolopez/ml_technical_challenge', 'size': '15214'}