Jump to content

Fundraising/techops/docs/analytics stack/pipelines

From Wikitech

Product & Tech data import pipelines

Overview

FR Tech imports data from Product & Tech (P&T)'s data lake (HDFS) into FR Tech's data lake (MinIO). P&T's Data Platform SRE team hosts an airflow instance for FR tech that orchestrates the pipelines. These pipelines' DAGs all follow a similar format: wait for data to be available, run a query to aggregate the data, transfer the aggregated data to MinIO, register the new partitions in Trino. The first three steps happen in the airflow DAG. Dagster runs a daily scheduled `sync_new_trino_partitions` job for the final step.

All imported data gets sent to the hive.product_and_tech_import schema in Trino, and the product-and-tech-import bucket in MinIO.

DAG Details

The Airflow DAGs live in the Airflow DAGs repo. The Airflow DAGs are parameterized and generated automatically from a DAG factory. Each pipeline DAG instantiates this factory with the necessary parameters E.g. the pageviews_hourly DAG sets all the parameters needed for the pipeline to run. Using the DAG factory makes it easy to create new pipelines and modify existing ones. As mentioned above, each DAG consists of 3 tasks:

  1. A sensor that checks for the presence of new data in the source table
  2. A compute task that uses Spark to parameterize and run a query defined in an .hql file and places the output in HDFS. These hql files are actually pulled in from the refinery repo, but a copy is kept in the Airflow DAGs repo for convenience. This means that changes to the hql file in the Airflow DAGs repo won't affect the pipeline. To update the query, it must first be merged into the refinery repo, and then deployed by the Data Engineering team.
  3. A transfer task that copies the newly aggregated data files from HDFS to a temp file, and then from the temp file to MinIO. The table directories are Hive partitioned, and Airflow automatically places each file in the appropriate partition subdirectory in HDFS and MinIO.



If you want to see for example what Trino table the data gets written to (the source table, the partition scheme, or the hql query), check the HiveToMinioConfig in each DAG's .py file.

Here is a diagram of the generic data flow:

Troubleshooting

The most recent data is missing!

First, make sure you understand what data should be present in Trino/MinIO/Metabase. Usually, data for a given day gets loaded the following morning, and the Trino sync_partition_metadata job runs each day at 9:10 AM UTC. So for example, if at 2026-08-02 9:30 AM UTC, data for 2026-08-01 isn't showing up in trino, something has gone wrong.

Start by checking for the data file in MinIO. For the above date, you should expect to see data (a parquet file) in product-and-tech-import/banner_activity_minutely/year=2026/month=08/day=1/hour=1. If the data does exist, then there may have been a delay, and you should run the 'sync_new_trino_partitions' job in Dagster. If the data does not exist, check for an email from Airflow, or check on the DAG directly in https://airflow-fr-tech.wikimedia.org/home. Follow these instructions to rerun the failed DAG task(s).

If the DAG fails after rerunning, reach out to #talk-to-data-engineering in Slack.

Just show me the configs!

Click on one of the DAGs in the repo! To see the unparameterized query that gets run, check the hql directory in that same repo. To check the parameterized query, check the Airflow logs of the relevant run's 'compute' task.

Retrieved from "https://wikitech.wikimedia.org/w/index.php?title=Fundraising/techops/docs/analytics_stack/pipelines&oldid=2449811"