任務 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 Daemon 的主機 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 橋接網路
"none"：此容器不使用網路
"container:<name>" 或 "container:<id>"：使用另一個容器的網路堆疊，透過 <name> 或 <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 Daemon 通訊時要使用的 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

推送所有 stdout 或僅最後一行。預設為 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}