<jupyter_start><jupyter_text>1.0 Libraries For Preprocessing<jupyter_code>import time
start = time.time()
print("Importing libraries for the preprocessing")
# Library for Wrangling and loading
import pandas as pd
import numpy as np

# Libraries for Visualization
import seaborn as sns

import matplotlib.pyplot as plt
import plotly.graph_objects as go
import plotly.express as px

# Libraries for Partitioning and Spark
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import shutil
import hashlib

# Others
import warnings
import os


# Configuration
pd.set_option("display.max_columns",None)
pd.set_option("display.max_rows",None)

warnings.filterwarnings("ignore")
end = time.time()
print("Imported all libraries | Time Taken {}sec".format(str(end-start)))<jupyter_output>Importing libraries for the preprocessing
Imported all libraries | Time Taken 31.777028799057007sec<jupyter_text>2.0 Partitioning  2.1 Partitioning Parameter<jupyter_code>N_PARTITION = 50    # Number of buckets
base_partitions_dir = "../data/external/Partition"
output_dir = "../data/external/output"<jupyter_output><empty_output><jupyter_text>2.2 Collecting data columns<jupyter_code># LOADING PARAMETER
chunksize= 1e5
data_directory = r"..\Data\external\used_cars_data.csv"
# loading chunk of the data for the column list
def col_list(data_path, chunksize, progress=None):
    print("Checking and loading file")
    time.sleep(0.9)
    for df_iter, chunk in enumerate(pd.read_csv(data_path, chunksize=chunksize, iterator=False)):
        pass
    col_list = list(chunk.columns)
    return col_list<jupyter_output><empty_output><jupyter_text>2.3 Function for Hashing id<jupyter_code># hashing the listing id to allow even partitioning across the dataset
def hash_(listing_id):
    """Creates an hashed column using the listing id for the vehicle"""
    return int(hashlib.md5(str(listing_id).encode("utf-8")).hexdigest(), 16)<jupyter_output><empty_output><jupyter_text>2.4 Creating dir for partition<jupyter_code>def create_partition():
    """Creates an empty partition directory for the buckets"""
    start = time.time()
    print("Checking if the directory exists...")
    time.sleep(0.9)
    if os.path.exists(base_partitions_dir):
        print("Directory found")
        time.sleep(0.4)
        print("Removing directory")
        time.sleep(1)
        shutil.rmtree(base_partitions_dir)
        print("Removed the directory")
    else:
        print("No Such Directory found.")

    # Delaying before creating the directories
    time.sleep(2.5)

    print("Creating empty folder list for partition")
    time.sleep(0.9)
    if not os.path.exists(base_partitions_dir):
        # Creating partition directory
        os.mkdir(base_partitions_dir)
        # Making a new directory for the partitions
        for i in range(N_PARTITION):
            partition_path = os.path.join(
                base_partitions_dir, "p{}".format(i)).replace("\\", "/")
            # Printing the path
            print('| {} | Partition left {} |'.format(partition_path,N_PARTITION-i))
            if not os.path.exists(partition_path):
                os.mkdir(partition_path)
            else:
                print("Path Already exist")
            time.sleep(0.6)
    end = time.time()
    print("| Completed | Time Taken ------------------------- {}sec |".format(str(end-start)))
# Making the directory
dir = create_partition()<jupyter_output>Checking if the directory exists...
Directory found
Removing directory
Removed the directory
Creating empty folder list for partition
| ../data/external/Partition/p0 | Partition left 50 |
| ../data/external/Partition/p1 | Partition left 49 |
| ../data/external/Partition/p2 | Partition left 48 |
| ../data/external/Partition/p3 | Partition left 47 |
| ../data/external/Partition/p4 | Partition left 46 |
| ../data/external/Partition/p5 | Partition left 45 |
| ../data/external/Partition/p6 | Partition left 44 |
| ../data/external/Partition/p7 | Partition left 43 |
| ../data/external/Partition/p8 | Partition left 42 |
| ../data/external/Partition/p9 | Partition left 41 |
| ../data/external/Partition/p10 | Partition left 40 |
| ../data/external/Partition/p11 | Partition left 39 |
| ../data/external/Partition/p12 | Partition left 38 |
| ../data/external/Partition/p13 | Partition left 37 |
| ../data/external/Partition/p14 | Partition left 36 |
| ../data/external/Partition/p15 | Partition left 3[...]<jupyter_text>2.5 Creating blank partition<jupyter_code>chunksize= 1e5
data_directory = r"..\Data\external\used_cars_data.csv"
def create_blank_partition():
    """Creating a blank partition with the number of bucket"""
    start = time.time()
    data_list = col_list(data_directory, chunksize)
    for i in range(N_PARTITION):
        time.sleep(0.9)
        file_base_dir = os.path.join(base_partitions_dir,"p{}".format(str(i)),"").replace("\\","/")
        print(file_base_dir)
        # Opening the file and writing it to the partition created
        with open(file_base_dir+"vehicle_used_data.csv", "w") as f:
            f.write(",".join(data_list))
    end = time.time()
    print("Time taken ------------------- | {}sec".format(str(end-start)))
    return file_base_dir

