Airflow 2.5.0中配置MySQL、Sqoop、HDFS Conn_ID求助

嘿，看起来你已经搭好了Sqoop导入的基础DAG框架，接下来咱们一步步搞定MySQL、Sqoop和HDFS的Conn_ID配置——这几个是让任务跑起来的核心环节！

一、先搞懂Airflow Connections的作用 #

Airflow的Connections是用来统一存储各类服务（数据库、大数据组件）的连接信息，既能避免把密码这类敏感信息硬写到代码里，也方便后续统一修改。所有Conn都要在Airflow UI里创建，然后在DAG的Operator中通过conn_id引用。

二、创建MySQL Connection（Sqoop连MySQL的关键） #

Sqoop是通过JDBC连接MySQL的，所以得先把MySQL的连接信息配置好：

1. 打开Airflow UI，点击顶部导航栏的Admin → Connections
2. 点Add a new record新建连接
3. 重点填这些字段：
   • Conn Id：建议命名为mysql_local（后面要在Sqoop配置里关联这个ID）
   • Conn Type：选择MySQL
   • Host：你的MySQL服务器地址（比如localhost或者实际的IP）
   • Schema：要导入数据的目标数据库名
   • Login：MySQL的用户名
   • Password：MySQL的登录密码
   • Port：MySQL默认端口是3306，按你的实际配置填
4. 点Save保存即可

三、创建Sqoop Connection #

你的DAG里用了SqoopOperator，它需要知道Sqoop的执行环境，所以得配置对应的Sqoop Conn：

1. 同样在Connections页面点Add a new record
2. 填写字段：
   • Conn Id：必须和你DAG代码里的conn_id一致，也就是sqoop_local
   • Conn Type：选择Sqoop
   • Host：如果Sqoop和Airflow在同一台服务器，填localhost就行，否则填Sqoop所在服务器的IP
   • Extra（可选但推荐）：用JSON格式配置Sqoop和Hadoop的安装路径，比如：

     {"sqoop_home": "/opt/sqoop", "hadoop_home": "/opt/hadoop"}
     

     替换成你实际的安装目录
3. 点Save保存

四、HDFS Connection（可选但实用） #

如果后续需要Airflow验证HDFS目标目录的合法性，或者要用到其他HDFS相关Operator，可以配置这个Conn：

1. 在Connections页面新建记录
2. 填写：
   • Conn Id：比如hdfs_default
   • Conn Type：选择HDFS
   • Host：HDFS NameNode的地址（比如localhost:9000，按你的HDFS配置来）
   • Login：HDFS的用户名（比如hadoop）
   • Extra：可以配置HDFS的默认文件系统，比如{"fs.defaultFS": "hdfs://localhost:9000"}
3. 保存即可

五、调整你的DAG代码（关键一步！） #

现在要把MySQL的Conn关联到SqoopOperator里，让Sqoop知道该连接哪个MySQL实例。修改你的代码，新增mysql_conn_id参数：

from airflow.models import DAG
from airflow.contrib.operators.sqoop_operator import SqoopOperator
from airflow.utils.dates import days_ago


Dag_Sqoop_Import = DAG(dag_id="SqoopImport",
                      schedule_interval="* * * * *",  # 注意：这个是每分钟执行一次，测试完记得改成合理周期，比如每天凌晨0点："0 0 * * *"
                      start_date=days_ago(2))

sqoop_mysql_import = SqoopOperator(
    conn_id="sqoop_local",  # 对应你创建的Sqoop Conn
    mysql_conn_id="mysql_local",  # 新增：关联刚才创建的MySQL Conn
    table="shipmethod",
    cmd_type="import",
    target_dir="/airflow_sqoopImport",
    num_mappers=1,
    task_id="SQOOP_Import",
    dag=Dag_Sqoop_Import
)

sqoop_mysql_import

六、测试验证 #

1. 把修改后的DAG文件放到Airflow的DAG目录（默认是$AIRFLOW_HOME/dags）
2. 等待Airflow自动加载DAG，或者手动刷新UI
3. 点击DAG的Trigger DAG按钮触发一次运行
4. 查看任务日志，如果成功执行，说明Conn配置没问题啦

如果遇到报错，比如MySQL连接失败，先检查MySQL Conn的用户名、密码、地址是否正确；如果Sqoop执行失败，看看Sqoop Conn里的安装路径是否正确，或者从日志里找具体的错误提示排查。

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