As 2021 turned into 2022, Homa's Data team underwent a fundamental change. To adapt to scaling up (in data volume, stakeholders' expectations, and the number of people within the group), we decided to do a sizable refactor on our data pipelines. In this article, we'd like to share some of our decisions, their results after one year, and the lessons we've learned along the way.
At the time, we were already using industry-standard infrastructure, including Redshift as the analytics database and Airflow with Kubernetes for orchestration.
However, some limitations in our codebase were starting to show:
We wanted to correct these shortcomings while keeping the things that worked and minimizing the amount of work to make it happen.
The data warehouse architecture was still the most fitting for us. It meant putting our analytics database at the center of all our data operations. For this to be possible, we needed to remove the storage limit.
We considered moving to other MPP databases that decouple storage and compute by default (Snowflake, BigQuery, or the flashy lake-house architecture with Databricks). However, such a change would require much effort, which we would rather avoid. Luckily, with the introduction of Redshift’s RA3 node, we could achieve this decoupling simply by changing our cluster configuration in a few clicks. We had to trade some computing power to get the extra storage, but it was well worth it.
A lot has been written about ETL vs. ELT. For us, it was an obvious choice to keep extraction dumb, to load data immediately to the warehouse, and to put all transformations inside the database using SQL. It makes transformation faster, more convenient, and more approachable for analysts. When we discover a data error, detecting whether it comes from the source system itself or our transformation logic is straightforward.
To help keep our data warehouse organized, logical, and thus easy to use, we have formalized a list of rules on how the data should be designed. For example:
We decided against the multi-repo structure because we envision that, in the foreseeable future, we will have only 5-10 people actively working on these data pipelines and related projects, all of which benefit from each other. We don't require the scalability that multi-repo provides and are unwilling to tolerate its difficulties. Here's a rough structure of our repo:
.
├── data_global (repo name)
│ ├── adhoc_analyses (mainly jupyter notebooks from the analyst team)
│ ├── data_raw (Python code for data extraction, populating data in Raw layer - see our data modeling principles above)
│ ├── data_integration (mainly dbt models, populating data in Integration layer - see our data modeling principles above)
│ ├── data_presentation (mainly dbt models, populating data in Presentation layer - see our data modeling principles above)
│ └── utils (both reusable Python helpers and dbt macros)
├── scripts (thin wrapper to expose our Python code as cli programs)
├── tests (both Python unit tests and dbt data tests)
└── dbt_project.yml
To assist new joiners in setting up their environment and running the code as quickly as possible, we also include an easy way to retrieve the necessary secrets from secure storage and default config files for the most used IDE in our team (VS Code). It should take only a few minutes from cloning the repo to being able to run their first application locally.
We have dozens of sources and more than a hundred reports to extract data. Therefore, we inevitably need a common framework to facilitate this process, allowing us to write configs instead of repeating boilerplate code.
Many commercial and open-source solutions already do this (a few names come to mind, such as Fivetran, Stitch, Airbytes, Singer, etc.). However, we've decided to write our framework because:
Regardless, we learned a few ideas from those open-source projects and applied them in our decision-making.
The design of our DataFetcher framework can be illustrated as follows:
Every "report" (a specific type of data that comes from a single endpoint and can be loaded into one database table) is handled by a concrete DataFetcher class. Inside this class, a concrete DataExtractor and DataUploader are defined and composed with the fetcher object. They specify how to retrieve and ingest the data into our database. The DataFetcher class inherits from BaseDataFetcher, which handles instantiation and coordination of the extractor/uploader and performs tasks such as calculating the dates to run, retrying, error handling, etc.
In detail, a concrete DataFetcher would look something like this:
class ironSourceAppsDataFetcher(BaseDataFetcher):
"""
Extract and load ironSource Apps data.
<https://developers.is.com/ironsource-mobile/air/application-api/>
"""
default_backfill_periods = 3
extractor = HTTPRequestsDataExtractor(
url="<https://platform.ironsrc.com/partners/publisher/applications/v6>",
auth=HeadersAuth(auth_headers_callable=get_auth_header),
response_parser=[JsonResponseParser()],
)
uploader = LocalToRedshiftDataUploader(
table_name="raw_ironsource_apps",
incremental_strategy = "overwrite",
)
While anyone can write a custom extractor for their use case, most of our data sources can be extracted using the standard classes, such as the HTTPRequestsDataExtractor in the above example. We have made these classes fully featured and flexible enough to adapt to most scenarios. We also have other standard extractors such as S3DataExtractor, GoogleSheetsDataExtractor, etc. So, in the future, whenever we see ourselves repeating the same code, we can quickly turn that into a standard extractor to be used in similar cases.
At the time of our refactoring, dbt has become mainstream. So it was a no-brainer for us to make dbt the central framework for managing our data warehouse pipelines. Not only is it a great tool that adds real value, but there is also a growing number of tools, especially in data quality, catalog, and observability, that support easy integration with dbt-backed warehouses.
Project structure and data model
As mentioned above, we put the dbt code in the same repo with everything else in our pipelines. We organized the repo according to our data model rather than using the default structure suggested by dbt. It requires a little tweak in our dbt_project.yml config and complicates things when they add support for the Python model. But overall, it's well worth it for us to customize this structure as it makes navigating our code much more intuitive.
In the "Data Modeling Principles" section, we've briefly mentioned the different layers in our data structure. To put it in more detail:
(You might find this similar to Databricks' Medallion Architecture. It is just a happy coincidence. Similar ideas around the community probably inspired us at the time.) We still need to fully apply an established data modeling methodology (Entity-Relation, Kimball, Inmon, Vault, etc.). So instead, we curated ideas from all of them into a hybrid method that makes the most sense for us. This allowed us to be fast and flexible but still able to follow good practices.
Trigger and Orchestration
Even with dbt, we still want to centralize all pipeline execution under our existing Airflow service. Therefore, the dbt commands are packaged as an Airflow task to be run inside Kubernetes pods. We have written some light wrapper code to make creating dbt tasks as easy as possible:
task_a = dbtTaskConfig(node="dim_currency_exchange", type="model")
task_b = dbtTaskConfig(node="dim_countries", type="seed+test")
We prefer running only one model/seed per Airflow task as it makes for more convenient monitoring. However, we do have to manually set the task dependency (rather than using dbt’s automated lineage).
Incremental models
Due to the large volume of data we have to handle, most of our models are incremental. To aid engineers in correctly implementing incremental models, we have a written guideline containing several different strategies. Here, we will share our most commonly used approach:
Dynamic incremental using dt_last_updated
-- This is an example for a dbt model that combines data from 2 large tables - raw_1 and raw_2.
-- If you need 1, 3 or more raw tables, the logic remains the same.
-- Note the unique_key being d_date.
WITH
-- Specifying the period that contains the new updated data (new since last run)
dates_with_new_data__from_raw_1 AS
(SELECT
MIN(d_date) AS min_date
FROM
WHERE TRUE
)
,dates_with_new_data__from_raw_2 AS
(SELECT
MIN(d_date) AS min_date
FROM
WHERE TRUE
)
,dates_with_new_data AS
(SELECT
LEAST(r1.min_date, r2.min_date) AS min_date
FROM dates_with_new_data__from_raw_1 AS r1
CROSS JOIN dates_with_new_data__from_raw_2 AS r2)
-- Selecting the data and building the model
,raw_1 AS
(SELECT
d_date,
...,
dt_last_updated AS dt_source_data_last_updated
-- This field is needed so that at next run we know which data from the raw table has already been processed and thus can be skipped
FROM
WHERE d_date >= (SELECT min_date FROM dates_with_new_data))
,raw_2 AS
(SELECT
d_date,
...,
MAX(dt_last_updated) AS dt_source_data_last_updated
-- If you need to use GROUP BY to transform data, the appropriate function to apply to this field is MAX()
FROM
WHERE d_date >= (SELECT min_date FROM dates_with_new_data)
GROUP BY d_date, ...)
SELECT
d_date,
...,
LEAST(raw_1.dt_source_data_last_updated, raw_2.dt_source_data_last_updated) AS dt_source_data_last_updated
FROM raw_1
JOIN raw_2 USING(d_date, ...)
With this approach, every time the model runs, it will process all the newly-updated data rows from the source table (the rows that have a higher dt_last_updated than the recorded time from the last run).
This is currently the preferred option because of the following advantages:
Important notes when using this approach:
delete+insert incremental strategy (which is the default and only strategy available for Redshift at the moment).d_date as the unique_key. But there's nothing stopping you from using other fields if you know what you're doing.dt_last_updated, or if your model logic changes, the changes won't be included in incremental runs. A full_refresh will be required.
Testing
We care enormously about data quality and write useful tests on all our essential models. The tests are also triggered through Airflow and will send an alert to our Slack when it fails, depending on the severity and importance of the test. We also link top critical tests with particular actions, such as direct notification to business users or pausing some user-facing data products from refreshing and taking in the bad data.
Documentation
We write documentation for our models when we can, especially when there are intricacies that users must know before using this model. For now, the docs are hosted on dbtCloud with a free account.
All the changes above took significant effort. But after those efforts were paid, we ended up in a much better state than when we started. The biggest benefits we've seen include:
In conclusion, the Homa data team's refactoring and re-data-warehousing efforts have been successful in addressing the limitations in our previous codebase. By removing the storage limit on their data warehouse and moving to an ELT process, we have made transformations faster and more convenient for analysts. Formalizing a list of data modeling rules and implementing a mono-repo structure has also helped to keep our data warehouse organized and easy to use. Additionally, providing an easy setup process for new team members has improved onboarding and productivity. Overall, the changes made have allowed the team to adapt to an increase in data volume and stakeholders' expectations while minimizing effort and maximizing efficiency.
By sharing this, we aim to give you more valuable ideas you can apply to your work.
If you want to learn more about open positions at Homa, consult this page.