在 Airflow 中使用 Sqoop 运算符使用 '//N' 空字符串参数导致 HIVE table 中的 '/N' 字符串值

Question

我正在尝试使用 sqoop 通过 Airflow Sqoop Operator 传输一些数据。此数据包含空值，我希望它们在我的 HIVE table.

中以 NULL 结束

sqoop_operator_task = SqoopOperator(
    task_id = table[0],
    conn_id = "JDBC_OPA",
    table = table_name,
    cmd_type = "import",
    target_dir = "/sourcedata/sqoop_tmp/"+table_name,
    num_mappers = 1,
    extra_import_options = {
        'delete-target-dir' : "",
        'fields-terminated-by' : r'"[=11=]1"',
        'hive-database' : database_name,
        'hive-table' : table[0],
        'hive-delims-replacement' : "' '",
        'hive-import' : "",
        'hive-overwrite' : "",
        'null-non-string' : r"'\N'",
        'null-string' : r"'\N'"
    },
    dag = dag
)

对空非字符串和空字符串使用 r"'\N'" 会导致从 Airflow 日志中获取以下命令：

sqoop import --username SELECT_OPA_DSC --password MASKED --num-mappers 1 --connect MASKED --target-dir /sourcedata/sqoop_tmp/OWNER_DMK.DMK_AVY --as-textfile --delete-target-dir --fields-terminated-by "[=12=]1" --hive-database test_db --hive-table DMK_AVY --hive-delims-replacement ' ' --hive-import --hive-overwrite --null-non-string '\N' --null-string '\N' --table OWNER_DMK.DMK_AVY

这是根据 Sqoop documentation 格式应如何以 NULL 结尾。但是，我在 table 中以 '\N' 结尾。然而，直接在 shell 中执行完整命令，确实在 HIVE table.

中给我正确的 NULL 值

我想问题出在 Sqoop 作业生成的 HIVE 查询的不同之处。从 Airflow 执行命令时的 HIVE 查询：

ROW FORMAT DELIMITED FIELDS TERMINATED BY '\001' LINES TERMINATED BY '\012' STORED AS TEXTFILE

直接从Shell执行命令时的HIVE查询：

ROW FORMAT DELIMITED FIELDS TERMINATED BY '[=14=]1' LINES TERMINATED BY '2' STORED AS TEXTFILE

请给我一个场景建议，改变我的 sqoop 运算符，使 Airflow 中生成的 HIVE 查询与直接使用来自 Shell 的命令相同，从而产生 NULL 结果蜂巢 table。只删除一个反斜杠是行不通的。

有关 sqoop 运算符和挂钩的更多信息 here。

Answer 1

我找到了解决办法。传球：

        'null-non-string' : r'\N',
        'null-string' : r'\N'

以在命令中作为 --null-non-string \N 传递的参数结束，并最终在 HIVE 中给出 NULL 结果。但是，直接在 shell 中传递这样的参数会引发非法转义字符错误...因此在 shell.

中处理命令的方式有所不同。

在 Airflow 中使用 Sqoop 运算符使用 '//N' 空字符串参数导致 HIVE table 中的 '/N' 字符串值

Using '//N' null-string argument results in '/N' string value in HIVE table using Sqoop Operator in Airflow

python

shell

hive

airflow

sqoop