如何使用 DataprocHiveOperator 从 Hive 作业输出日志中提取查询结果？

Question

我正在尝试使用 Airflow 构建数据迁移管道，源是 Dataproc 集群上的 Hive 表，目标是 BigQuery。我正在使用 DataprocHiveOperator 来获取架构以及来自源的数据。此运算符在内部使用 Dataproc REST API 在我们指定的 Dataproc 集群上提交和执行作业。输出将作为作业日志的一部分写入文件 Google 云存储。我只需要这些日志中的查询结果。

到目前为止，我已经修改了 gcp_dataproc_hook.py 代码，通过 driverOutputResourceUri 参数将输出文件的内容下载为字符串，从而将输出返回给调用方法。此输出的返回类型是 Pandas 数据框（可以根据我们的方便更改为任何其他类型）。但这包括完整的日志。我必须从中提取查询结果。

这是我在 gcp_dataproc_hook.py 中添加的代码 sn-p 以返回已提交查询的输出日志：

    #download the output
    def getOutput(self,project, output_bucket,output_path):
        client = storage.Client(project=self.project_id)
        bucket = client.get_bucket(output_bucket)
        output_blob = ('/'.join(output_path)+"."+"000000000")
        return bucket.blob(output_blob).download_as_string()

    #get logs including query output
    def getQueryResult(self):
        result=self.job_ouput
        output = self.getOutput(result['reference']['projectId'],result['driverOutputResourceUri'].split('/')[2],result['driverOutputResourceUri'].split('/')[3:])
        df = pd.read_csv(io.BytesIO(output), sep='\n|', nrows=500000, engine='python')
        return df

这是我尝试执行的示例查询：

SHOW CREATE TABLE my_tbl;

输出日志如下所示：

Connecting to jdbc:hive2://prod-metastore-test-cluster1-m:10000
0            Connected to: Apache Hive (version 2.3.5)             
1                    Driver: Hive JDBC (version 2.3.5)             
2    Transaction isolation: TRANSACTION_REPEATABLE_...             
3    . . . . . . . . . . . . . . . . . . . . . . .>...             
4    |                   createtab_stmt            ...             
5    +---------------------------------------------...             
6    | CREATE TABLE `my_tbl`(       ...             
7    |   `col1` string,            ...             
8    |   `col2` bigint,                   ...             
9    |   `col3` string,                 ...                         
..                                                 ...                         
141  |   `coln` string)                 ...             
142  | ROW FORMAT SERDE                            ...             
143  |   'org.apache.hadoop.hive.ql.io.orc.OrcSerde...             
144  | STORED AS INPUTFORMAT                       ...             
145  |   'org.apache.hadoop.hive.ql.io.orc.OrcInput...             
146  | OUTPUTFORMAT                                ...             
147  |   'org.apache.hadoop.hive.ql.io.orc.OrcOutpu...             
148  | LOCATION                                    ...             
149  |   'gs://my_hive_data_bucket/tmp/base_table/my_tbl...             
150  | TBLPROPERTIES (                             ...             
151  |   'transient_lastDdlTime'='1566842329')     ...             
152  +---------------------------------------------...             
153                  143 rows selected (0.154 seconds)             
154               Beeline version 2.3.5 by Apache Hive             
155  Closing: 0: jdbc:hive2://prod-metastore-test-c... 

预期的输出应该是这样的：

CREATE TABLE `my_tbl`(
  `col1` string,
  `col2` bigint,
  `col3` string,
  ..
  `coln` string,  
)
ROW FORMAT SERDE
  'org.apache.hadoop.hive.ql.io.orc.OrcSerde'
STORED AS INPUTFORMAT
  'org.apache.hadoop.hive.ql.io.orc.OrcInputFormat'
OUTPUTFORMAT
  'org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat'
LOCATION
  'gs://my_hive_data_bucket/tmp/base_table/my_tbl'
TBLPROPERTIES (
  'transient_lastDdlTime'='1566842329')

请建议我一种可以接近解决方案的方法。

Answer 1

在 Dataproc 中，Hive 查询使用 Beeline 而不是已弃用的 Hive CLI，这就是默认情况下格式不同的原因。 Beeline 通常会以花哨的边框格式格式化人类可读的输出，而不是更容易解析的格式。

幸运的是，有一些直线选项可以使格式非常接近旧 Hive CLI 所做的。您可以简单地创建一个初始化操作，在创建 Dataproc 集群时添加到选项中，并在您的 Airflow 运算符中指定 init_actions_uris。创建一个包含以下内容的文件：

#!/bin/bash

sed -i 's/beeline/beeline --outputformat=tsv2 --silent=true/' /usr/bin/beeline

然后将该文件上传到 GCS，例如 gs://some-gcs-bucket/beeline-legacyfmt.sh，并将该 GCS URI 设置为 Dataproc 集群的初始化操作。默认情况下，这将应用直线所需的命令行选项。然后，您发送的任何 Dataproc Hive 作业现在将以“tsv2”和“静默”模式输出，这意味着没有多余的日志语句，并且输出将是原始 tsv。