We will do the following steps using Python

• Read several CSV, JSON and XML file types.
• Extract data from the above file types and combine it.
• Transform data.
• Save the transformed data in a ready-to-load format which data engineers can use to load into an RDBMS

 

Please Download the :datasorce.zip" Zip file and extract it. Replace “C:\Nazmul\Blog\Python\ETL\source\” to the location where you have extracted the Zip File. The following Python script is attacehed "ETL script using Python (Example 2).py"

 

##Import the required modules and functions

import glob # this module helps in selecting files

import pandas as pd # this module helps in processing CSV files

import xml.etree.ElementTree as ET # this module helps in processing XML files.

from datetime import datetime

 

### Set Path

tmpfile = "dealership_temp.tmp" # file used to store all extracted data

logfile = "dealership_logfile.txt" # all event logs will be stored in this file

targetfile = r"C:\Nazmul\Blog\Python\ETL\datasource\dealership_transformed_data.csv" #

file where transformed data is stored

 

### CSV Extract Function

def extract_from_csv(file_to_process):

dataframe = pd.read_csv(file_to_process)

return dataframe

 

### JSON Extract Function

def extract_from_json(file_to_process):

dataframe = pd.read_json(file_to_process,lines=True)

return dataframe

 

### XML Extract Function

def extract_from_xml(file_to_process):

dataframe = pd.DataFrame(columns=['car_model','year_of_manufacture','price', 'fuel'])

tree = ET.parse(file_to_process)

root = tree.getroot()

for person in root:

car_model = person.find("car_model").text

year_of_manufacture = int(person.find("year_of_manufacture").text)

price = float(person.find("price").text)

fuel = person.find("fuel").text

dataframe = dataframe.append({"car_model":car_model,

"year_of_manufacture":year_of_manufacture, "price":price, "fuel":fuel},

ignore_index=True)

return dataframe

 

### Extract Function

def extract():

extracted_data = pd.DataFrame(columns=['car_model','year_of_manufacture','price',

'fuel']) # create an empty data frame to hold extracted data

#process all csv files

for csvfile in glob.glob(r"C:\Nazmul\Blog\Python\ETL\datasource\*.csv"):

extracted_data = extracted_data.append(extract_from_csv(csvfile), ignore_index=True)

#process all json files

for jsonfile in glob.glob(r"C:\Nazmul\Blog\Python\ETL\datasource\*.json"):

extracted_data = extracted_data.append(extract_from_json(jsonfile), ignore_index=True)

#process all xml files

for xmlfile in glob.glob(r"C:\Nazmul\Blog\Python\ETL\datasource\*.xml"):

extracted_data = extracted_data.append(extract_from_xml(xmlfile), ignore_index=True)

return extracted_data

 

### Transform

#Round the price columns to 2 decimal places

def transform(data):

data['price'] = round(data.price,2)

return data

### Loading

def load(targetfile,data_to_load):

data_to_load.to_csv(targetfile)

### Logging

def log(message):

timestamp_format = '%Y-%h-%d-%H:%M:%S' # Year-Monthname-Day-Hour-Minute-Second

now = datetime.now() # get current timestamp

timestamp = now.strftime(timestamp_format)

with open("logfile.txt","a") as f:

f.write(timestamp + ',' + message + '\n')

 

###Running ETL Process

log("ETL Job Started")

log("Extract phase Started")

extracted_data = extract()

log("Extract phase Ended")

extracted_data

log("Transform phase Started")

transformed_data = transform(extracted_data)

log("Transform phase Ended")

transformed_data

log("Load phase Started")

load(targetfile,transformed_data)

log("Load phase Ended")

log("ETL Job Ended")