3
\$\begingroup\$

What I am trying to do, is to get weather data, data about bike trips and populate the postgresql database in order to be able to work with this data from Power BI/Tableau/any other tool.

I have two different data sources. The first of them is yr.no API, which I am using to access weather data. The second one is oslobysykkel.no, from where I get data related to bike trips.

The goal is to have this initial load done with docker containers and azure blob storages, as I would like to learn more about them. But that would be the second stage. I have not really done any "ETL-ish" processing in python before, so I would love to get some feedback from you related to my code and flow.

The current structure looks like that:

enter image description here

I am not really sure if that looks OK but I was thinking about having two different docker containers(one for extract/process part and the second one for loading the data to postgresql.

The first one to be created is calendar table:

import pandas as pd
from datetime import datetime
import os
"""
Simple script, using pandas library to create a date table. One time job.
"""
local_path_processed = os.environ.get('LOCAL_PATH_PROCESSED')
start_date = '2010-01-01'
end_date = '2030-12-31'
"""
#Additional settings for pandas to make printouts more clear in the console, it's getting easier to debug.  
pd.set_option('display.max_columns',12)
desired_width=320
pd.set_option('display.width', desired_width)
"""

def create_date_table(start_date, end_date):
  df = pd.DataFrame({"date_hour": pd.date_range(start_date, end_date,freq='H')})
  df["date"] = df.date_hour.dt.date
  df["hour"] = df.date_hour.dt.hour
  df["day"] = df.date_hour.dt.day
  df["day_name"] = df.date_hour.dt.day_name()
  df["week"] = df.date_hour.dt.weekofyear
  df["quarter"] = df.date_hour.dt.quarter
  df["year"] = df.date_hour.dt.year
  df["year_half"] = (df.quarter + 1) // 2
  return df

timestampStr = datetime.now().strftime("%d-%b-%Y (%H:%M:%S.%f)")
"""
Create date table
"""
date_df = create_date_table(start_date,end_date)
"""
Add date id, ETL timestamp and write down data
"""
date_df["date_id"] = date_df.date_hour.apply(lambda x: x.strftime('%Y%m%d%H')).astype(int)
date_df["etl_timestamp"] = datetime.now().strftime("%d-%b-%Y (%H:%M:%S.%f)")
date_df_sorted = date_df[ ['date_id'] + [ col for col in date_df.columns if col != 'date_id' ] ]
date_df_sorted.to_csv(local_path_processed + 'date_table.csv',index=False)

Table with bike trips. First, I am getting raw files using selenium(one file is one month) and then I am doing some simple transformations.

import pandas as pd
import glob, os
import time
"""
#Additional settings for pandas to make printouts more clear in the console, it's getting easier to debug.  
pd.set_option('display.max_columns',12)
desired_width=320
pd.set_option('display.width', desired_width)
"""

"""
Get raw data using selenium and oslo bysykkel website. It takes some time to download data depending on your internet connectione.
Therefore I have used time(sleep) in order to avoid running further processing on partial data.
"""
from selenium.webdriver.chrome.options import Options
from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager

local_path_raw = os.environ.get('LOCAL_PATH_RAW')
local_path_processed = os.environ.get('LOCAL_PATH_PROCESSED')

remote_path = "https://developer.oslobysykkel.no/apne-data/historisk"

def download_data(remote_path, local_path, options,month_range):
    driver = webdriver.Chrome(ChromeDriverManager().install(), options=options)
    driver.command_executor._commands["send_command"] = ("POST", '/session/$sessionId/chromium/send_command')
    params = {'cmd': 'Page.setDownloadBehavior', 'params': {'behavior': 'allow', 'downloadPath': local_path}}
    driver.execute("send_command", params)
    driver.implicitly_wait(5)
    driver.get(remote_path)
    for month in range(1,month_range):
        driver.find_element_by_xpath('//*[@id="__next"]/div/div[2]/div/div/article/ul[1]/li[{}]/div/div[2]/a[2]/button'.format(month)).click()

op = Options()
op.add_argument('--disable-notifications')
op.add_experimental_option("prefs",{
    "download.prompt_for_download": False,
    "download.directory_upgrade": True,
    "safebrowsing.enabled": True
})

download_data(remote_path, local_path_raw, op, 15)
time.sleep(30)
"""
Processing part:
  Merging all raw csv files into one dataframe.
  Processing of dataframe, adding all columns that I use for the reporting layer.
"""
os.chdir(local_path_raw)
counter = 0
for file in glob.glob("*.csv"):
    if os.stat(local_path_raw+"\{}".format(file)).st_size <= 264:
        pass
    else:
        if counter == 0:
            bike_trip_df = pd.read_csv(file)
        else:
            bike_trip_df.append(pd.read_csv(file))
        counter += 1

def process_df(dataframe):
    dataframe['bike_trip_id'] = dataframe.index
    dataframe['started_at_floor'] = pd.to_datetime(dataframe['started_at']).dt.floor(freq='H')
    dataframe['ended_at_floor'] = pd.to_datetime(dataframe['ended_at']).dt.floor(freq='H')
    dataframe['date_id'] = dataframe.apply(lambda x: list(pd.date_range(x['started_at_floor'], x['ended_at_floor'], freq="1H")), axis='columns')
    dataframe = dataframe.explode('date_id')
    dataframe['date_id'] = dataframe['date_id'].dt.strftime('%Y%m%d%H')
    return dataframe

bike_trip_df = process_df(bike_trip_df)
bike_trip_df.to_csv(local_path_processed+"bike_trip.csv",sep=";")

Then a table with weather observations. I am getting hourly data for each day. YYYYMMDDHH is also a key that I want to use in my data model to connect everything. In the next stage, I would like to use azure blob storages instead of local memory, so that I can create those independent docker images as well:

import requests
import pandas as pd
import datetime
from datetime import datetime, timedelta
from dateutil import parser
import os
"""
#Additional settings for pandas to make printouts more clear in the console, it's getting easier to debug.  
pd.set_option('display.max_columns',12)
desired_width=320
pd.set_option('display.width', desired_width)
"""
start_date = os.environ.get('START_DATE_WEATHER')
end_date = os.environ.get('END_DATE_WEATHER')
local_path_processed = os.environ.get('LOCAL_PATH_PROCESSED')
def get_date_range(begin, end):
    beginDate = parser.parse(begin)
    endDate =  parser.parse(end)
    delta = endDate-beginDate
    numdays = delta.days + 1
    dayList = [datetime.strftime(beginDate + timedelta(days=x), '%m-%d-%Y') for x in range(0, numdays)]
    return dayList

list_of_dates = get_date_range(start_date,end_date)

def call_api(list_of_dates):
    row_values = []
    for date in list_of_dates:
        try:
            raw_json = requests.get('https://www.yr.no/api/v0/locations/1-72837/observations/{}'.format(date)).json()
            for day in raw_json.get('historical').get('days'):
                for hour in day.get('hours'):
                    row_object = {}
                    for key, value in hour.items():
                        try:
                            row_object[key] = next(iter(value.values()))
                        except:
                            row_object[key] = value
                    for key, value in row_object.items():
                        try:
                            if len(value) == 0:
                                row_object[key] = None
                        except:
                            pass
                    row_values.append(row_object)
        except:
            pass
    return process_dataframe(row_values)

def process_dataframe(row_values):
    df = pd.DataFrame(row_values)
    df['date'] = pd.to_datetime(df['time'])
    df['date_id'] = df.date.apply(lambda x: x.strftime('%Y%m%d%H'))
    df['rush_hour'] = df.date_id.apply(
        lambda x: "Yes" if (int(x[:-2]) in range(6, 10) or int(x[-2:])) in range(15, 19) else "No")
    return df

observation_df = call_api(list_of_dates)
observation_df["etl_timestamp"] = datetime.now().strftime("%d-%b-%Y (%H:%M:%S.%f)")
observation_df.to_csv(local_path_processed + "weather_observation.csv",sep=";")
    

And finally, I am writing data to three different tables in my postgresql database.

import psycopg2.extras
import pandas as pd
import io
import psycopg2
import os

base_path = os.environ.get('BASE_PATH')
database = os.environ.get('DATABASE')
username = os.environ.get('USERNAME')
password = os.environ.get('PASSWORD')
host = os.environ.get('HOST')

def db_connect (db_parm, username_parm, host_parm, pw_parm):
    credentials = {'host': host_parm, 'database': db_parm, 'user': username_parm, 'password': pw_parm}
    conn = psycopg2.connect(**credentials,cursor_factory=psycopg2.extras.RealDictCursor)
    conn.autocommit = True
    cur = conn.cursor()
    print ("Connected Successfully to DB: " + str(db_parm) + "@" + str(host_parm))
    return conn, cur

def db_insert(filename, table_name, file_path, conn, cur):
    dataframe = pd.read(file_path+filename)
    output = io.StringIO()
    dataframe.to_csv(output, sep='\t', header=True, index=False)
    output.seek(0)
    copy_query = "COPY {} FROM STDOUT csv DELIMITER '\t' NULL ''  ESCAPE '\\' HEADER ".format(table_name)  # Replace your table name in place of mem_info
    cur.copy_expert(copy_query, output)
    conn.commit()


conn, cur = db_connect(database, username, host, password)

db_insert("filename", "date", base_path, conn, cur)
db_insert("filename", "weather_observation", base_path, conn, cur)
db_insert("filename", "bike_trip", base_path, conn, cur)

Thank you in advance for any feedback & suggestions!

\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

Shadowing

You declare these globals:

start_date = '2010-01-01'
end_date = '2030-12-31'

and also these parameters:

def create_date_table(start_date, end_date):

That is confusing; the local parameters will take priority. One way to distinguish the two is to capitalize the global constants, which is standard anyway.

snake_case

timestampStr should be timestamp_str.

Selenium

I don't know a lot about the website, but a brief visit makes it seem like this is simple enough for you to avoid Selenium - which tries to emulate a browser - and do direct HTTP using the Requests library plus BeautifulSoup, which will be much more efficient.

pathlib

This:

os.stat(local_path_raw+"\{}".format(file))

will be simplified using pathlib.Path(local_path_raw).

Exception swallowing

This:

    except:
        pass

is extremely dangerous. It will prevent user break (Ctrl+C) from working, and will hide anything going wrong in that section of the code - even if it's a critical failure. At the absolute least, except Exception instead of except, and ideally print what's gone wrong.

os.environ.get

You don't seem to be treating these parameters as optional; you don't provide defaults. So this will create some failures later than they should occur. Use [] instead to move the failure up to a point where it's more obvious that a parameter is missing.

Autocommit

Since you've enabled this, why do you also

conn.commit()

?

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.