Google Cloud部分Dataproc集群无法列出作业问题求助

问题背景 #

在Google Cloud同一项目下部署多个Dataproc集群，使用服务账号调用API列出集群作业时，部分集群可正常返回作业内容，但少数集群无报错信息却返回空结果。用户使用的代码如下：

from google.cloud import dataproc_v1 as dataproc
from google.cloud import bigquery
import google.cloud  as resources
from google.oauth2 import service_account as sa
from google.cloud.dataproc_v1 import JobControllerClient

client = Clients.get_client(ClientType.DATAPROC, name="x")
client_cluster =  Clients.get_client(ClientType.DATAPROC_CLUSTER, name="x")
clus_collec = []

try:
        cl_seq = client_cluster.list_clusters(project_id='y', 
                       region='us-central1')
        for k in cl_seq:
             clus_collec.append(k.cluster_name)

        for m in clus_collec:
            data = client.list_jobs(project_id='y', region='us-central1', filter=f"clusterName={m} AND labels.execution_date=2022-08-25")
            print(data)
except Exception as e:
        print(e)

可能的原因 #

• 标签过滤语法或匹配问题：Dataproc的标签过滤要求值需用单引号包裹，原代码中labels.execution_date=2022-08-25未加引号，可能导致过滤逻辑失效；此外需确认少数集群的作业是否真的带有execution_date=2022-08-25标签，包括标签名、值的拼写和大小写是否完全一致。
• 集群区域不匹配：原代码固定使用us-central1区域查询作业，但少数集群可能部署在其他区域。Dataproc作业是区域级资源，必须使用集群所在区域的JobController客户端才能查询到对应作业。
• 服务账号权限限制：虽然在同一项目下，少数集群可能在创建时配置了更严格的IAM权限，导致服务账号缺少dataproc.jobs.list权限，无法读取这些集群的作业数据。
• 作业生命周期过期：Dataproc默认保留作业历史30天，如果少数集群的作业创建时间早于30天，可能已被自动清理，因此查询返回空结果。
• 自定义客户端初始化问题：代码中使用Clients.get_client自定义方法获取客户端，可能该方法对少数集群的区域或配置处理有误，导致客户端无法正确连接到对应区域的Dataproc服务。

代码优化建议 #

针对上述问题，可对代码进行以下调整：

from google.cloud import dataproc_v1 as dataproc

# 初始化集群控制器客户端（指定区域）
cluster_client = dataproc.ClusterControllerClient(
    client_options={"api_endpoint": "us-central1-dataproc.googleapis.com:443"}
)
cluster_list = []

try:
    # 列出指定区域的所有集群，保存集群名称和所在区域
    clusters = cluster_client.list_clusters(project_id='y', region='us-central1')
    for cluster in clusters:
        cluster_list.append({
            "name": cluster.cluster_name,
            "region": cluster.region
        })

    # 遍历每个集群，使用对应区域的作业控制器查询作业
    for cluster in cluster_list:
        job_client = dataproc.JobControllerClient(
            client_options={"api_endpoint": f"{cluster['region']}-dataproc.googleapis.com:443"}
        )
        # 修正标签过滤语法，给值添加单引号
        jobs = job_client.list_jobs(
            project_id='y',
            region=cluster['region'],
            filter=f"clusterName={cluster['name']} AND labels.execution_date='2022-08-25'"
        )
        # 迭代打印作业详情，而非直接打印迭代器对象
        print(f"集群 {cluster['name']} 的作业：")
        for job in jobs:
            print(f"作业ID: {job.job_id}, 状态: {job.status.state}")
except Exception as e:
    print(f"错误信息: {e}")

优化点说明 #

1. 保存集群的区域信息，针对每个集群的区域创建对应的JobController客户端，避免区域不匹配问题。
2. 修正标签过滤语法，给标签值添加单引号，符合Dataproc的过滤规则。
3. 遍历作业结果并打印具体信息，原代码直接打印data（迭代器对象）无法看到实际作业内容。
4. 使用官方标准的客户端初始化方式，避免自定义客户端可能带来的配置问题。

内容的提问来源于stack exchange，提问作者maries