Make OpenLineage facets more compatible with spec

[dolfinus]

OpenLineage v1.9.0 integration for Airflow produces following facets:
DAG run:

job:
  facets:
    airflow:
      # AiflowJobFacet
      tasks:
      taskGroups:
      taskTree:
run:
  facets:
    airflowState:
      # AirflowStateRunFacet
      dagRunState:
      tasksState:

Task run:

job:
  facets: {}
run:
  airflow:
    # AirflowRunFacet
    dag:  # dag_id, description, owner, schedule, start_date and so on
    dagRun: # run_id, run_type and so on
    task:  # task_id, queue, trigger_rule, operator properties and so on
    taskInstance:  # duration, try_number, log_url
    taskUuid:

There are few drawbacks with this approach:

• DAG and DAG run information is present only in task run events, but not in DAG run events OpenLineage: Add AirflowRunFacet to DAG run events #40798.
• DAG and Task information like dag_id, task_id, description, owner is present only as run facet, not job facet, although this information is not related to a particular DAG/Task run, and describes the DAG/Task itself.

While implementing #40854, I've also got issues with "airflow" key of all these facets. For example, adding a DAG run facet leads to sending events like:

job:
  facets:
    airflow:
      # AiflowJobFacet
      tasks:
      taskGroups:
      taskTree:
run:
  facets:
    airflow:
      dag:
      dagRun:
      # task, taskInstance, taskUuid fields are missing

So OpenLineage events send by Airflow for DAG run and Task run have the facet with the same key "airflow", which has quite different content. This is not very comfortable to handle on OpenLineage backend side.

My proposal is:

• Make facet schema names more descriptive:
• • AirflowRunFacet -> AirflowTaskRunFacet
• • AirflowJobFacet -> AirflowDagInfoJobFacet
• • AirflowStateRunFacet -> AirflowDagStateRunFacet
• Use specific key for each facet type, instead of generic "airflow" which may lead to conflicts.
• Move dag and task fields from run facet to job facet.
• Drop taskUuid field as it just copies runId, and has no meaning for Airflow

This how the result can look like:

Option 1 - nested facets

DAG run:

job:
  facets:
    airflow_dagInfo:
      # AirflowDagInfoJobFacet
      dag:  # dag_id, description, owner, schedule, start_date and so on
      tasks:  # list of task names and UI options
      taskGroups:  # list of task groups
      taskTree:  # tree of tasks and groups
run:
  facets:
    airflow_dagRun:
      # AirflowDagRunFacet
      dagRun: # run_id, run_type and so on
    airflow_dagState:
      # AirflowDagStateRunFacet
      dagRunState:
      tasksState:

Task run:

job:
  facets:
    airflow_taskInfo:
      # AirflowTaskInfoJobFacet
      task:  # task_id, queue, trigger_rule, operator properties and so on
run:
  airflow_taskRun:
    # AirflowTaskRunFacet
    taskInstance:  # duration, try_number, log_url

Option 2 - flat facets

DAG run:

job:
  facets:
    airflow_dagInfo: # dag_id, description, owner, schedule, start_date and so on
    airflow_tasksDescription: # list of task names and UI options
    airflow_taskGroups:  # list of task groups
    airflow_taskTree:  # tree of tasks and groups
run:
  facets:
    airflow_dagRun: # run_id, run_type and so on
    airflow_dagState:
    airflow_taskStates:

Task run:

job:
  facets:
    airflow_taskInfo: # task_id, queue, trigger_rule, operator properties and so on
run:
  airflow_taskRun: # duration, try_number, log_url

The concern is:

• Is that something that should be done? That could be only my point of view, but other users may be completely fine with current approach.
• What about backward compatibility?

[mobuchowski]

Is that something that should be done? That could be only my point of view, but others are completely fine with current approach.
What about backward compatibility?

From my experience I would love everything to be consistent with some norms we choose.

However, especially "dropping" migration is very hard for existing event consumers - especially data observability tools providers that consume events generated by their clients Airflow instances - for example Atlan, Astronomer.

However, we'll have good opportunity in next 6-12 months to do more breaking changes with Airflow 3 - that would necessary require somehow large changes to the structure of provider anyway. It would probably necessitate "major" release of the provider.

I would recommend:

• adding new stuff consistent with the norms
• keeping this doc and making some bigger changes in the future

[mobuchowski]

But, I think exception should be made for AirflowJobFacet - I think attaching it at the the same place as AirflowRunFacet is an exceptionally poor choice.