dir_path = create_blank_partition()<jupyter_output>Checking and loading file
../data/external/Partition/p0/
../data/external/Partition/p1/
../data/external/Partition/p2/
../data/external/Partition/p3/
../data/external/Partition/p4/
../data/external/Partition/p5/
../data/external/Partition/p6/
../data/external/Partition/p7/
../data/external/Partition/p8/
../data/external/Partition/p9/
../data/external/Partition/p10/
../data/external/Partition/p11/
../data/external/Partition/p12/
../data/external/Partition/p13/
../data/external/Partition/p14/
../data/external/Partition/p15/
../data/external/Partition/p16/
../data/external/Partition/p17/
../data/external/Partition/p18/
../data/external/Partition/p19/
../data/external/Partition/p20/
../data/external/Partition/p21/
../data/external/Partition/p22/
../data/external/Partition/p23/
../data/external/Partition/p24/
../data/external/Partition/p25/
../data/external/Partition/p26/
../data/external/Partition/p27/
../data/external/Partition/p28/
../data/external/Partition/p29/
../data/external/Partiti[...]<jupyter_text>2.6 Partitioning by hashing<jupyter_code>N_PARTITION = 50    # Number of buckets
# Partitioing and hashing the 
def partition_by_hashing(df, progress=None):
    # hashing the listing_id column into the number of partitions
    df["hashed"] = df["listing_id"].apply(hash_) % N_PARTITION
    for partitions, data in df.groupby("hashed"):
        start = time.time()
        print("Data partition: {}".format(str(partitions)))
        # Removing hash columns from the dataset
        data = data.drop("hashed", axis=1)
        # Wrting the data to the partition
        path_dir = os.path.join(base_partitions_dir,"p{}".format(partitions),"").replace("\\","/")
        print("| writing data partition to {} |".format(path_dir))
        # Writing the data to the path
        with open(path_dir+"vehicle_used_data.csv", "a", encoding="utf-8") as f:
            f.write("\n")
            data.to_csv(f, header=False, index=False)
            
        end = time.time()
        print("Time taken {}sec".format(str(end-start)))<jupyter_output><empty_output><jupyter_text>2.7 Writing data to each partition<jupyter_code>chunksize = 1e5
for df_iter, data in enumerate(pd.read_csv(r"..\Data\external\used_cars_data.csv", iterator=True, chunksize=chunksize, encoding="latin1"),1):
    print("Iter:",df_iter)
    partition = partition_by_hashing(df=data)
    print("\n")
    #data = partition_by_hashing(df, name="listing_id", progress=None)<jupyter_output>Iter: 1


Data partition: 0
| writing data partition to ../data/external/Partition/p0/ |
Time taken 1.7771995067596436sec
Data partition: 1
| writing data partition to ../data/external/Partition/p1/ |
Time taken 0.3827812671661377sec
Data partition: 2
| writing data partition to ../data/external/Partition/p2/ |
Time taken 0.4177563190460205sec
Data partition: 3
| writing data partition to ../data/external/Partition/p3/ |
Time taken 0.45574069023132324sec
Data partition: 4
| writing data partition to ../data/external/Partition/p4/ |
Time taken 0.3867764472961426sec
Data partition: 5
| writing data partition to ../data/external/Partition/p5/ |
Time taken 0.3787803649902344sec
Data partition: 6
| writing data partition to ../data/external/Partition/p6/ |
Time taken 0.40177106857299805sec
Data partition: 7
| writing data partition to ../data/external/Partition/p7/ |
Time taken 0.47272825241088867sec
Data partition: 8
| writing data partition to ../data/external/Partition/p8/ |
Time taken 0[...]<jupyter_text>3.0 Data Distribution with Spark   3.1 Data cleaning in parallel(configuration)  4.0 Data Cleaning  4.1 Loading first Partition<jupyter_code># LOADING FIRST PARTITION FILE
PARTITION = 0
TARGET = "price"
PARTITION_DIR = os.path.join("..\Data\external\Partition", "p{}".format(str(PARTITION)),"vehicle_used_data.csv").replace("\\","/")

