Migrate TAAR jobs from S3 to GCS

Telemetry-Airflow

Apache Airflow is a platform to programmatically author, schedule and monitor workflows.

This repository codifies the Airflow cluster that is deployed at workflow.telemetry.mozilla.org (behind SSO) and commonly referred to as "WTMO" or simply "Airflow".

Some links relevant to users and developers of WTMO:

The dags directory in this repository contains some custom DAG definitions
Many of the DAGs registered with WTMO don't live in this repository, but are instead generated from ETL task definitions in bigquery-etl
The Data SRE team maintains a WTMO Developer Guide (behind SSO)

Writing DAGs

See the Airflow's Best Practices guide to help you write DAGs.

⚠ Warning: Do not import resources from the `dags` directory in DAGs definition files ⚠

As an example, if you have dags/dag_a.py and dags/dag_b.py and want to use a helper function in both DAG definition files, define the helper function in the utils directory such as:

`utils/helper.py`

def helper_function():
    return "Help"

`dags/dag_a.py`

from airflow import DAG

from utils.helper import helper_function

with DAG("dag_a", ...):
    ...

`dags/dag_b.py`

from airflow import DAG

from utils.helper import helper_function

with DAG("dag_b", ...):
    ...

WTMO deployments use git-sync sidecars to synchronize DAG files from multiple repositories via telemetry-airflow-dags using git submodules. Git-sync sidecar pattern results in the following directory structure once deployed.

airflow
├─ dags
│  └── repo
│      └── telemetry-airflow-dags
│          ├── <submodule repo_a>
│          │    └── dags
│          │        └── <dag files>
│          ├── <submodule repo_b>
│          │    └── dags
│          │        └── <dag files>
│          └── <submodule repo_c>
│               └── dags
│                   └── <dag files>
├─ utils
│  └── ...
└─ plugins
   └── ...

Hence, defining helper_function() in dags/dag_a.py and importing the function in dags/dag_b.py as from dags.dag_a import helper_function will not work after deployment because of the directory structured required for git-sync sidecars.

Prerequisites

This app is built and deployed with docker and docker-compose. Dependencies are managed with pip-tools pip-compile.

You'll also need to install PostgreSQL to build the database container.

Installing dependencies locally

⚠ Make sure you use the right Python version. Refer to Dockerfile for current supported Python Version ⚠

You can install the project dependencies locally to run tests with Pytest. We use the official Airflow constraints file to simplify Airflow dependency management. Install dependencies locally using the following command:

make pip-install-local

Updating Python dependencies

Add new Python dependencies into requirements.in or requirements-dev.in then execute the following commands:

make pip-compile
make pip-install-local

Build Container

Build Airflow image with

make build

Local Deployment

To deploy the Airflow container on the docker engine, with its required dependencies, run:

make build
make up

macOS

Assuming you're using Docker for Docker Desktop for macOS, start the docker service, click the docker icon in the menu bar, click on preferences and change the available memory to 4GB.

Testing

Adding dummy credentials

Tasks often require credentials to access external credentials. For example, one may choose to store API keys in an Airflow connection or variable. These variables are sure to exist in production but are often not mirrored locally for logistical reasons. Providing a dummy variable is the preferred way to keep the local development environment up to date.

Update the resources/dev_variables.env and resources/dev_connections.env with appropriate strings to prevent broken workflows.

Usage

You can now connect to your local Airflow web console at http://localhost:8080/.

All DAGs are paused by default for local instances and our staging instance of Airflow. In order to submit a DAG via the UI, you'll need to toggle the DAG from "Off" to "On". You'll likely want to toggle the DAG back to "Off" as soon as your desired task starts running.

Testing GKE Jobs (including BigQuery-etl changes)

See https://go.corp.mozilla.com/wtmodev for more details.

make build && make up
make gke

When done:
make clean-gke

From there, connect to Airflow and enable your job.

Testing Dataproc Jobs

Dataproc jobs run on a self-contained Dataproc cluster, created by Airflow.

To test these, jobs, you'll need a sandbox account and corresponding service account. For information on creating that, see "Testing GKE Jobs". Your service account will need Dataproc and GCS permissions (and BigQuery, if you're connecting to it). Note: Dataproc requires "Dataproc/Dataproc Worker" as well as Compute Admin permissions. You'll need to ensure that the Dataproc API is enabled in your sandbox project.

Ensure that your dataproc job has a configurable project to write to. Set the project in the DAG entry to be configured based on development environment; see the ltv.py job for an example of that.

From there, run the following:

