任務 Docker 裝飾器¶

用 @task.docker 裝飾器封裝的 Python 可呼叫物件及其引數將在 docker 容器內執行。

引數¶

Docker 任務裝飾器支援以下引數。

multiple_outputs

如果設定，函式返回值將被展開為多個 XCom 值。字典將以鍵作為 XCom 鍵展開為 XCom 值。預設為 False。

use_dill

是否使用 dill 或 pickle 進行序列化

python_command

執行函式的 Python 命令，預設為 python3

image

用於建立容器的 Docker 映象。如果省略映象標籤，將使用“latest”。

api_version

遠端 API 版本。設定為 auto 可自動檢測伺服器版本。

container_name

容器名稱。可選（模板化）

cpus

分配給容器的 CPU 數量。此值將乘以 1024。

docker_url

執行 docker 守護程序的主機 URL。預設為 DOCKER_HOST 環境變數的值，如果未設定，則為 unix://var/run/docker.sock。

environment

在容器中設定的環境變數。（模板化）

private_environment

在容器中設定的私有環境變數。這些變數不是模板化的，並且在網站上隱藏。

env_file

到 .env 檔案的相對路徑，其中包含要在容器中設定的環境變數。會被 environment 引數中的變數覆蓋。

force_pull

每次執行時拉取 docker 映象。預設為 False。

mem_limit

容器可使用的最大記憶體量。可以是表示位元組限制的浮點值，或形如 128m 或 1g 的字串。

host_tmp_dir

指定主機上臨時目錄的位置，該目錄將對映到 tmp_dir。如果未提供，預設為使用標準系統臨時目錄。

network_mode

容器的網路模式。可以是以下之一

"bridge": 使用預設 docker bridge 網路為容器建立新的網路堆疊
"none": 此容器沒有網路
"container:<name>" 或 "container:<id>": 使用透過或指定的另一個容器的網路堆疊
"host": 使用主機網路堆疊。與 port_bindings 不相容
"<network-name>" 或 "<network-id>": 將容器連線到使用者建立的網路（使用 docker network create 命令）

tls_ca_cert

PEM 編碼的證書頒發機構檔案的路徑，用於保護 docker 連線。

tls_client_cert

PEM 編碼的客戶端證書檔案的路徑，用於驗證 docker 客戶端。

tls_client_key

PEM 編碼的客戶端金鑰檔案的路徑，用於驗證 docker 客戶端。

tls_verify

設定為 True 以驗證所提供證書的有效性。

tls_hostname

要與 docker 伺服器證書匹配的主機名，或設定為 False 以停用檢查。

tls_ssl_version

與 docker 守護程序通訊時使用的 SSL 版本。

mount_tmp_dir

指定臨時目錄是否應從主機繫結掛載到容器。

tmp_dir

容器內部的掛載點，對應於操作員在主機上建立的臨時目錄。該路徑也可透過容器內的環境變數 AIRFLOW_TMP_DIR 獲取。

user

docker 容器內的預設使用者。

mounts

要掛載到容器內的掛載點列表，例如 ['/host/path:/container/path', '/host/path2:/container/path2:ro']。

working_dir

在容器上設定的工作目錄（相當於 docker 客戶端的 -w 開關）

entrypoint

覆蓋映象的預設 ENTRYPOINT

xcom_all

推送所有標準輸出還是僅推送最後一行。預設為 False（最後一行）。

docker_conn_id

要使用的 Airflow 連線 ID

dns

Docker 自定義 DNS 伺服器

dns_search

Docker 自定義 DNS 搜尋域

auto_remove

當容器程序退出時啟用容器的移除。可能的值

never: （預設）不移除容器
success: 成功時移除
force: 總是移除容器

shm_size

/dev/shm 的大小（位元組）。大小必須大於 0。如果省略，使用系統預設值。

tty

為容器分配偽 TTY。需要設定此項才能檢視 Docker 容器的日誌。

hostname

容器的可選主機名。

privileged

授予此容器擴充套件許可權。

cap_add

包含容器能力

extra_hosts

在容器內部解析的額外主機名，以主機名到 IP 地址的對映表示。

timeout

API 呼叫的預設超時時間，單位為秒。

device_requests

將主機資源（如 GPU）暴露給容器。

log_opts_max_size

日誌滾動前的最大大小。一個正整數加上表示度量單位（k、m 或 g）的修飾符。例如：10m 或 1g。預設為 -1（無限制）。

log_opts_max_file

最多可以存在的日誌檔案數量。如果滾動日誌建立了多餘的檔案，將刪除最舊的檔案。僅在設定了 max-size 時有效。一個正整數。預設為 1。

ipc_mode

設定容器的 IPC 模式。

skip_on_exit_code

如果任務以此退出碼退出，將任務保持在 skipped 狀態（預設：None）。如果設定為 None，任何非零退出碼都將被視為失敗。

port_bindings

將容器的埠釋出到主機。這是一個字典值，其中鍵表示要在容器內開啟的埠，值表示繫結到容器埠的主機埠。與 network_mode 中的 "host" 不相容。

ulimits

要為容器設定的 ulimit 選項列表。每個項應為一個 docker.types.Ulimit 例項。

使用示例¶

tests/system/docker/example_taskflow_api_docker_virtualenv.py

    @task.docker(image="python:3.9-slim-bookworm", multiple_outputs=True)
    def transform(order_data_dict: dict):
        """
        #### Transform task
        A simple Transform task which takes in the collection of order data and
        computes the total order value.
        """
        total_order_value = 0

        for value in order_data_dict.values():
            total_order_value += value

        return {"total_order_value": total_order_value}