used_data = pd.read_csv(PARTITION_DIR, header=0, delimiter=",")
used_data.head(2)
data = used_data.copy()
def clean_data(data):
    """Removing some irrelevant columns in the dataframe"""
    

    # Defining some columns to remove
    cols = ["vin", 'description', "exterior_color", "wheel_system", "vehicle_damage_category", "trimId",
            "theft_title", "sp_id", "main_picture_url", "longitude", "listing_id", "listing_color", "latitude",
                "interior_color", "cabin", "major_options", "back_legroom", "bed", "bed_height", "bed_length", 
                 "is_certified","owner_count","is_cpo", "is_oemcpo", "salvage", "wheelbase", "width","combine_fuel_economy",
                     "daysonmarket","dealer_zip","engine_cylinders","horsepower",
                    "franchise_dealer","front_legroom","fuel_tank_volume","height","length","franchise_make","savings_amount","transmission_display","trim_name","sp_name"
                    ]
    # Dropping the columns
    data = data.drop(columns=cols)
    # data = data.drop(cols, axis=1)
    
    # listed date to pandas datetime
    data["listed_date"] = pd.to_datetime(data["listed_date"])
    
    # Transforming
    data["transmission"] = data["transmission"].apply(lambda inf: str(inf).replace("A","Automatic").replace("M","Manual"))
    data["Vehicle_power"] = data["power"].apply(lambda inf: str(inf).split("@")[0].strip().split(" ")[0])
    data["Vehicle_torque"] = data["torque"].apply(lambda inf: str(inf).split("@")[0].strip().split(" ")[0])
    data["max_seating"] = data["maximum_seating"].apply(lambda inf: str(inf).strip().split(" ")[0])
    data["Listing_year"] = data["listed_date"].apply(lambda inf: inf.year)
    data["engine_displacement"] = data["engine_displacement"].apply(lambda inf: inf/1000)
    data["combined_MPG"] = (data["highway_fuel_economy"] + data["city_fuel_economy"])/2
    
    # Replacing any "nan" with np.nan value ===> Null
    data = data.replace({"nan":np.nan, "_": np.nan,"--":np.nan})
    
    # Changing datatype
    data["Vehicle_power"] = data['Vehicle_power'].astype(np.float).astype("Int32")
    data["Vehicle_torque"] = data['Vehicle_torque'].astype(np.float).astype("Int32")
    data["max_seating"] = data['max_seating'].astype(np.float).astype("Int32")


    # data["isCab"] = pd.DataFrame(data["isCab"])
    # data["isCab"] = data["isCab"].map({"False":False,"True":True})
    
    
    # Dropping some more columns
    data = data.drop(columns=["listed_date","maximum_seating","power",'torque'])
    # Renaming some columns
    data = data.rename(columns={"wheel_system_display":"Drivetrain",
                                "transmission":"Transmission",
                                "body_type":"body_style",
                                "city_fuel_economy":"city_MPG",
                                "engine_displacement":"engine_size",
                                "highway_fuel_economy":"highway_MPG",
                                "year": "Vehicle_year"})
    
    
    return data
    
def clean_data(data):
    """C
    
    """
    # Filtering the data
    filtered_data = data[data["make_name"].value_counts() > 10000]
    # Dropping some columns
    filtered_data = filtered_data.drop(
        ["Unnamed: 0", "listed_date", "listed_year"], axis=1)

    # Data Cleaning
    filtered_data["fuel_tank_volume"] = filtered_data["fuel_tank_volume"].apply(
        lambda volume: int(volume.strip().split(" ")[0]))
    filtered_data["maximum_seating"] = filtered_data["maximum_seating"].apply(
        lambda seating: int(seating.strip().split(" ")[0]))
    filtered_data["transmission"] = filtered_data["transmission"].apply(
        lambda trans: trans.replace("A", "Automatic").replace("M", "Manual"))
    filtered_data["engine_displacement"] = filtered_data["engine_displacement"].apply(
        lambda disp: disp/1000)
    filtered_data["engine_cylinders"] = filtered_data["engine_cylinders"].apply(
        lambda cyl: cyl.strip().split(" ")[0].strip())
    filtered_data["mpg"] = (
        filtered_data["city_fuel_economy"] + filtered_data["highway_fuel_economy"])/2

    # Renaming some columns
    filtered_data = filtered_data.rename(columns={"fuel_tank_volume": "Tank_Volume",
                                                  "engine_displacement": "engine_size",
                                                  "city_fuel_economy": "MPG city",
                                                  "highway_fuel_economy": "MPG highway",
                                                  "wheel_system": "Drivetrain"})
    return filtered_data

