Airflow 1.9运行HiveOperator报错：'HiveCliHook'对象无'upper'属性

这个错误我之前帮别人排查过，本质就是参数传递类型不匹配导致的——你大概率是把HiveCliHook的实例传给了原本应该接收字符串的参数，Airflow内部代码会对这个参数调用upper()方法处理，但Hook对象根本没有这个方法，所以报错了。

错误原因分析 #

Airflow 1.9的HiveOperator里，hive_cli_conn_id（或者hive_conn_id）参数是用来指定你在Airflow UI中配置的Hive连接ID的，这个参数需要传字符串。如果错误地把你创建的HiveCliHook实例传给了这个参数，代码执行到处理连接ID的逻辑时，就会尝试调用upper()，自然就抛出了'HiveCliHook' object has no attribute 'upper'的错误。

正确的写法分两种情况： #

情况1：使用默认/已配置的连接，不需要自定义Hook

直接传连接ID字符串给hive_cli_conn_id参数就行，不用手动创建HiveCliHook实例：

import airflow
from airflow.operators.hive_operator import HiveOperator
from airflow.models import DAG
from datetime import timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': ['support@mail.com'],
    'email_on_failure': True,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'hive_dag_example',
    default_args=default_args,
    description='A simple Hive DAG',
    schedule_interval=timedelta(days=1),
)

# 正确实例化HiveOperator
hive_query_task = HiveOperator(
    task_id='execute_hive_query',
    hive_cli_conn_id='your_hive_connection_id',  # 这里填你在Airflow UI配置的连接ID字符串
    hql='SELECT * FROM your_target_table LIMIT 10',
    dag=dag
)

情况2：需要自定义HiveCliHook的配置

如果你要修改Hook的默认行为（比如自定义超时、修改CLI参数等），要通过hook参数传递实例，而不是覆盖hive_cli_conn_id：

import airflow
from airflow.operators.hive_operator import HiveOperator
from airflow.hooks.hive_hooks import HiveCliHook
from airflow.models import DAG
from datetime import timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': ['support@mail.com'],
    'email_on_failure': True,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'custom_hive_dag_example',
    default_args=default_args,
    description='A custom Hive DAG',
    schedule_interval=timedelta(days=1),
)

# 创建自定义Hook实例
custom_hive_hook = HiveCliHook(
    hive_cli_conn_id='your_hive_connection_id',
    # 这里可以加自定义配置，比如指定队列参数
    hive_cli_params="-hiveconf mapred.job.queue.name=your_queue"
)

# 传递给HiveOperator的hook参数
hive_query_task = HiveOperator(
    task_id='execute_custom_hive_query',
    hook=custom_hive_hook,
    hql='INSERT INTO your_table SELECT * FROM source_table',
    dag=dag
)

最后提醒 #

检查你原本的代码，看看是不是把HiveCliHook()的实例不小心传给了hive_cli_conn_id或者hive_conn_id参数，改成上面的正确写法就能解决这个问题了。

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