<jupyter_start><jupyter_text>![image.png](attachment:image.png) Task:* Take any csv/text file of 2+ GB of your choice. --- (You can do this assignment on Google colab)* Read the file ( Present approach of reading the file )* Try different methods of file reading eg: Dask, Modin, Ray, pandas and present your findings in term of computational     efficiency* Perform basic validation on data columns : eg: remove special character , white spaces from the col name* As you already know the schema hence create a YAML file and write the column name in YAML file. --define separator of     read and write file, column name in YAML* Validate number of columns and column name of ingested file with YAML.* Write the file in pipe separated text file (|) in gz format.* Create a summary of the file:    Total number of rows,    total number of columns    file size     Data Ingestion sample code walkthrough   > Create a utility file    > Config file creation    > Data ingestion pipeline<jupyter_code>import logging
import os
import subprocess
import yaml
import pandas as pd
import datetime 
import gc
import re
import time
import dask.dataframe as dd
import modin.pandas as md
import ray
%%writefile testutility.py
import logging
import os
import subprocess
import yaml
import pandas as pd
import datetime 
import gc
import re
import time
import dask.dataframe as dd
import modin.pandas as md
import ray

################
# File Reading #
################

def read_config_file(filepath):
    with open(filepath, 'r') as stream:
        try:
            return yaml.safe_load(stream)
        except yaml.YAMLError as exc:
            logging.error(exc)


def replacer(string, char):
    pattern = char + '{2,}'
    string = re.sub(pattern, char, string) 
    return string

def col_header_val(df,table_config):
    '''
    replace whitespaces in the column
    and standardized column names
    '''
    df.columns = [column.strip() for column in df.columns]
    df.columns = df.columns.str.lower()
    df.columns = df.columns.str.replace('[^\w]','_',regex=True)
    df.columns = list(map(lambda x: x.strip('_'), list(df.columns)))
    df.columns = list(map(lambda x: replacer(x,'_'), list(df.columns)))
    expected_col = list(map(lambda x: x.lower(),  table_config['columns']))
    expected_col.sort()
    df.columns =list(map(lambda x: x.lower(), list(df.columns)))
    df = df.reindex(sorted(df.columns), axis=1)
    if len(df.columns) == len(expected_col) and list(expected_col)  == list(df.columns):
        print("column name and column length validation passed")
        return 1
    else:
        print("column name and column length validation failed")
        mismatched_columns_file = list(set(df.columns).difference(expected_col))
        print("Following File columns are not in the YAML file",mismatched_columns_file)
        missing_YAML_file = list(set(expected_col).difference(df.columns))
        print("Following YAML columns are not in the file uploaded",missing_YAML_file)
        logging.info(f'df columns: {df.columns}')
        logging.info(f'expected columns: {expected_col}')
        return 0<jupyter_output>Overwriting testutility.py<jupyter_text>Write YAML file<jupyter_code>%%writefile file.yaml
file_type: csv
dataset_name: testfile
file_name: Parking_Violations_Issued_-_Fiscal_Year_2017
table_name: edsurv
inbound_delimiter: ","
outbound_delimiter: "|"
skip_leading_rows: 1
columns: 
     - Summons Number 
     - Plate ID 
     - Registration State 
     - Plate Type
     - Issue Date 
     - Violation Code 
     - Vehicle Body Type 
     - Vehicle Make
     - Issuing Agency 
     - Street Code1 
     - Street Code2 
     - Street Code3
     - Vehicle Expiration Date
     - Violation Location 
     - Violation Precinct
     - Issuer Precinct
     - Issuer Code 
     - Issuer Command 
     - Issuer Squad
     - Violation Time
     - Time First Observed 
     - Violation County
     - Violation In Front Of Or Opposite
     - House Number
     - Street Name
     - Intersecting Street 
     - Date First Observed
     - Law Section
     - Sub Division 
     - Violation Legal Code 
     - Days Parking In Effect
     - From Hours In Effect 
     - To Hours In Effect 
     - Vehicle Color
     - Unregistered Vehicle?
     - Vehicle Year 
     - Meter Number
     - Feet From Curb 
     - Violation Post Code 
     - Violation Description
     - No Standing or Stopping 
     - Violation 
     - Hydrant Violation
     - Double Parking Violation
# Read config file
import testutility as util
config_data = util.read_config_file("file.yaml")
lim = config_data['inbound_delimiter']
#inspecting data of config file
config_data<jupyter_output><empty_output><jupyter_text>Reading File with Daskcsv file size : **2.09 GB**reading time  : **1.47 seconds**<jupyter_code># Normal reading process of the file
start_time = time.time()
df_sample = dd.read_csv("Parking_Violations_Issued_-_Fiscal_Year_2017.csv",delimiter=',')
print("--- %s seconds ---" % (time.time() - start_time))
print("\n")
# read the file using config file
file_type = config_data['file_type']
source_file = "./" + config_data['file_name'] + f'.{file_type}'
#print("",source_file)
df = dd.read_csv(source_file,delimiter = lim)
#validate the header of the file
util.col_header_val(df,config_data)
print("columns of files are:" ,df.columns)
print("columns of YAML are:" ,config_data['columns'])
if util.col_header_val(df,config_data)==0:
    print("validation failed")
    # write code to reject the file
else:
    print("col validation passed")
    # write the code to perform further action
    # in the pipleine<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.0954239091876552, 'fraction_numerical': 0.004256828662646328, 'mean_word_length': 3.402029664324746, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '53951', 'license': 'MIT', 'n_tokens_mistral': 1739, 'n_tokens_neox': 1652, 'n_words': 560, 'path': 'DSVIICODE/test_ingestion_dask.ipynb', 'repo_name': 'UGURSELIMOZEN/Data_Glacier_DS_Internship', 'size': '5638'}