Airflow导入CSV至MySQL触发secure-file-priv报错求解

你遇到的1290报错由3个常见问题叠加导致，和你之前改配置不生效的现象完全匹配：

• 路径不匹配：Airflow文件校验任务检查的路径是/opt/airflow/store_files/students_data.csv，但你写的LOAD DATA语句里文件路径是/store_files/students_data.csv，两边路径完全不一致。
• 配置未生效：修改my.cnf里的secure-file-priv参数后必须重启MySQL服务才会加载，你可以登录MySQL执行SHOW VARIABLES LIKE 'secure_file_priv';验证，返回值和你配置不一致就说明配置没加载。如果是容器化部署，还要确认你修改的是MySQL容器内的配置文件，而不是Airflow worker容器里的配置。
• 逻辑认知偏差：普通LOAD DATA INFILE语句读取的是MySQL服务端所在机器的文件，不是Airflow worker节点上的文件。如果你的CSV只存在Airflow worker本地，MySQL服务端进程本身就访问不到这个文件，就算secure-file-priv配置完全正确也会报错。

以下两种方案任选其一即可，优先推荐方案2，不需要调整MySQL全局配置，适配绝大多数Airflow部署场景。

方案1：适配服务端LOAD DATA逻辑 #

如果你的CSV文件可以放在MySQL服务端可访问的路径下，按以下步骤调整：

1. 确认配置生效：修改my.cnf的[mysqld]段下的secure-file-priv参数（如果要允许任意路径，值设为空字符串即可；如果要限制指定目录，就填目录的绝对路径，注意路径末尾要加斜杠），修改后重启MySQL服务，执行SHOW VARIABLES LIKE 'secure_file_priv';确认返回值和配置一致。

   容器部署注意：必须将CSV所在目录挂载到MySQL容器内部，例如把宿主机的/opt/airflow/store_files挂载到MySQL容器的/import_data路径，secure-file-priv就配置为/import_data/。

2. 修正SQL语句里的文件路径，填写MySQL服务端实际能访问到的绝对路径，例如挂载后的路径为/import_data/students_data.csv，SQL修改为：

   LOAD DATA INFILE '/import_data/students_data.csv' INTO TABLE students_db FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' IGNORE 1 ROWS;
   

3. 给Airflow连接使用的MySQL账号授予FILE权限，登录MySQL执行以下语句：

   GRANT FILE ON *.* TO '你的MySQL账号'@'%';
   FLUSH PRIVILEGES;
   

方案2：绕开服务端文件读取限制（最稳妥） #

不需要修改MySQL全局配置，直接读取Airflow worker上的CSV文件导入，两种实现方式可选：

• 方式一：使用LOAD DATA LOCAL INFILE语法
  该语法读取的是MySQL客户端（也就是Airflow worker）本地的文件，不受服务端secure-file-priv参数限制，只需要两个调整：
  1. 修改导入SQL语句，加上LOCAL关键字，路径填worker上的CSV实际路径：

     LOAD DATA LOCAL INFILE '/opt/airflow/store_files/students_data.csv' INTO TABLE students_db FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' IGNORE 1 ROWS;
     

  2. 打开Airflow里对应MySQL连接（conn_id为mysql_conn）的配置，在Extra字段添加如下配置开启本地文件加载支持：

     {"local_infile": true}
     

• 方式二：用Python直接读取CSV批量插入
  完全绕开LOAD DATA语法，适合文件体积不大的场景，把原来的t3任务替换为PythonOperator即可，示例代码：

  import csv
  from airflow import DAG
  from airflow.operators.bash import BashOperator
  from airflow.providers.mysql.operators.mysql import MySqlOperator
  from airflow.operators.python import PythonOperator
  from airflow.providers.mysql.hooks.mysql import MySqlHook
  from datetime import timedelta
  import airflow.utils.dates
  
  default_args = {
      "owner": "airflow",
      "start_date": airflow.utils.dates.days_ago(1),
      "retries": 1,
      "retry_delay": timedelta(seconds=5)
  }
  
  def load_csv_to_mysql():
      mysql_hook = MySqlHook(mysql_conn_id="mysql_conn")
      conn = mysql_hook.get_conn()
      cursor = conn.cursor()
      # 按你的表结构调整插入字段和占位符数量
      insert_sql = "INSERT INTO students_db (col1, col2, col3) VALUES (%s, %s, %s)"
      with open('/opt/airflow/store_files/students_data.csv', 'r', encoding='utf-8') as f:
          csv_reader = csv.reader(f)
          next(csv_reader) # 跳过表头行
          data = [row for row in csv_reader]
      cursor.executemany(insert_sql, data)
      conn.commit()
      cursor.close()
      conn.close()
  
  with DAG('sql_operator_from_csv_to_mysql',default_args=default_args,schedule_interval='@daily', template_searchpath=['/opt/airflow/sql_files_mysql'], catchup=True) as dag:
      t1 = BashOperator(task_id='check_file_exists', bash_command='ls /opt/airflow/store_files/students_data.csv | sha1sum', retries=2, retry_delay=timedelta(seconds=15))
      t2 = MySqlOperator(task_id='create_mysql_table', mysql_conn_id="mysql_conn", sql="create_table.sql")
      t3 = PythonOperator(task_id='insert_into_table', python_callable=load_csv_to_mysql)
      
      t1 >> t2 >> t3
  

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