气流失败：ParseException 行 2:0 无法识别附近的输入答案

【问题标题】：Airflow Failed: ParseException line 2:0 cannot recognize input near气流失败：ParseException 行 2:0 无法识别附近的输入
【发布时间】：2018-11-13 19:13:51
【问题描述】：

我正在尝试在 Airflow 上运行测试任务，但我不断收到以下错误：

失败：ParseException 2:0 无法识别“create_import_table_fct_latest_values”附近的输入。 'hql'

这是我的 Airflow Dag 文件：

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

args = {
    'owner': 'raul',
    'start_date': datetime(2018, 11, 12),
    'provide_context': True,
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email': ['raul.gregglino@leroymerlin.ru'],
    'email_on_failure': True,
    'email_on_retry': False
}

dag = DAG('opus_data', 
    default_args=args,
    max_active_runs=6,
    schedule_interval="@daily"
)

import_lv_data = HiveOperator(
    task_id='fct_latest_values',
    hive_cli_conn_id='metastore_default',
    hql='create_import_table_fct_latest_values.hql ',
    hiveconf_jinja_translate=True,
    dag=dag
    )

deps = {}

# Explicity define the dependencies in the DAG
for downstream, upstream_list in deps.iteritems():
    for upstream in upstream_list:
        dag.set_dependency(upstream, downstream)

这是我的 HQL 文件的内容，以防万一这可能是问题而我无法确定：

*I'm testing the connection to understand if the table is created or not, then I'll try to LOAD DATA, hence the LOAD DATA is commented out.
CREATE TABLE IF NOT EXISTS opus_data.fct_latest_values_new_data (
    id_product          STRING,
    id_model            STRING,
    id_attribute        STRING,
    attribute_value     STRING
) ROW FORMAT DELIMITED FIELDS TERMINATED ',';

#LOAD DATA LOCAL INPATH
#'/media/windows_share/schemas/opus/fct_latest_values_20181106.csv'
#OVERWRITE INTO TABLE opus_data.fct_latest_values_new_data;

【问题讨论】：

标签： hadoop hive airflow

【解决方案1】：

在 HQL 文件中应该是FIELDS TERMINATED BY ','：

CREATE TABLE IF NOT EXISTS opus_data.fct_latest_values_new_data (
    id_product          STRING,
    id_model            STRING,
    id_attribute        STRING,
    attribute_value     STRING
) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';

并且 cmets 应该在 HQL 文件中以 -- 开头，而不是 #

这似乎不正确并导致异常hql='create_import_table_fct_latest_values.hql '

看看这个例子：

 #Create full path for the file
    hql_file_path = os.path.join(os.path.dirname(__file__), source['hql'])
    print hql_file_path
    run_hive_query = HiveOperator(
        task_id='run_hive_query',
        dag = dag,
        hql = """
        {{ local_hive_settings }}
        """ + "\n " + open(hql_file_path, 'r').read()
)

更多详情请见here。

或者将所有HQL放入hql参数中：

hql='CREATE TABLE IF NOT EXISTS opus_data.fct_latest_values_new_data ...'

【讨论】：

第一，你的昵称让我笑了。好东西。第二。感谢您更新蜂巢中的评论，有一段时间我不使用了。第三。我试图找出的错误正是造成错误的原因，即使使用您给我的示例，我也无法解释和应用。

【解决方案2】：

我设法找到了我的问题的答案。

这与我的 HiveOperator 调用文件的路径有关。由于没有定义任何变量来告诉 Airflow 在哪里寻找，我收到了我在帖子中提到的错误。

一旦我使用网络服务器界面定义了它（见图），我的 dag 就开始正常工作了。

我对我的 DAG 代码进行了更改，仅针对组织的文件位置，这就是我的 HiveOperator 现在的样子：

import_lv_data = HiveOperator(
    task_id='fct_latest_values',
    hive_cli_conn_id='metastore_default',
    hql='hql/create_import_table_fct_latest_values2.hql',
    hiveconf_jinja_translate=True,
    dag=dag
    )

感谢 (@panov.st) 亲自帮助我发现了我的问题。

【讨论】：