airflow提供了很多Macros Variables，可以直接使用jinja模板语言调用宏变量。
templated_command = """
echo "dag_run:{{ dag_run }}"
echo "run_id:{{ run_id }}"
echo "execution_date:{{ execution_date }}"
echo "ts:{{ ts }}"
echo "ti:{{ ti }}"
sleep 3
"""
t1 = BashOperator(
task_id='xcom',
bash_command=templated_command,execution_date并不是task的真正执行时间，而是上一周期task的执行时间。
即可以理解为“actual_execution_date= execution_date +schedual_interval”，或者我们换一种说法，我们在airflow上看到一个任务是6am执行的，而且interval=4hours，那么execution_date的值是2am，而不是6am，所以获取某个task的真正执行时间，需要获取execution_date的下个执行周期时间，即使用dag.following_schedule(execution_date)
execution_date示例：10-10T08:37:30.962731+00:00
{{ execution_date }} | the execution_date, (datetime.datetime) |
数据表相关：
dag_run 表:
包含dag的运行历史记录, 该表也有两个id栏位, 一个是id, 一个是run_id, id栏位是该表的PK, run_id栏位是这次运行的一个名字(字符型), 同一个dag, 它的run_id 不能重复.
物理PK: 即id栏位
逻辑PK: dag_id + execution_date 组合
execution_date 栏位, 表示触发dag的准确时间，是datetime类型字段。
---------------
airflow系统表:
---------------
connection 表:
我们的Task往往需要通过jdbc/ftp/http/webhdfs方式访问其他资源, 一般地访问资源时候都需要一些签证, airflow允许我们将这些connection以及鉴证存放在connection表中. 可以现在WebUI的Admin->Connections管理这些连接, 在代码中使用这些连接.
需要说明的是, connection表有2个id栏位, 一个是id, 一个是conn_id, id栏位是该表的PK, conn_id栏位是connection的名义id, 也就是说我们可以定义多个同名的conn_id, 当使用使用时airflow将会从同名的conn_id的列表中随机选一个, 有点基本的load balance的意思.
user 表 :
包含user的username和email信息. 我们可以在WebUI的Admin->Users管理.
variable 表 :
包含定义variable
xcom 表:
包含xcom的数据
dag 表:
包含dag的定义信息, dag_id是PK(字符型)
dag_run 表:
包含dag的运行历史记录, 该表也有两个id栏位, 一个是id, 一个是run_id, id栏位是该表的PK, run_id栏位是这次运行的一个名字(字符型), 同一个dag, 它的run_id 不能重复.
物理PK: 即id栏位
逻辑PK: dag_id + execution_date 组合
execution_date 栏位, 表示触发dag的准确时间
注意: 没有 task 表:
airflow的task定义在python源码中, 不在DB中存放注册信息.
task_instance 表:
物理PK: 该表没有物理PK
逻辑PK: dag_id + task_id + execution_date 组合.
execution_date 栏位, 表示触发dag的准确时间,是datetime类型字段
start_date/end_date 栏位,表示执行task的起始/终止时间, 也是datetime类型字段
job 表:
包含job(这里可以理解为批次)的运行状态信息