本文旨在解决Airflow中S3Hook的`download_file`函数在下载S3文件时，目标路径意外生成`airflow_tmp_`临时子目录导致`FileNotFoundError`的问题。我们将深入探讨`download_file`的默认行为，并提供使用`preserve_file_name`和`use_autogenerated_subdir`参数来精确控制文件下载路径和命名的方法，确保文件按预期存储。
理解S3Hook download_file 的默认行为
在Apache Airflow中，S3Hook提供了一个便捷的方式与Amazon S3服务进行交互。其中，download_file函数用于将S3存储桶中的文件下载到本地文件系统。然而，开发者在使用此函数时常会遇到一个非预期的行为：即使指定了明确的本地目标路径，下载的文件有时会被放置在一个自动生成的airflow_tmp_开头的临时子目录中，这可能导致后续文件操作失败并抛出FileNotFoundError。
这种默认行为的出现，是由于S3Hook在设计上为了某些内部处理或确保原子性，可能会在目标路径下创建临时目录来存放下载的文件。当用户期望文件直接位于指定路径时，这种行为就会造成困扰。
考虑以下一个典型的使用场景，尝试从S3下载文件并读取其内容：
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.operators.python import PythonOperator
from airflow.models.dag import DAG
from datetime import datetime
import os
def s3_extract(key: str, bucket_name: str, local_path: str) -> str:
"""
从S3下载文件并读取其内容。
"""
source_s3_key = key
source_s3_bucket = bucket_name
dest_file_path = local_path # 期望的本地目标目录
# 确保本地目标目录存在
if not os.path.exists(dest_file_path):
os.makedirs(dest_file_path)
print(f"Created directory: {dest_file_path}")
source_s3 = S3Hook(aws_conn_id="aws_conn_str") # 假设已配置名为"aws_conn_str"的AWS连接
# 尝试下载文件，期望其位于 dest_file_path/filename.txt
# 注意：这里直接拼接了文件名，但 S3Hook 可能会在 dest_file_path 下创建子目录
target_local_file = os.path.join(dest_file_path, os.path.basename(key))
# 原始问题中的调用方式：
# source_s3.download_file(source_s3_key, source_s3_bucket, f"{dest_file_path}/filename.txt")
# 这种方式可能导致文件被下载到 f"{dest_file_path}/filename.txt/airflow_tmp_..."
# 更准确的原始问题模拟，直接指定目标文件路径，但S3Hook可能在其父目录创建临时文件夹
source_s3.download_file(
key=source_s3_key,
bucket_name=source_s3_bucket,
local_path=target_local_file # 期望的完整本地文件路径
)
# 尝试打开文件
try:
with open(target_local_file, "r") as file:
text = file.read()
print(f"File content: {text[:100]}...") # 打印前100个字符
return text
except FileNotFoundError as e:
print(f"Error: File not found at {target_local_file}. Details: {e}")
# 在这里，如果S3Hook创建了临时子目录，这个错误就会发生
raise # 重新抛出异常以便Airflow捕获
with DAG(
dag_id='s3_download_tutorial_dag',
start_date=datetime(2023, 1, 1),
schedule_interval=None,
catchup=False,
tags=['s3', 'tutorial'],
) as dag:
download_job = PythonOperator(
task_id="s3_download_task",
python_callable=s3_extract,
op_kwargs={
'key': 'airflow/docs/filename.txt',
'bucket_name': 's3-dev-data-001', # 替换为你的S3桶名
'local_path': '/tmp/airflow_data' # 替换为你的本地路径，确保Airflow worker有写入权限
}
)当上述代码执行时，如果S3Hook的默认行为触发，可能会观察到类似以下FileNotFoundError：
FileNotFoundError: [Errno 2] no such file or directory: '/tmp/airflow_data/filename.txt/airflow_tmp_90_6ogw5'
这表明S3Hook并没有将文件直接下载到/tmp/airflow_data/filename.txt，而是在其下创建了一个名为airflow_tmp_90_6ogw5的子目录，并将文件放置其中。
解决方案：控制文件下载行为
为了解决这个问题，S3Hook.download_file函数提供了两个关键参数，允许我们精确控制文件的下载位置和命名：
- use_autogenerated_subdir (默认为 True): 当设置为 False 时，S3Hook将不会在指定的 local_path 下自动创建临时子目录。文件将直接下载到 local_path 所指定的路径。
- preserve_file_name (默认为 False): 当设置为 True 时，下载的文件将保留其原始S3对象的名称。这确保了即使local_path只指定了一个目录，文件也会以其S3名称存储在该目录下。结合use_autogenerated_subdir=False使用时，它能更好地保证文件名称的预期性。
通过将这两个参数设置为 False 和 True，我们可以强制S3Hook将文件直接下载到我们指定的完整本地文件路径。
以下是修改后的 s3_extract 函数：
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.operators.python import PythonOperator
from airflow.models.dag import DAG
from datetime import datetime
import os
def s3_extract_corrected(key: str, bucket_name: str, local_path: str) -> str:
"""
从S3下载文件并读取其内容，使用参数控制文件下载路径。
"""
source_s3_key = key
source_s3_bucket = bucket_name
dest_dir = local_path # 期望的本地目标目录
# 确保本地目标目录存在
if not os.path.exists(dest_dir):
os.makedirs(dest_dir)
print(f"Created directory: {dest_dir}")
source_s3 = S3Hook(aws_conn_id="aws_conn_str")
# 构建完整的本地文件路径
# os.path.basename(key) 从S3 key中提取文件名
target_local_file_path = os.path.join(dest_dir, os.path.basename(key))
print(f"Attempting to download S3://{source_s3_bucket}/{source_s3_key} to {target_local_file_path}")
# 使用 preserve_file_name=True 和 use_autogenerated_subdir=False
# 将文件直接下载到 target_local_file_path
source_s3.download_file(
key=source_s3_key,
bucket_name=source_s3_bucket,
local_path=target_local_file_path,
preserve_file_name=True, # 确保文件名与S3对象名一致
use_autogenerated_subdir=False # 禁用自动生成临时子目录
)
# 尝试打开文件
try:
with open(target_local_file_path, "r") as file:
text = file.read()
print(f"Successfully downloaded and read file from {target_local_file_path}. Content snippet: {text[:100]}...")
return text
except FileNotFoundError as e:
print(f"Error: File not found at {target_local_file_path}. Details: {e}")
raise
except Exception as e:
print(f"An unexpected error occurred while reading the file: {e}")
raise
with DAG(
dag_id='s3_download_tutorial_dag_corrected',
start_date=datetime(2023, 1, 1),
schedule_interval=None,
catchup=False,
tags=['s3', 'tutorial', 'fix'],
) as dag_corrected:
download_job_corrected = PythonOperator(
task_id="s3_download_task_corrected",
python_callable=s3_extract_corrected,
op_kwargs={
'key': 'airflow/docs/filename.txt',
'bucket_name': 's3-dev-data-001', # 替换为你的S3桶名
'local_path': '/tmp/airflow_data' # 替换为你的本地路径，确保Airflow worker有写入权限
}
)注意事项与最佳实践
- 目标目录存在性： 在调用download_file之前，务必确保local_path（即你希望文件存放的父目录）是存在的。S3Hook不会自动创建这些父目录。可以使用os.makedirs(local_path, exist_ok=True)来确保目录存在。
- Airflow Worker权限： 确保运行Airflow Worker的用户对指定的local_path具有写入权限。否则，即使路径正确，下载操作也会因权限问题而失败。
- AWS连接配置： S3Hook依赖于一个配置好的AWS连接。在Airflow UI中，导航到 Admin -> Connections，创建一个类型为 Amazon Web Services 的连接，并配置 aws_access_key_id 和 aws_secret_access_key，或使用IAM角色。确保aws_conn_id参数与你创建的连接ID匹配。
- 错误处理： 在文件操作（如打开、读取）周围添加try-except块，以优雅地处理可能发生的FileNotFoundError或其他IO错误。
- 清理临时文件： 如果你的任务是临时下载文件进行处理，处理完毕后最好清理这些本地文件，以避免占用过多的磁盘空间，特别是在共享的Airflow Worker环境中。
总结
S3Hook.download_file函数在Airflow中是一个强大的工具，但其默认的临时文件处理行为可能会导致意外的FileNotFoundError。通过理解并正确使用preserve_file_name=True和use_autogenerated_subdir=False这两个关键参数，开发者可以完全控制文件的下载路径和命名，确保数据管道的稳定性和可预测性。在构建Airflow任务时，始终建议查阅相关Hook的官方文档，以充分了解其参数和行为，从而避免常见陷阱。
