1

How do I set a variable for use during a particular dag_run. I'm aware of setting values in xcom, but not all the operators that I use has xcom support. I also would not like to store the value into the Variables datastore, in case another dag run begins while the current one is running, that need to store different values.

Joyce
  • 1,431
  • 2
  • 18
  • 33
  • Your question is **unclear**; please **elaborate your question** with better explanation of *use-case* / *problem-statement* – y2k-shubham Feb 16 '19 at 06:40

1 Answers1

1

The question is not clear, but from whatever I can infer, I'll try to clear your doubts

not all the operators that I use has xcom support

Apparently you've mistaken xcom with some other construct because xcom feature is part of TaskInstance and the functions xcom_push() and xcom_pull() are defined in BaseOperator itself (which is the parent of all Airflow operators)

I also would not like to store the value into the Variables datastore, in case another dag run begins while the current one is running, that need to store different values.

It is straightforward (and no-brainer) to separate-out Variables on per-DAG basis (see point (6)); but yes for different DagRuns of a single DAG, this kind of isolation would be a challenge. I can think of xcom to be the easiest workaround for this. Have a look at this for some insights on usage of Xcoms.

Additionally, if you want to manipulate Variables (or any other Airflow model) at runtime (though I would recommend you to avoid it particularly for Variables), Airflow also gives completely liberty to exploit the underlying SQLAlchemy ORM framework for that. You can take inspiration from this.

y2k-shubham
  • 10,183
  • 11
  • 55
  • 131
  • Great answer. So currently, I've figured out to use Xcom, also within a file which needs string substitution thru macros, and grabbing values from xcom from the `ti` variable. So I'm interested in being able to make use of the xcom variable values during the definition of the task itself, where it instantiates an obj of the operator, where within one of the arguments, I'd like to access the xcom variables to calculate something. – Joyce Feb 17 '19 at 01:46
  • **@Joyce** For that you have 3 options **[1]** Use [`context["task_instance"]`](https://airflow.apache.org/concepts.html#xcoms) **[2]** Use [`BaseOperator.pull_xcom(..)`](https://github.com/apache/airflow/blob/v1-10-stable/airflow/models.py#L3173) **[3]** Directly exploit `SQLAlchemy` ORM like [this](https://github.com/apache/airflow/blob/v1-10-stable/airflow/bin/cli.py#L1166). Also see [this](https://stackoverflow.com/a/46061162/3679900) – y2k-shubham Feb 17 '19 at 02:03
  • **@Joyce**, correction here: for your query `..being able to make use of the xcom variable values during the definition of the task itself, where it instantiates an obj of the operator..`, this is only possible via method **[3]** above: using `SQLAlchemy` ORM – y2k-shubham Feb 18 '19 at 11:18