Python implementation of a dataflow job for airbnb data:

This exercise is a python implementation of a dataflow job on the google cloud which ingests raw CSV file (AB_NYC_2019.csv), processes it, converts the records into a dictionary format aligned with the schema of BigQuery table (w266-final-199221:springml.airbnb) and writes to it. After the ingestion of the csv file, a second workflow in parallel processes the lines to collect neighbourhood listings (highlighted by the unique id of each record), applies the GroupByKey() transform to collect the listings by neighbourhoods, combines the frequency or count of each listing per neighbourhood, formats it according to a simplified schema and writes the summary data into another BigQuery table (w266-final-199221:springml.airbnbsummary).

Broadly it uses three google cloud components (as shown in the schematic high-level flow diagram):

1. Cloud Storage
2. Dataflow
3. BigQuery

Before embarking on the dataflow job, a pre-requisite is to enable the relevant APIs to be able to communicate with the disparate components.

The datafile from the link is unzipped and stored on the BUCKET (created expressly for this exercise - springml-airbnb) as raw data file.

A cursory EDA of the raw datafile reveals there are 48,895 records and 16 columns with one or more having missing values in several records. The header line of the raw data file is represented in the BigQuery schema below:

Raw data presents a few problematic data-wrangling challenges:

1. The second column (name) is essentially multi-line string with one or more newline ('\n') and CR ('\r') characters and potentially with multiple comma puncutations. A typical csv used a double-quote enclosure to isolate a text field but the apache_beam.io.ReadFromText breaks it into multi-records causing the data to be messed up in the dataflow ingestion step. This must be addressed beofre the historical data is ingested through the pipeline
2. The presence of double-quotes (viz., "Central Park") within the double-quotes that the csv file uses as an enclosure of the multiple commas within the string
3. The program that generated this raw data file most likely used non-unicode (most likely 'utf-8') encoding but the current dataflow uses python 2.7. This gives rise to codec encoding error when processing non-ASCII characters.

The script dataflow_file.py addresses these precise challenges by reading pandas read_csv function, deleting the potential one or more '\n' and '\r' characters within the dataframe column name, replacing the missing numerical values (NaN) with 0 and filling the missing strings fields with blanks. It then uses unix utility sed to delete non-ASCII characters from the text in-place. As a matter of abundant pre-caution, it also changes the delimiter to '\t' when writing the file lest the presence of one or more commas mess up the text within the 'name' field. It then saves the new file using 'to_csv' utility with ".txt" extension and stores it on the bucket as a new file. It is consistent with the best practice to have the original data be available. Hence the choice to write it as a separate file. Now the data is ready to be ingested.

The workflow in airbnb.py can best be represented by the following schematic workflow diagram:

The workflow involves three main steps:

1. Read in the file.
2. Process &/or Transform the CSV file into a dictionary format aligned with the BigQuery table schema.
3. Write the data to BigQuery.

Dataflow reads in each row of data from the file using the built in TextIO connector with the help of several workers in parallel and distributes it to the next stage of the data pipeline.

This allows larger file sizes and large number of input files to scale well within beam.

This is the stage of the code where a custom logic is put because each file is unique and requires custom business logic. In this exercise we are simply processing each row of the data and transforming from a CSV format into a python dictionary aligned with the destination BigQuery table. The dictionary maps column names to the values we want to store in BigQuery.

Passing the table name and a few other optional arguments into BigQueryIO sets up the final stage of the pipeline.

This stage of the pipeline is typically referred to as our sink. The sink is the final destination of data. No more processing will occur in the pipeline after this stage.

The results of the exercise, along with the screenshots of the google cloud instance are included as follows:

gcs bucket on the instance:

dataflow job on the instance:

dataflow job status on the instance:

bigquery output on the instance :