Airflow使用SnowflakeLoadOperator报expected str not function错误咨询

错误根因 #

你在MonthlyBillingS3PartitionLatestFile类的formatted_partition_value属性中，直接将存储函数对象的self.source_resolver作为参数传给了os.path.split，没有执行函数调用操作。os.path.split只能接收字符串、字节或者路径类对象，无法处理函数对象，因此触发类型错误。

修复方案 #

方案1：直接调用传入的解析函数 #

修改formatted_partition_value属性逻辑，执行函数拿到返回的S3路径字符串：

@property
def formatted_partition_value(self):
    # 加括号执行source_resolver函数，获取返回的路径字符串
    source = self.source_resolver()
    prefix, filename = os.path.split(source)
    # 保留你后续原有逻辑

如果get_latest_io_monthly_billing_file_from_new_prefix需要传入参数，调用时同步传入即可，比如：

source = self.source_resolver(bucket=self.bucket, base_path=self.location_base)

方案2：兼容路径字符串/解析函数两种传参场景 #

如果希望source_resolver参数同时支持直接传固定路径、传动态解析函数两种用法，可以加可调用判断：

@property
def formatted_partition_value(self):
    source = self.source_resolver
    # 如果是可调用对象，先执行拿到返回值
    if callable(source):
        source = source()
    prefix, filename = os.path.split(source)

验证注意点 #

• 确认get_latest_io_monthly_billing_file_from_new_prefix的执行时机是在Airflow任务运行阶段，而非DAG解析阶段，避免函数依赖的S3连接、Airflow变量还未加载就执行报错
• 测试函数返回值是否为合法的S3路径字符串，避免空值、格式错误的问题

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