Airflow中S3Hook导入报错问题咨询及解决方法求助

导入错误的直接原因 #

你的导入报错核心是两个问题：

1. 大小写错误：要导入的类是S3Hook（首字母大写），不是小写的s3Hook
2. 路径错误：Airflow 2.x及以后版本中，AWS相关的Hook统一放在airflow.providers.amazon.aws路径下，旧版本的from airflow.hooks import S3Hook已经废弃

正确的导入语句是：

from airflow.providers.amazon.aws.hooks.s3 import S3Hook

额外代码问题修正 #

除了导入错误，你的DAG代码还有其他需要修正的问题，才能正常运行：

• 缺少datetime和timedelta的导入
• default_arg里的retires拼写错误，应为retries
• process_data函数中，DataFrame列名与返回的字典键不匹配，且some_processing_function未定义（需替换为实际处理逻辑）
• load_s3函数中直接硬编码AWS密钥不符合安全规范，应使用Airflow的AWS连接；同时存在变量名错误（process_data.to_csv应为processed_data.to_csv）
• upload_task的op_kwargs传参错误，不能直接传process_task，需通过XCom拉取处理后的数据

修正后的完整DAG代码 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
import io
import pandas as pd
from datetime import datetime, timedelta


default_arg = {
    'owner': 'airflows',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'start_date': datetime(2024, 5, 21)
}

dag = DAG(
    'upload_s3',
    schedule_interval='@daily',
    default_args=default_arg,
    catchup=False
)

def Extract_data():
    details = {
        'Name': ['Ankit', 'Aishwarya', 'Shaurya', 'Shivangi'],
        'Age': [23, 21, 22, 21],
        'University': ['BHU', 'JNU', 'DU', 'BHU'],
    }
    return details

def process_data(**context):
    data = context['ti'].xcom_pull(task_ids="Extract_data")
    # 列名与返回的字典键对应
    df = pd.DataFrame(data, columns=['Name', 'Age', 'University'])
    # 替换为你的实际处理逻辑，示例添加一列
    df['Is_Adult'] = df['Age'] >= 18
    return df.to_dict('records')

def load_s3(**context):
    # 通过Airflow的AWS连接ID获取凭证（需提前在Airflow UI配置aws_default连接）
    s3_hook = S3Hook(aws_conn_id='aws_default')
    processed_data = context['ti'].xcom_pull(task_ids="process_data")
    df = pd.DataFrame(processed_data)
    
    csv_buffer = io.StringIO()
    df.to_csv(csv_buffer, index=False)
    
    # 使用S3Hook上传，无需手动处理boto3会话
    s3_hook.load_string(
        string_data=csv_buffer.getvalue(),
        key='airflow.csv',
        bucket_name='airflowstuff',
        replace=True
    )

extract_task = PythonOperator(
    task_id='Extract_data',
    python_callable=Extract_data,
    dag=dag
)

process_task = PythonOperator(
    task_id='process_data',
    python_callable=process_data,
    provide_context=True,
    dag=dag
)

upload_task = PythonOperator(
    task_id='upload_s3',
    python_callable=load_s3,
    provide_context=True,
    dag=dag
)

extract_task >> process_task >> upload_task

关键修改说明 #

1. 安全规范：使用S3Hook通过Airflow配置的AWS连接获取凭证，避免硬编码密钥
2. XCom传参：统一通过context['ti'].xcom_pull拉取上游任务数据，避免直接传递任务对象
3. 代码正确性：修正拼写错误、变量名错误，确保DataFrame列名与数据源匹配
4. 简化操作：使用S3Hook.load_string方法直接上传字符串内容，无需手动创建boto3客户端

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