如何在Airflow中使用Mongoexport导出MongoDB集合数据？

你当前代码里的mongo_col.export()无法生效，因为pymongo的Collection对象并没有这个方法。要实现和命令行mongoexport --uri="URI" --collection=mongo_col --type json --out=mongo_col.json相同的导出效果，有两种实用方案：

方案1：用BashOperator直接调用mongoexport命令 #

这是最直接的方式，因为mongoexport本身就是命令行工具，Airflow的BashOperator可以直接执行它。需要确保Airflow Worker节点已安装MongoDB工具包（包含mongoexport），且能访问目标MongoDB实例。

代码示例： #

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime
import os

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    'mongodbexport',
    default_args=default_args,
    description='Export data from MongoDB using mongoexport',
    schedule_interval="0 0 * * *",
    catchup=False,
    tags=['export','mongodb'],
) as dag:
    # 从环境变量拼接MongoDB连接URI
    mongo_uri = (
        f"mongodb://{os.environ.get('MUSER_NAME')}:{os.environ.get('MPASSWORD')}"
        f"@{os.environ.get('HOST_IP')}:PORT/?authSource=admin"
    )
    
    export_task = BashOperator(
        task_id='export-mongodb',
        bash_command=(
            f"mongoexport --uri='{mongo_uri}' "
            "--collection=mongo_col "
            "--type=json "
            "--out=mongo_col.json"
        )
    )

方案2：在PythonOperator中用subprocess调用mongoexport #

如果需要先执行一些Python逻辑再触发导出，可以用subprocess模块调用命令行工具：

代码示例： #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import subprocess
import os

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
}

def export_from_mongo():
    try:
        # 构建MongoDB连接URI
        mongo_uri = (
            f"mongodb://{os.environ.get('MUSER_NAME')}:{os.environ.get('MPASSWORD')}"
            f"@{os.environ.get('HOST_IP')}:PORT/?authSource=admin"
        )
        
        # 构造mongoexport命令参数列表
        cmd = [
            'mongoexport',
            '--uri', mongo_uri,
            '--collection', 'mongo_col',
            '--type', 'json',
            '--out', 'mongo_col.json'
        ]
        
        # 执行命令并捕获输出
        result = subprocess.run(cmd, check=True, capture_output=True, text=True)
        print(f"导出成功：{result.stdout}")
    except subprocess.CalledProcessError as e:
        print(f"导出失败：{e.stderr}")
        raise
    except Exception as e:
        print(f"错误信息：{str(e)}")
        raise

with DAG(
    'mongodbexport',
    default_args=default_args,
    description='Export data from MongoDB using mongoexport via Python',
    schedule_interval="0 0 * * *",
    catchup=False,
    tags=['export','mongodb'],
) as dag:
    export_task = PythonOperator(
        task_id='export-mongodb',
        python_callable=export_from_mongo
    )

额外注意事项 #

1. Airflow连接优化：建议不要直接在代码中拼接URI，而是在Airflow UI的「Admin -> Connections」中配置MongoDB连接，通过MongoHook获取连接参数，这样更安全且易于维护：

   from airflow.providers.mongo.hooks.mongo import MongoHook
   hook = MongoHook(mongo_conn_id='your_mongo_connection_id')
   # 可从hook中提取参数构建URI
   

2. 文件路径问题：--out指定的是Worker节点本地路径，如果需要将文件存储到外部存储（如S3），可后续添加S3UploadOperator等任务完成上传。
3. 权限检查：确保Airflow Worker进程有执行mongoexport的权限，且能访问目标MongoDB实例和输出目录。

内容的提问来源于stack exchange，提问作者Alex T