ETL Pipeline Description

The image below illustrates our simple ETL work flow. It’s worth noting that Airflow records the task and DAG logging automatically in SQLite by default. However, if you wish to store the record logs in a different SQL environment, Airflow has some support for that as well. We’ll be using the default event and task logging that comes out of the box.

Let’s go over the rest of the system components. The Seattle Public data will be coming from the Socrata API endpoint that is utilized by the city of Seattle. This is a standard API interface where you can easily query data based on multiple criterion. The Apache Airflow instance is something that we’ll delve into further in this article. It’s basically just a very flexible and robust system that allows you to schedule and deploy bash and python scripts. It also allows you to establish “connection hooks” to external data sources, which is very helpful when managing data pipelines to and from various systems. One great thing about Airflow, is that it has a nifty GUI, that allows you to audit event logging, and keep track of prior deployment statuses.

The destination of our data will be a PostgreSQL database instance, that I have spun up on RDS in AWS. It’s pretty easy to do, and you don’t have to use a Postgres database. RDS on AWS supports MariaDB, MySQL, Oracle, and MSSQL Server. I have already created the database tables we’ll be funneling data into. You could use a tool like SQLWorkbenchJ to perform a simple query to create the table (but you’ll have to explicitly declare the classes), or you could import some sample data into the database via psycopg2 beforehand, just to make sure you avoid any data type errors.

ETL Diagram

Set Up

Let’s set up the ETL pipeline to schedule and deploy. To install Airflow, I used a neat article written by David Barlow, that you can find here. Some familiarity with AWS is needed. In particular, you have to have some knowledge of deploying EC2 (Elastic Computing Cloud) instances, setting up permissions in a VPC (Virtual Computing Cloud), and working with a Linux CLI (Command Line Interface). David Barlow recommends using a Jupyter Notebook environment in conjunction with Airflow, which I also prefer. In addition to this, it’s recommended that you install airflow, and the jupyter notebook services in a separate python virtual environment. This will help isolate any dependency errors or inconsistencies resulting from package updates.

The set up of airflow, the EC2 instance, and the VPC, can be quite time consuming. The most frustrating, and annoying thing to deal with though is also the perpetual source of chagrin for most developers and data folk: database drivers. Oh boy, don’t get me started on database drivers. Most of them cause a significant amount of pain and anguish, but none however are as terrible, inconceivably frustrating, and cause as much torture as MSSQL drivers. Since we’re utilizing PostgreSQL in this project, I’ll avoid expanding further on this topic. Once you have installed a database driver, you can set up a what’s known as a “Connection Hook” in Airflow. A connection hook is just another name for a connection to an external interface. The great thing about it, is that once it’s set up, using it is as easy as calling a method. Another bonus is that you’re able to run ad-hoc queries on your connection hook via Airflow’s web front-end.

I’ve already configured the connection hook to my PostgreSQL database. In the screen capture below, I’m running a simple query where I count all of the 911 incident “types” in the CleanFireCalls table.

Ad-Hoc Query

Tasks

The Extraction and Loading tasks illustrated in the diagram above, will be accomplished by the triggering of two tasks which we’ll refer to as tasks run_insert_f911 and Inserting_distinct. These tasks trigger two functions InsertingTheCalls and clean_firecalls respectively. Let’s take a look at InsertingTheCalls:

def InsertingTheCalls(ds, **kwargs):
### Connect to postgres server
    postgresserver = PostgresHook('Postgres Jcaro')
### Connect to seattle public data portal via Socrata
    client = Socrata("data.seattle.gov",
                   "AxejqJVsnt6qbNhpmUDUtz8um",
                   username=username_SOC,
                   password=password_SOC)
##Test print date
    print(datetime.date.today())
    now_pt = datetime.datetime.now() - datetime.timedelta(hours=8)
    filter_time = now_pt - datetime.timedelta(minutes = 20)

#filter_time = filter_time.isoformat()
    formatted_filter = filter_time.strftime("%Y-%m-%dT%H:%M:%S")
    formatted_filter = str(formatted_filter)
## Test filter print
    print(formatted_filter)
    filter = "datetime >= '" + formatted_filter + "'"
    print(filter)
### Get data after or equal to filter time 
    results = client.get("kzjm-xkqj",where = filter, limit=50)

# Convert to pandas DataFrame
    results_df = pd.DataFrame.from_records(results)
    results_df = results_df.loc[:,"address":"incident_number"]
    calls_df = results_df
    calls_df["report_location"] = "point"
    calls_df.rename({"report_location":"report_location.type"})
    print(calls_df)

### Convert to tuples for inserting
    calls_tuple = [tuple(r) for r in calls_df.to_numpy()]
    
### Insert into postgres
    postgresserver.insert_rows("public.seattlefirecalls", calls_tuple,commit_every =1)

First we begin by defining our connection hook postgresserver to point to the PostgreSQL database. Then, we define our connection to the SOCRATA API (using sodapy). We print the current date, and then subtract 8 hours to the current system time (difference between UTC and PST). We plan on querying the SOCRATA API for the all of the calls in the previous 75 minutes, so we declare filter_time as the current time in PST, minus 75 minutes. We then format filter_time to a string which is compatible with the SOCRATA API endpoint, and create the actual filter string, subsequently calling the endpoint for the pertinent data, and storing the data in results_df.

