如何使用 AirFlow 提取使用 Apache Livy 批量 POST 方法提交的 Spark 作业客户端日志

Question

我正在使用 Apache Livy 批量 POST 方法提交 Spark 作业。

此 HTTP 请求是使用 AirFlow 发送的。提交作业后，我正在使用批次 ID 跟踪状态。

我想在 Air Flow 日志上显示驱动程序（客户端日志）日志，以避免转到多个位置 AirFlow 和 Apache Livy/Resource Manager。

使用 Apache Livy REST API 是否可以做到这一点？

Answer 1

Livy 有一个端点来获取日志/sessions/{sessionId}/log& /batches/{batchId}/log。

文档：

• https://livy.incubator.apache.org/docs/latest/rest-api.html#get-sessionssessionidlog
• https://livy.incubator.apache.org/docs/latest/rest-api.html#get-batchesbatchidlog

您可以创建如下所示的 Python 函数来获取日志：

http = HttpHook("GET", http_conn_id=http_conn_id)

def _http_rest_call(self, method, endpoint, data=None, headers=None, extra_options=None):
    if not extra_options:
        extra_options = {}

    self.http.method = method
    response = http.run(endpoint, json.dumps(data), headers, extra_options=extra_options)

    return response


def _get_batch_session_logs(self, batch_id):
    method = "GET"
    endpoint = "batches/" + str(batch_id) + "/log"
    response = self._http_rest_call(method=method, endpoint=endpoint)
    # return response.json()
    return response