数据流水线的中间结果往哪放是个容易被忽略、出事才知道疼的问题。直接写本地盘任务被调度到别的节点就找不到文件挂共享 NFS又多一层要维护的高可用。Airflow 自带的 Amazon S3 provider 其实就是个标准 S3 客户端只要把连接的endpoint_url从 AWS 指到 RustFSDAG 里的读写就全落到对象存储上了。这篇顺着官方集成指南用 Apache Airflow 跑通一个写对象、等键、读回来的最小闭环。更重要的是对象存储作为 DAG 各任务之间的交换区是天然无状态的任务失败重跑时新节点直接去桶里拿上一阶段的产物不用关心上次跑在哪台机器这恰恰解决了开头说的「文件散落各节点」这个痛点。Airflow 的 S3 交互都经过 provider 客户端改个 endpoint 就指向 RustFSAirflow 里凡是用 S3 的 operator、sensor、hook底层都走apache-airflow-providers-amazon这个 provider 的 S3 客户端。它和直接用 boto3 没什么本质区别区别只在于凭证和端点是从 Airflow 的「连接connection」对象里读的。所以接入 RustFS 不需要改任何 DAG 代码只要在连接里把endpoint_url填成 RustFS 的地址provider 就会把本来发往s3.amazonaws.com的请求发到你的存储上。官方验证用的镜像是apache/airflow:3.3.2standalone 模式SequentialExecutor镜像里已经预装了全部 provider包括apache-airflow-providers-amazon不用额外装包。S3 客户端同样是 path-style 寻址非 AWS 端点靠endpoint_url这一个字段就能重定向。起 standalone 实例先把演示桶建好先以关闭示例 DAG 的方式起一个 standalone 实例顺手把演示桶建出来。关示例是为了不让一堆 demo 把界面和日志弄乱dockerrun-d--nameairflow--networkoo-rustfs_default-p8080:8080\-eAIRFLOW__CORE__LOAD_EXAMPLESFalse\-v$PWD/dags:/opt/airflow/dags\apache/airflow:3.3.2 standalone rc mb rustfs/airflow-demo桶用 RustFS 的rc mb建名字和后面 DAG 里写的s3_bucket保持一致。RustFS 当前稳定版是 1.0.42026-10-25自 1.0.0 GA2026-09-16起这套 S3 接口即生产可用接 Airflow 不需要任何特殊开关。注册一个自定义 S3 连接把 boto3 从 AWS 拐到 RustFSS3 provider 的端点和凭证都从连接的extra字段里读。把下面这段里的占位符换成真实的 RustFS 地址和密钥注意aws_access_key_id和aws_secret_access_key也写在extra里不是连接的主机名字段dockerexecairflow airflow connectionsaddrustfs\--conn-type aws\--conn-extra{endpoint_url: http://your-rustfs-endpoint:9000, region_name: us-east-1, aws_access_key_id: your-access-key, aws_secret_access_key: your-secret-key}endpoint_url这一项就是把 boto3 客户端从 AWS 重定向到 RustFS 的全部操作。之后 DAG 里任何用aws_conn_idrustfs的 S3 操作都会走这个自定义端点。extra 里的region_name是 provider 要求的字段但 RustFS 并不按区域做数据分区填us-east-1之外的任意值也不会影响读写它存在的意义只是满足 boto3 客户端的参数校验。真正决定请求打到哪里去的只有endpoint_url和那对访问密钥把这两项填对连接就通了。写个 DAG写对象、等键出现、再读回来下面这个 DAG 用 operator 写一个对象用 sensor 等这个键出现再用 hook 读回来正好覆盖了「写—等—读」三种典型交互。scheduleNone表示手动触发避免它自己跑起来importdatetimefromairflow.providers.amazon.aws.hooks.s3importS3Hookfromairflow.providers.amazon.aws.operators.s3importS3CreateObjectOperatorfromairflow.providers.amazon.aws.sensors.s3importS3KeySensorfromairflow.sdkimportdag,taskdag(scheduleNone,start_datedatetime.datetime(2026,1,1),catchupFalse,tags[rustfs],)defrustfs_demo():createS3CreateObjectOperator(task_idwrite_object,s3_bucketairflow-demo,s3_keydags/airflow-put.txt,datawritten by airflow to rustfs,aws_conn_idrustfs,replaceTrue,)waitS3KeySensor(task_idwait_for_object,bucket_keydags/airflow-put.txt,bucket_nameairflow-demo,aws_conn_idrustfs,timeout120,poke_interval10,modereschedule,)taskdefread_object():hookS3Hook(aws_conn_idrustfs)bodyhook.read_key(keydags/airflow-put.txt,bucket_nameairflow-demo)print(read back:,body)assertbodywritten by airflow to rustfscreate[wait,read_object()]rustfs_demo()新 DAG 默认是暂停状态触发前要先取消暂停否则 run 会一直卡在排队dockerexecairflow airflow dags unpause rustfs_demodockerexecairflow airflow dags trigger rustfs_demodockerexecairflow airflow dags list-runs rustfs_demo|head-3跑完去 RustFS 桶里确认对象真的落下去了rclsrustfs/airflow-demo/-rrccatrustfs/airflow-demo/dags/airflow-put.txt演示之外的真实 ETL还能加两步上面这个 DAG 只是为了证明读写链路通真要做数据流水线provider 里现成的传输 operator 比手写S3Hook更顺手。把本地文件推上 RustFS用LocalFilesystemToS3Operator把桶里的产物拉回本地做后续处理用S3ToLocalFilesystemOperator。它们都吃同一个aws_conn_idrustfs端点自然还是指向你的存储fromairflow.providers.amazon.aws.transfers.local_to_s3importLocalFilesystemToS3Operator pushLocalFilesystemToS3Operator(task_idpush_parquet,filename/data/result.parquet,dest_keyresults/result.parquet,dest_bucketairflow-demo,aws_conn_idrustfs,replaceTrue,)这么做的好处是中间产物不再落在 Airflow 工作节点的临时目录里。任务失败后重跑新节点直接去桶里拿上一阶段的输出不用关心上次跑在哪台机器上。对象存储在这里充当了 DAG 各任务之间稳定的交换区而不是某个节点的本地盘。这也是为什么把 Airflow 和对象存储接起来比在每台 worker 上挂盘或者维护一套 NFS 都更省心存储的扩缩容和耐久完全不用 Airflow 操心编排平台只管调度数据的事交给桶。不过要提醒一句边界这套接法依赖 provider 把 S3 能力完整实现个别高级操作未必覆盖落地前翻一遍 S3 兼容性说明 确认你要用的那几个 API 在列比上线后报错再查要省时间。把 demo 搬进生产还有两步要补。一是执行器standalone 加 SequentialExecutor 只能单任务串行真实流水线要用 Celery 或 Kubernetes 执行器多个任务才能真正并行二是隔离演示里直接用了 RustFS 的访问密钥生产应该按 访问密钥管理 给 Airflow 建一个只对这个桶有读写权限的专用凭证别让编排平台拿着全量权限的密钥跑。这两件做好对象存储作为交换区才既稳又能收敛权限。三个上手才会遇到的卡点官方排障段把几个典型坑列得很直接这里按出现频率排一下。第一触发后 DAG run 一直queued。原因是 Airflow 3 里新 DAG 默认暂停暂停状态下触发的 run 永远不会执行执行一次airflow dags unpause rustfs_demo排队的 run 会自己启动。第二导入报错cannot import name S3KeySensor from airflow.providers.amazon.aws.operators.s3。S3KeySensor在sensors模块S3CreateObjectOperator在operators模块从同一个模块导入两者会失败按上面代码分两个 import 就行。第三任务报连接错误。endpoint_url必须从 Airflow 容器内部可达要用 RustFS 的 Docker 网络主机名不能写localhost——容器里的localhost是 Airflow 自己不是存储。排查组件状态时Airflow 3 的健康检查端点在/api/v2/monitor/health。这套接法的边界也要说清楚官方的部署明确写了「用于本地集成测试不适用于生产环境」它用 standalone 单机和 SequentialExecutor生产要用 Celery 或 Kubernetes 执行器、给 Airflow 和 RustFS 各建专用凭证。另外 provider 里还有S3ToLocalFilesystemOperator、LocalFilesystemToS3Operator这类传输 operator落地真实 ETL 时按 Amazon provider 文档 取用即可启用更多 hook 前先翻一遍 S3 兼容性说明 确认接口覆盖。官方入口RustFS 与 Airflow 的集成指南在 docs.rustfs.com/zh/developer/integration/big-data/airflowS3 兼容性边界见 docs.rustfs.com/zh/administration/protocols/s3源码与版本在 github.com/rustfs/rustfs。