Here’s what the results from the query look like:

results_df.head()

Afterwards, we subset the dataframe results_df, subsequently coercing it to a tuple and inserting it into the PostgreSQL table public.seattlefirecalls via Airflow’s insert_rows() method.

This task scheduled to run every 15 minutes. Since there is a slight lag in execution time between the scheduled start time, and the actual start time, I add a buffer of 5 minutes to the data I query from the SOCRATA API. It’s worth noting there is a slight misnomer between what one might define to be an execution time, and what Airflow designates as the execution time. In Airflow, if I run a task every 15 minutes, a task scheduled to be executed at 12:45 PM won’t actually fire until around 1:00 PM. The reason is described here on stack overflow.

Since I built in the 5 minute buffer we run the risk of inserting duplicate data into the table. To resolve this issue, we can proceed two ways. We can use the python environment to query the table we’re inserting the data into and do an except join with our new data. This would effectively filter out the duplicate data already contained in our PostgreSQL table. Another way to resolve this issue, is to use an auxiliary table in our PostgreSQL database, and have our postgres instance do this for us through a quick except (left anti-join) SQL statement. I have chosen the latter, due to a restriction of resources on my EC2 instance.

The task aptly titled Inserting_distinct, triggers the function clean_firecalls. Let’s take a look at the function:

def clean_firecalls(ds, **kwargs):
### Define connection to postgres DB
    postgresserver = PostgresHook('Postgres Jcaro')
#### Insert into the clean table
    postgresserver.run("insert into public.cleanfirecalls (select * from public.seattlefirecalls except / (select * from public.cleanfirecalls))", autocommit = True)
#### Delete data after insert    
    postgresserver.run("delete from public.seattlefirecalls where true", autocommit = True)

After defining the function, I define my connection hook as postgresserver, and then I run two commands. First, I tell the database to insert data from public.seattlefirecalls that is not already present in public.cleanfirecalls. Then I simply delete the data in public.cleanfirecalls, as it’s already been inserted into public.cleanfirecalls.

Tools

Now that we have successfully deployed our ETL pipeline, we can take a look at some neat tools Airflow offers. The first tool is the task monitor. It allows you to see the duration of each task in your DAG. In the image below, you’ll see some irregular spikes in duration for the task InsertingTheCalls. It may be due to a large influx of size in the data we’re writing to our Postgres database. Although, since there are no failures, and the task durations cap out at 10 seconds, it’s not really worth investigating.

TaskMonitor

From the Airflow landing page, you can start or stop your ETL pipelines, check if any have failed, see which are running, and check the scheduled interval.

Airflow Landing

Cursory Analysis

We can easily query our database for any of the data we have loaded onto it, without affecting any API limits. In python, this involves using a library like psycopg2, and connecting to the database where we stored the data. Our database is in “the cloud”, but accessing a database in the cloud is just as easy as accessing one hosted locally on your machine.

conn = psycopg2.connect(
    host= host_addr,
    database="testing",
    user="jcaropostgres",
    password=user_pw)
    
def sqlQuery(querystr):
    getquer = pd.read_sql(querystr, conn)
    return(getquer)
    
firecalls_df = sqlQuery("SELECT * FROM public.cleanfirecalls where datetime < '2020-11-28' and datetime >= '2020-11-13'")

top5_types = firecalls_df.groupby("type", as_index = False).\
        agg(typecount = pd.NamedAgg("type", "count")).\
        sort_values("typecount", ascending = False).iloc[1:6].type

After connecting to our database, we define a quick function that makes repeated querying much easier. We’ll call it sqlQuery, and all we have to supply to it is the query string. That is the query statement in a string data type. Here, we’ll query all of the data from the table public.cleanfirecalls between (and including) 2020-11-13 and 2020-11-27. We then declare top5_types, as a series of the top 5 highest incident type by ocurrences.

firecalls_df = firecalls_df.loc[:,["type", "datetime", "incident_number"]][firecalls_df["type"].\
            isin(top5_types)]

firecalls_df = firecalls_df.assign(date = lambda x: x["datetime"].dt.date)

firecalls_count_df = firecalls_df.groupby(["type", "date"], as_index = False).\
            agg(type_count = pd.NamedAgg("date", "count"))



type_line_plot = ggplot(firecalls_count_df, aes(x = "date", y = "type_count", group = "type", color = "type"))\
                    + geom_line()\
                    + scale_x_datetime(breaks=date_breaks('3 days'))\
                    + theme(dpi = 350, aspect_ratio= .85, subplots_adjust={'right': 0.75},\
                            axis_text_x= element_text(angle = 25, hjust = 1))


print(type_line_plot)

## <ggplot: (316581289)>

We filter our data to only include incidents in the top5_types series, and perform a simple type count, by date. We can then plot a color coded line-plot that illustrates the quantity of the top 5 incident types by date. This is just a simple example ilustrating how easy it is to perform some analysis on the data stored in your database.