python - 通过 Python SDK 将 U-SQL 作业部署到 Azure 时出现无明确原因的异常

Question

我正在使用 python SDK 使用 adlaJobClient 将作业提交到 Azure，我有大约 30 个使用 JINJA2 构建的动态 USQL，我将其填充在列表中，然后使用 adlaJobClient 将它们一一推送到 Azure，我遇到的问题我面临的是在随机数量的成功部署之后，python 在程序控制台中抛出异常，没有任何进一步的解释。Azure 中也没有 U-SQL 作业失败的实例，下面提到的是 python 中的错误堆栈跟踪，当我运行相同的 U-SQL 查询，我正在动态生成该查询......其执行正在停止，它在 Azure 中运行而不会失败(手动)

***** starting query number **** 24Job is not yet done, waiting for 3 seconds. Current state: CompilingJob is not yet done, waiting for 3 seconds. Current state: CompilingJob is not yet done, waiting for 3 seconds. Current state: CompilingJob is not yet done, waiting for 3 seconds. Current state: startingJob is not yet done, waiting for 3 seconds. Current state: startingAn exception has occurred, use %tb to see the full traceback.SystemExit: 1

 for script in sql_query_list:
    jobId = str(uuid.uuid4())

    jobResult = adlaJobClient.job.create(adla,jobId,JobInformation(name='Submit ADLA Job '+jobId,type='USql',properties=USqlJobProperties(script=script)))

    try:

        while(jobResult.state != JobState.ended):
            print('Job is not yet done, waiting for 3 seconds. Current state: ' + jobResult.state.value)
            time.sleep(3)
            jobResult = adlaJobClient.job.get(adla, jobId)

        print(' ******* JOB ID  ********',jobId)
        print("****QUERY no FINISHED *****",sql_query_list.index(script))
        print ('**** JOB ID RESULT: ****** ' + jobResult.result.value)

    except Exception as e:
        raise ValueError
        print ("xxxxxx JOB SUBMISSION TO ADLA FAILED xxxxxxx")
        print(e)

Answer 1

选项 A:手动登录 Portal

检查失败作业的最简单方法是登录 Azure 门户 ( http://portal.azure.com )，导航到 Data Lake Analytics 帐户，然后单击“查看所有作业”。从作业列表中，您可以导航到您的作业并查看带有特定错误消息的输出。 (继续阅读自动化方法。)

选项 B:自动化 Python 作业

您可以使用job_result.properties.errors获取带有错误消息的作业属性。下面的代码示例将对任何 U-SQL 作业错误执行“ pretty-print ”，并引发带有这些详细信息的异常。

解析错误信息:

def get_pretty_error(error_message_obj):
    """
    Returns a string describing the USQL error.
    error_message_obj can be obtained via `job_result.error_message[0]`
    """
    err_info = error_message_obj.__dict__
    error_msgs = "=" * 80 + "\n"
    error_msgs += "=" * 6 + " ERROR: {}".format(err_info.pop("description", None)) + "\n"
    error_msgs += "=" * 80 + "\n"
    error_msgs += err_info.pop("details", None).replace("\\r\\n", "\n").replace("\\n", "\n").replace("\\t", "\t").rstrip() + "...\n"
    error_msgs += "=" * 80 + "\n"
    error_msgs += "Message:    {}\n".format(err_info.pop("message", None))
    error_msgs += "Severity:   {}\n".format(str(err_info.pop("severity", None)).upper())
    error_msgs += "Resolution: {}\n".format(err_info.pop("resolution", None))
    inner = err_info.pop("inner_error", None)
    for key in ["end_offset", "line_number", "start_offset", "source", "additional_properties"]:
        # ignore (don't print these)
        err_info.pop(key, None)
    err_info = {x: y for x, y in err_info.items() if y} # Remove empty keys
    error_msgs += "Addl. Info:\n\t{}\n".format(
        yaml.dump(err_info,
                  default_flow_style=True
                  ).replace("\\t", "\t").replace("\\n", "\n").replace("\n", "\n\t"))
    if inner:
        # If there's an inner error, concatenate that message as well recursively
        error_msgs += _get_pretty_error(inner, ordinal_text + " (B)")
    return error_msgs

在等待作业函数中使用它:

def wait_for_usql_job(adlaJobClient, adla_account, job_id):
    """Wait for completion, on error raise an exception (with specific details)"""
    print("Waiting for job ID '{}'".format(job_id))
    job_result = adlaJobClient.job.get(adla_account, job_id)
    while(job_result.state != JobState.ended):
        print('Job is not yet done, waiting for 3 seconds. Current state: ' + job_result.state.value)
        time.sleep(3)
        job_result = adlaJobClient.job.get(adla_account, job_id)
    job_properties = job_result.properties.__dict__
    detail_msg = (
        "\tCompilation Time: {}\n".format(job_properties.pop("total_running_time", None)) +
        "\tQueued Time:      {}\n".format(job_properties.pop("total_compilation_time", None)) +
        "\tExecution Time:   {}\n".format(job_properties.pop("total_queued_time", None)))
    print('Job completed with result: {}\n{}'
          .format(job_result.result.value, detail_msg))
    if job_result.result.value == "Succeeded":
        return job_result.result.value
    elif job_result.result.value == "Failed":
        error_msgs = ""
        for error in job_result.error_message:
            # Loop through errors and concatenate error messages
            error_msgs += get_pretty_error(error)
        raise Exception("Job execution failed for job_id '{}':\n{}"
                        .format(job_id, error_msgs))

注释:

• 由于这些类在网上没有详细记录，因此我使用 __dict__ 属性来探索 JobProperties 和 ErrorInfo 对象的所有属性，并处理任何空白或我不需要的属性并打印其余的。
• 您可以选择重写此代码以显式调用这些属性，而无需使用 __dict__。
• 任何错误都可能有一个内部错误，这就是为什么我将 get_pretty_error() 编写为它自己的函数 - 以便它可以递归地调用自身。