make build && make up
./bin/add_gcp_creds $GOOGLE_APPLICATION_CREDENTIALS google_cloud_airflow_dataproc

You can then connect to Airflow locally. Enable your DAG and see that it runs correctly.

Production Setup

This repository was structured to be deployed using the offical Airflow Helm Chart.. See the Production Guide for best practices.

Debugging

Some useful docker tricks for development and debugging:

make clean

# Remove any leftover docker volumes:
docker volume rm $(docker volume ls -qf dangling=true)

# Purge docker volumes (helps with postgres container failing to start)
# Careful as this will purge all local volumes not used by at least one container.
docker volume prune

	env=mozetl_envvar("addon_aggregates", {
	"date": "{{ ds_nodash }}",
	"output-bucket": "{{ task.__class__.private_output_bucket }}"
	}),

	env=mozetl_envvar("clients_daily", {
	# Note that the output of this job will be earlier
	# than this date to account for submission latency.
	# See the clients_daily code in the python_mozetl
	# repo for more details.
	"date": "{{ ds }}",
	"output-bucket": "{{ task.__class__.private_output_bucket }}"
	}),

	def check_s3fs_success(bucket, prefix, num_partitions):
	"""Check the s3 filesystem for the existence of `_SUCCESS` files in dataset partitions.

	:prefix: Bucket prefix of the table
	:num_partitions: Number of expected partitions
	"""
	s3 = boto3.resource("s3")
	objects = s3.Bucket(bucket).objects.filter(Prefix=prefix)
	success = set([obj.key for obj in objects if "_SUCCESS" in obj.key])

	return len(success) == num_partitions

	env=tbv_envvar("com.mozilla.telemetry.views.ClientsDailyView", {
	"date": "{{ ds_nodash }}",
	"output-bucket": "{{ task.__class__.private_output_bucket }}"
	}),

	run_sql = gke_command(
	task_id="run_sql",
	cmds=["bash"],
	env_vars={"DATASET": "glam_etl", "SUBMISSION_DATE": "{{ ds }}"},
	command=["script/glam/run_fenix_sql"],
	docker_image="mozilla/bigquery-etl:latest",
	gcp_conn_id="google_cloud_glam_fenix_dev",
	dag=dag,
	)

	email=["[email protected]", "[email protected]"],
	env={"date": "{{ ds_nodash }}", "bucket": "{{ task.__class__.private_output_bucket }}"},
	uri="https://raw.githubusercontent.com/mozilla/firefox-test-tube/master/notebook/import.py",

	def register_status(operator, name, description, on_success=False):
	"""Wrap an operator with an external status page.

	The default behavior will only set the state of a dataset to partial_outage.

	env=mozetl_envvar("retention", {
	"start_date": "{{ ds_nodash }}",
	"bucket": "{{ task.__class__.private_output_bucket }}",
	"slack": 4
	}),

	email=["[email protected]", "[email protected]", "[email protected]"],
	env=mozetl_envvar("search_dashboard", {
	"submission_date": "{{ ds_nodash }}",
	"bucket": "{{ task.__class__.private_output_bucket }}",
	"prefix": "harter/searchdb",
	"save_mode": "overwrite"
	}),

	engagement_ratio = EMRSparkOperator(
	task_id="engagement_ratio",
	job_name="Update Engagement Ratio",
	execution_timeout=timedelta(hours=6),
	instance_count=10,
	uri="https://raw.githubusercontent.com/mozilla/telemetry-airflow/master/jobs/engagement_ratio.sh",
	output_visibility="public",
	dag=dag)

	"masterAuthorizedNetworksConfig":
	{
	"enabled": True,
	},

mozilla / telemetry-airflow Goto Github PK

telemetry-airflow's Introduction

Telemetry-Airflow

Writing DAGs

⚠ Warning: Do not import resources from the dags directory in DAGs definition files ⚠

utils/helper.py

dags/dag_a.py

dags/dag_b.py

Prerequisites

Installing dependencies locally

Updating Python dependencies

Build Container

Local Deployment

macOS

Testing

Adding dummy credentials

Usage

Testing GKE Jobs (including BigQuery-etl changes)

Testing Dataproc Jobs

Production Setup

Debugging

telemetry-airflow's People

Contributors

Stargazers

Watchers

Forkers

telemetry-airflow's Issues

Recommend Projects

Recommend Topics

Recommend Org

⚠ Warning: Do not import resources from the `dags` directory in DAGs definition files ⚠

`utils/helper.py`

`dags/dag_a.py`

`dags/dag_b.py`