airflow - 如何在 Airflow 中传递不记名 token

Question

我有一份有 3 个任务的工作
1) 使用 POST 请求获取 token
2)获取 token 值并存储在变量中
3) 使用步骤 2 中的 token 发出 GET 请求并传递不记名 token

问题是第 3 步不起作用，我收到 HTTP 错误。我能够在步骤 2 中打印 token 的值并在代码中进行验证

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}
token ="mytoken" //defined with some value which will be updated later

get_token = SimpleHttpOperator(
        task_id='get_token',
        method='POST',
        headers={"Authorization": "Basic xxxxxxxxxxxxxxx=="},
        endpoint='/token?username=user&password=pass&grant_type=password',
        http_conn_id = 'test_http',
        trigger_rule="all_done",
        xcom_push=True,
        dag=dag
    )

def pull_function(**context):
    value = context['task_instance'].xcom_pull(task_ids='get_token')
    print("printing token")
    print value
    wjdata = json.loads(value)
    print(wjdata['access_token'])
    token=wjdata['access_token']
    print token


run_this = PythonOperator(
    task_id='print_the_context',
    provide_context=True,
    python_callable=pull_function,
    dag=dag,
)

get_config = SimpleHttpOperator(
        task_id='get_config',
        method='GET',
        headers={"Authorization": "Bearer " + token},
        endpoint='someendpoint',
        http_conn_id = 'test_conn',
        trigger_rule="all_done",
        xcom_push=True,
        dag=dag
    )

get_token >> run_this >> get_config

Answer 1

您的存储方式 token因为“全局”变量不起作用。 Dag 定义文件(您在其中定义任务的脚本)与执行每个任务的运行时上下文不同。每个任务都可以在单独的线程、进程甚至另一台机器上运行，具体取决于执行程序。您在任务之间传递数据的方式不是通过全局变量，而是使用 XCom - 您已经部分做到了。
请尝试以下操作:
- 远程全局token多变的
- 在 pull_function而不是 print token做 return token - 这将再次将值推送到 XCom，以便下一个任务可以访问它
- 在下一个任务中从 XCom 访问 token 。

最后一步有点棘手，因为您使用的是 SimpleHttpOperator ，并且只有模板化字段是 endpoint和 data ，但不是 headers .
例如，如果你想传入一些 data从先前任务的 XCom 中，您将执行以下操作:

get_config = SimpleHttpOperator(
        task_id='get_config',
        endpoint='someendpoint',
        http_conn_id = 'test_conn',
        dag=dag,
        data='{{ task_instance.xcom_pull(task_ids="print_the_context", key="some_key") }}'
    )

但是不幸的是，您不能对 header 执行相同的操作，因此您必须通过 PythonOperator“手动”执行此操作，或者您可以继承 SimpleHttpOperator并创建自己的，例如:

class HeaderTemplatedHttpOperator(SimpleHttpOperator):
    template_fields = ('endpoint', 'data', 'headers')  # added 'headers' headers

然后使用那个，例如:

get_config = HeaderTemplatedHttpOperator(
        task_id='get_config',
        endpoint='someendpoint',
        http_conn_id = 'test_conn',
        dag=dag,
        headers='{{ task_instance.xcom_pull(task_ids="print_the_context") }}'
    )

请记住，我没有对此进行测试，这只是为了解释这个概念。尝试使用该方法，您应该会到达那里。