new_data = clean_data(data)
new_data.head(2)
for feature in new_data.columns:
    if new_data[feature].dtype == object:
        print(feature, new_data[feature].nunique())<jupyter_output>body_style 9
city 3660
engine_type 33
fleet 2
frame_damaged 2
fuel_type 8
has_accidents 2
isCab 2
make_name 57
model_name 819
Transmission 4
Drivetrain 5<jupyter_text>Dealing with the missing values<jupyter_code>def impute_missing_values(new_data, verbose: int):
    """Dealing with the missing values by imputation"""
    # Performing input validation
    if (verbose not in [0, 1]):
        raise TypeError("Code ran into an Exception \
                        Because verbose is either a string or not 0 or 1")
    from sklearn.experimental import enable_iterative_imputer
    from sklearn.impute import IterativeImputer
    from sklearn.ensemble import RandomForestRegressor
    from sklearn.preprocessing import LabelEncoder
    from sklearn.neighbors import KNeighborsRegressor
    cat_fea = [
        feature for feature in new_data.columns if new_data[feature].dtype == object]
    num_missing_fea = [feature for feature in new_data.columns if new_data[feature].isnull(
    ).sum() > 0 and new_data[feature].dtype != object]
    not_null_fea = [
        feature for feature in new_data.columns if new_data[feature].isnull().sum() == 0]

    # Label encoding the categorical feature
    le = LabelEncoder()
    print("Encoding the categorical feature")
    for feature in cat_fea:
        new_data[feature] = le.fit_transform(new_data[feature])
        
    # imputing the missing features
    estimator = RandomForestRegressor(random_state=42)
    estimator_neighbour = KNeighborsRegressor(n_neighbors=5)
    print("Imputing the missing values")
    imputer = IterativeImputer(
        estimator=estimator, max_iter=7, verbose=verbose, random_state=42
    )
    imputer.fit(new_data)
    transformed = imputer.transform(new_data)
    transformed_data = pd.DataFrame(transformed, columns=new_data.columns)
    
    # Reverting the encoded cat features
    print("Reverting encoded feature to original")
    for feature in cat_fea:
        transformed_data[feature] = le.inverse_transform(
            transformed_data[feature])

    return transformed_data
used_data.isnull().sum()
data.isnull().sum()
# SPARK PARAMETER

# SPARK_HOME = r"C:\spark\spark" # The spark location
# LOG_DIR = r"C:\spark\tmp\spark-event" # Eventlogging location

import findspark
findspark.init()

# Spark configuration
import pyspark
conf = pyspark.SparkConf()

# Enabling logging
conf.set("spark.eventLog.enabled", False)
conf.set("spark.eventLog.dir", r"\tmp")
conf.set("spark.eventLog.compress",True)

# Enabling all cores usage
conf.set("spark.num.executor", 1)
conf.set("spark.executor.memory", "8g")
conf.set("spark.executor.cores",2)

# setting the appname for the spark ui
conf.setAppName("Data Cleaning")

# Set parent
conf.set("spark.master", "local[2]")
conf.getAll()


sc = pyspark.SparkContext(conf=conf)
sc
sc.stop()
a = sc.parallelize(range(0,100), N_PARTITION).map(lambda x: clean_data(x))
#using the main thing 
a.collect()
SELECT FROM table *

print(list(range(0,50)))
# Parallelizing the process
start = time.time()
sc.parallelize(list(range(0,100)),N_PARTITION).map(lambda x: clean_data(x)).collect()
sc.stop()
end = time.time()
used_data.head()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13726180005863384, 'fraction_numerical': 0.024626209322779244, 'mean_word_length': 3.6259831841605643, 'pattern_counts': {'":': 17, '<': 36, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '490542', 'license': 'MIT', 'n_tokens_mistral': 5707, 'n_tokens_neox': 5343, 'n_words': 1147, 'path': 'notebooks/preprocessing.ipynb', 'repo_name': 'psalishol/new-project', 'size': '17055'}