SPIKE: [2d] Explore celery as POC airflow executor

[rshewitt]

Purpose

We want to understand celery as an airflow executor, but we're not sure how to do that.

Given above uncertainty, conducting research is needed to provide factual knowledge on future steps.

2 days of effort has been allocated and once complete, findings will be demonstrated and specific future actions will be decided.

Acceptance Criteria

[ACs should be clearly demo-able/verifiable whenever possible. Try specifying them using BDD.]

• GIVEN a local instance of airflow
  AND a collection of datasets to test on
  AND celery configured as the airflow executor
  WHEN 2 days expires
  THEN demonstrate findings of using celery as an airflow executor
  AND provide potential questions for team members to ask at airflow summit.

Background

[Any helpful contextual notes or links to artifacts/evidence, if needed]

Sketch

• launch local airflow instance with celery configured
• run datasets through example ETL dag
• identify additional datasets to expose strong suits/shortcomings of celery as executor (ie large DCAT-US source)
• record findings ( e.g. odd behaviors, failures, inconsistencies, successes )
• compile list of questions for team members.

[rshewitt]

Not related to celery but the default maximum list/dict length xcom can push to trigger task mapping is 1024 via AIRFLOW__CORE__MAX_MAP_LENGTH ( source )

[rshewitt]

are we going to find ourselves in a position where state changes from when a task is requested to when it's executed?

[rshewitt]

celery workers prefetch tasks by default ( see worker_prefetch_multiplier ). productivity could be impacted if a task has been prefetched by a worker and not processed. the default visibility_timeout is an hour. decreasing that could pop the task from the busy worker and assign it to an available one?

[rshewitt]

if the time to execute a task > visibility timeout the task could be perpetually re-assigned and never completed.

[rshewitt]

celery tasks don't have time limits by default. this can be configured using task_soft_time_limit ( throws exception ) and/or task_time_limit ( kill then replace worker ).

[rshewitt]

supported execution pools:

• prefork ( default ) (CPU)
  • children are process-based.
• solo (CPU)
  • the worker does the work instead of a pool of children. less oversight = faster processing?
• eventlet (I/O)
  • children are greenlet-based "cooperatively scheduled threads".
• gevent (I/O)
  • children are greenlet-based.

I/O tasks:

• extract requires downloading something.
• transform could require using an mdtranslator service.
• load requires writing to an s3 bucket.

CPU tasks:

• validation?