How to read CSV files line by line and remove mismatched datatypes rows using pyspark/python?

Anos 0
Viewed 30

I have CSV files with misplaced records issue. I have inferred/final column-datatypes for those files. So, whenever we are getting mismatched datatypes by comparing with final/inferred datatypes, it should read line by line and remove the record(row) with mismatch.

I have the following steps as a requirement,

For example,
1.read csv line by line
2.find the datatype of each column and match with inferred column-datatype
3.if there is mismatch ,then remove the record(row)
otherwise keep all good rows in valid_record_csv file

INFERRED datatypes-column(final datatype) to compare with files.
name: string
date: date
phone: long
col1: integer
col2: double

SAMPLE INPUT csv file

Name   date         phone         col1     col2
124    PANAMA       440894563              0.9800
BB     2022-9-23    449035667              9.08765
BB     GRENADA       BBCC                  PASS


SAMPLE OUTPUT csv file: 
I have one column empty/blank, but it is having inferred/final datatype as integer. Another one with double datatype

NAME   DATE       PHONE       col1    col2
BB     2022      449035667            9.08765
pyspark apache-spark-sql python-3.x python
1 Answers

pandas solution:

import re
import pandas as pd

#0. read the csv file (supposing you have csv file named 'INPUT.csv')

df = pd.read_csv('INPUT.csv')

df
    Name    date        phone       col1    col2
0   124     PANAMA      440894563   PASS    901
1   BB      2022-9-23   449035667   520     9.08765
2   BB      GRENADA     BBCC        0.536   PASS

#1. deal with columns (cleaning):
df['date'] = pd.to_datetime(df['date'], errors='coerce') # deal with date column: keep only valid dates
df['phone'] = pd.to_numeric(df['phone'], errors='coerce').astype(pd.Int64Dtype()) # deal with phone column: keep only valid integers
df['Name'] = df['Name'].map(lambda x: x if isinstance(x, str) and re.match("^[a-zA-Z\s]+$", x) else None) # deal with Name column: keep only Name : strings than only contains letters and spaces

#a. cleaning pure integers
def f_integer(x):
    if '.' in x:
        return None 
    else:
        try:
            return int(x)
        except:
            return None 

df['col1'] = df['col1'].map(f_integer).astype(pd.Int64Dtype())

#b. cleaning pure doubles
def f_double(x):
    try:
        if float(x) and ('.' in x):
            return float(x)
        else:
            return None
    except:
        return None 
df['col2'] = df['col2'].map(f_double)

#-->
    Name    date        phone       col1    col2
0   None    NaT         440894563   <NA>    NaN
1   BB      2022-09-23  449035667   520     9.08765
2   BB      NaT         NaN         <NA>    NaN

#2. remove the record(row) with mismatch:
df.loc[df['Name'].notna() & df['date'].notna() & df['phone'].notna() & df['col1'].notna() & df['col2'].notna()]

#-->
df
    Name    date        phone       col1    col2
1   BB      2022-09-23  449035667   520     9.08765

#3. save the cleaned dataframe to csv
df.to_csv('OUTPUT.csv', index=False)
khaled koubaa0
Related