Airflow: Set custom run_id for TriggerDagRunOperator

Question

When using TriggerDagRunOperator to trigger another DAG, it just gives a generic name like trig_timestamp:

Is it possible to give this run id a meaningful name so I can easily identify different dag runs?

Answer 1

You can't immediately do this with the TriggerDagOperator as the "run_id" is generated inside it's execute method. However, you could implement your own operator, CustomTriggerDagOperator that would behave the way you want/need. For example:

class CustomTriggerDagOperator(TriggerDagOperator):
    def execute(self, context):
        if self.execution_date is not None:
            run_id = 'trig__{}'.format(self.execution_date)
            self.execution_date = timezone.parse(self.execution_date)
        else:
            run_id = 'trig__' + timezone.utcnow().isoformat()

        run_id += f'{self.trigger_dag_id}'

        dro = DagRunOrder(run_id=run_id)
        if self.python_callable is not None:
            dro = self.python_callable(context, dro)
        if dro:
            trigger_dag(dag_id=self.trigger_dag_id,
                        run_id=dro.run_id,
                        conf=json.dumps(dro.payload),
                        execution_date=self.execution_date,
                        replace_microseconds=False)
        else:
            self.log.info("Criteria not met, moving on")

This example above just appends the id of the triggered dag. You could use this same strategy to set the run_id arbitrarily.