设置气流中的环境变量

Question

我正在为我的任务创建一个气流DAG。这个作业在s3桶中创建具有当前日期的json文件。我计划将日期作为环境变量传递。这可能吗?我正在尝试的是下面

recommendations = KubernetesPodOperator(
namespace='data',
image = "quay.io/comp/data-rp-recommender:0.1.1",
name = "data-rp-recommender-run",
task_id = "data-rp-recommender-task",
dag = dag,
get_logs = True,
in_cluster = True,
is_delete_operator_pod = False,
volumes = [
Volume(
name = 'etcctek',
)
],
volume_mounts = [
VolumeMount(
name = 'etcctek',
mount_path = '/opt/docker/etc',
sub_path = None,
read_only = True
)
],
env_vars = {
'DATE' :
'USER_PRECEDING_STATS_S3' : 's3a://ml/rp-recommendations/userPreceedingStats/int/DATE/json'
'NUGGET_CATEGORICALS_S3' : 's3a://ml/rp-recommendations/nuggetCategorials/int/DATE/json'
'PREREQUISITE_STATS_S3' : 's3a://ml/rp-recommendations/prerequisiteStats/int/DATE/json'
'USER_CURRENT_STATS_S3' : 's3a://ml/rp-recommendations/userCurrentStats/int/DATE/json'
},

resources = {
'request_memory': '2048M',
'request_cpu': '1.5'
}
)

在上面的例子中，我想设置env_vars为DATE作为当前日期，然后在下面的其他变量中使用该当前日期。这是我第一次在气流中，需要一些帮助

Answer 1

假设要写任务执行开始时间:

var = '{{ ti.start_date }}'

关于任务实例模型的更多细节在这里