ETL using Python (Example 1)

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 "SOURCE.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 1).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 = "temp.tmp" # file used to store all extracted data

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

targetfile = "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=["name", "height", "weight"])

tree = ET.parse(file_to_process)

root = tree.getroot()

for person in root:

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

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

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

dataframe = dataframe.append({"name":name, "height":height, "weight":weight},

ignore_index=True)

return dataframe

### Extract Function

def extract():

extracted_data = pd.DataFrame(columns=['name','height','weight']) # create an empty data

frame to hold extracted data

#process all csv files

for csvfile in glob.glob(r"C:\Nazmul\Blog\Python\ETL\source\*.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\source\*.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\source\*.xml"):

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

return extracted_data

### Transform

#The transform function does the following tasks.

#1. Convert height which is in inches to millimeter

#2. Convert weight which is in pounds to kilograms

def transform(data):

#Convert height which is in inches to millimeter

#Convert the datatype of the column into float

#data.height = data.height.astype(float)

#Convert inches to meters and round off to two decimals(one inch is 0.0254 meters)

data['height'] = round(data.height * 0.0254,2)

#Convert weight which is in pounds to kilograms

#Convert the datatype of the column into float

#data.weight = data.weight.astype(float)

#Convert pounds to kilograms and round off to two decimals(one pound is 0.45359237

kilograms)

data['weight'] = round(data.weight * 0.45359237,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")