警告
本文最后更新于 2023-05-17，文中内容可能已过时。
airflow-in-docker
千万不能用 previlleged ，这个会导致无法 mount /var/run/docker.socket
1
2
3
4
5
6
7
8
9
|
## root 执行
## 不能用 previlleged，·普通用户也可以
docker run -dit \
-v /var/run/docker.sock:/var/run/docker.sock \
-v /usr/bin/docker:/bin/docker \
--name=airflow \
--pull=always \
-p 18080:8080 \
10.32.111.107:5000/airflow:v1.0
|
如果是 dockerfile （本质上是普通用户，所以不用 root）不用完 previlleged
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
version: "3"
services:
airflow:
container_name: airflow
image: 10.32.111.107:5000/airflow:v1.0
## 不能用 previlleged，·普通用户也可以, 为了 /var/run/docker.socket
privileged: false
tty: true
volumes:
- ~:/mnt
- /var/run/docker.sock:/var/run/docker.sock ## docker-in-docker
- /usr/bin/docker:/bin/docker ## docker 可执行，如果有 so 也需要 mount
- /mnt/cephfs:/root/cephfs ## 可以在 docker-in-docker 使用 cephs
- ~/git/jobs/dags:/app/dags ## for dags
# - ~/mysql:/var/lib/mysql ## for mysql
ports:
- "18080:8080" ## web
command:
- /bin/bash
- -c
- |
/usr/sbin/init
pull_policy: always
restart: always
|
DAG 使用 DockerOperator
注意，在 airflow 处于 docker-in-docker 的情况下，这时候我们需要特殊处理 DockerOperator
docker_url: "unix://var/run/docker.sock"network_mode: host，应该为 outside 的 host 网络mounts: 这个最重要，决定了 docker-in-docker 能够有读取相关路径的权限，需要注意的是，这时候我们需要把 dockeroperator 理解成在 host 机器运行，所以的路径对应的是 host 的路径
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
from docker.types import Mount
qry_ctp = DockerOperator(
task_id = 'docker_qry_ctp',
docker_url = "unix://var/run/docker.sock",
image = '10.32.111.107:5000/tradingops/myctp.sx1:stable',
container_name = 'task___qry_ctp',
api_version = 'auto',
auto_remove = 'success', ## 'success', 'force', 'never'
network_mode = "host",
command = "",
mounts = [
Mount(
source='/mnt/cephfs', ## 这个是 host 的路径
target='/root/cephfs', ## 这个是 Docker-in-Docker 的路径
type='bind'
),
],
mount_tmp_dir = False,
)
|
当然，如果有 commnad，我们可以使用
- 单个命令，直接 comnand
- 多个命令，使用
/bin/bash -c '<command1> <command2>'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
rsync_tora = """
rsync --progress -avPzr --exclude='*'.log \
Colo114:/home/ops/tora.qry/log'*' \
/root/cephfs/ops/data/PublicInfo/Daily/PreTrading/InfoFromTORA
"""
rysnc_xele_sh = """
rsync --progress -avPzr --exclude='*'.log \
Colo109:/home/ops/xeleq.qry/log'*' \
/root/cephfs/ops/data/PublicInfo/Daily/PreTrading/InfoFromXELEQ/SH
"""
rysnc_xele_sz = """
rsync --progress -avPzr --exclude='*'.log \
Colo110:/home/ops/xeleq.qry/log'*' \
/root/cephfs/ops/data/PublicInfo/Daily/PreTrading/InfoFromXELEQ/SZ
"""
rsync_info = DockerOperator(
task_id = 'docker_rsync_info',
docker_url = "unix://var/run/docker.sock",
image = '10.32.111.107:5000/r7:v1.0',
container_name = 'task_rsync_info',
api_version = 'auto',
auto_remove = 'success', ## 'success', 'force', 'never'
network_mode = "host",
mount_tmp_dir = False,
mounts = [
Mount(
source='/mnt/cephfs', ## 这个是 host 的路径
target='/root/cephfs', ## 这个是 Docker-in-Docker 的路径
type='bind'
),
],
command = f"""
/bin/bash -c
'
{rsync_tora}
{rysnc_xele_sh}
{rysnc_xele_sz}
'
"""
)
|
DAG 使用 BashOperator
如果使用 BashOperator，这相对简单，直接调用 docker run 即可
1
2
3
4
5
6
7
8
|
qry_ctp = BashOperator(
task_id = 'docker_qry_ctp',
bash_command = """
docker run --pull=always --privileged=true \
--name=myctp.sx1 --rm=true \
10.32.111.107:5000/tradingops/myctp.sx1:stable
"""
)
|
william 收录于 Tools
支付宝
微信