如何在Databricks通用集群中获取Spark JobName以提取Delta操作数据？

问题根源：作业集群（Job Cluster）会自动在集群标签中添加RunName字段，但通用集群（All-Purpose Cluster）没有这个标签，导致__get_job_name返回None，无法匹配Delta历史中的记录。

以下是针对不同场景的解决方案：

1. 适配Jobs API提交到通用集群的任务 #

通过作业ID匹配比JobName更准确，作业ID可从Spark配置中直接获取：

修改__extract_delta_operation_data函数如下：

def __extract_delta_operation_data(self) -> tuple[int, int]:
    delta_table = DeltaTable.forPath(self.spark, self.target_table_path)
    history_df = delta_table.history()
    count, bytes_size = 0, 0

    if not is_local_spark():
        relevant_op_data_row = None
        # 获取当前作业ID（Jobs API提交的任务会携带该配置）
        curr_job_id = self.spark.conf.get("spark.databricks.job.id", None)
        
        if curr_job_id:
            # 用jobId匹配Delta历史记录
            relevant_op_data_row = (
                history_df.filter(history_df["job"]["jobId"] == curr_job_id)
                .filter(history_df["operation"].isin(["STREAMING UPDATE", "WRITE"]))
                .sort(history_df.timestamp.desc())
                .collect()
            )
        else:
            # 处理手动笔记本运行场景，提取笔记本名称作为匹配标识
            try:
                notebook_path = dbutils.notebook.entry_point.getDbutils().notebook().getContext().notebookPath().get()
                notebook_name = notebook_path.split("/")[-1]
                relevant_op_data_row = (
                    history_df.filter(history_df["job"]["jobName"] == notebook_name)
                    .filter(history_df["operation"].isin(["STREAMING UPDATE", "WRITE"]))
                    .sort(history_df.timestamp.desc())
                    .collect()
                )
            except Exception:
                # 无有效标识时，直接取最新的符合条件的操作记录
                relevant_op_data_row = (
                    history_df.filter(history_df["operation"].isin(["STREAMING UPDATE", "WRITE"]))
                    .sort(history_df.timestamp.desc())
                    .limit(1)
                    .collect()
                )
        
        if relevant_op_data_row:
            last_op_data_row = relevant_op_data_row[0]
            # 用字段名访问操作指标，避免硬编码索引导致的兼容性问题
            operation_metrics = last_op_data_row["operationMetrics"]
            count = operation_metrics.get("numOutputRows", 0)
            bytes_size = operation_metrics.get("numOutputBytes", 0)

    return count, bytes_size

2. 独立获取运行标识的辅助函数 #

如果需要单独封装标识获取逻辑，可添加以下函数，同时兼容作业集群和通用集群：

def __get_run_identifier(self) -> str:
    # 先尝试获取作业集群的RunName
    all_cluster_tags = {}
    try:
        for tag in json.loads(self.spark.conf.get("spark.databricks.clusterUsageTags.clusterAllTags")):
            all_cluster_tags[tag["key"]] = tag["value"]
        run_name = all_cluster_tags.get("RunName")
        if run_name:
            return run_name
    except Exception:
        pass
    
    # 再尝试获取通用集群的笔记本名称
    try:
        notebook_path = dbutils.notebook.entry_point.getDbutils().notebook().getContext().notebookPath().get()
        return notebook_path.split("/")[-1]
    except Exception:
        return None

在__extract_delta_operation_data中调用该函数即可：

curr_identifier = self.__get_run_identifier()
if curr_identifier:
    relevant_op_data_row = (
        history_df.filter(history_df["job"]["jobName"] == curr_identifier)
        .filter(history_df["operation"].isin(["STREAMING UPDATE", "WRITE"]))
        .sort(history_df.timestamp.desc())
        .collect()
    )
else:
    relevant_op_data_row = (
        history_df.filter(history_df["operation"].isin(["STREAMING UPDATE", "WRITE"]))
        .sort(history_df.timestamp.desc())
        .limit(1)
        .collect()
    )

3. 关键优化点 #

• 放弃依赖RunName集群标签：该标签仅作业集群独有，通用集群无此配置。
• 避免硬编码字段索引：使用last_op_data_row["operationMetrics"]替代索引12，防止Delta版本更新导致字段顺序变化。
• 增加降级逻辑：当无法获取任何标识时，直接取最新的目标操作记录，保证代码可用性。

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