Azure Databricks如何用Python命令获取当前集群配置信息

不同集群环境的实现方法不同，以下是主流大数据集群场景的可运行Python代码：

1. PySpark托管集群 #

适用所有用Spark做ETL的场景，可直接获取当前任务分配的工作节点资源：

from pyspark.sql import SparkSession
import pandas as pd

# 初始化Spark会话
spark = SparkSession.builder.appName("fetch_cluster_config").getOrCreate()
sc = spark.sparkContext

# 拉取执行器信息，过滤Driver节点
executor_infos = sc._jsc.sc().statusTracker().getExecutorInfos()
driver_host = sc._jsc.sc().conf().get("spark.driver.host")
single_exe_mem = sc._jsc.sc().conf().get("spark.executor.memory")

worker_config = []
for exe in executor_infos:
    if exe.host() != driver_host:
        worker_config.append({
            "executor_id": exe.executorId(),
            "worker_host": exe.host(),
            "allocated_cores": exe.cores(),
            "allocated_memory": single_exe_mem
        })

# 转Pandas DataFrame
worker_df = pd.DataFrame(worker_config)
# 也可以转Spark DataFrame存入数仓做长期跟踪
spark.createDataFrame(worker_config).write.mode("append").saveAsTable("ods.cluster_worker_spec_history")

可直接获取的核心指标：工作节点数量为len(worker_config)，单节点规格为每个节点的核心数、内存配置。

2. YARN管理的Hadoop集群 #

适用需要拉取YARN集群全量在线工作节点配置的场景：
首先安装依赖：pip install hadoop-yarn-api

from yarn_api_client import ResourceManager
import pandas as pd

# 替换为你的YARN ResourceManager地址、端口
rm = ResourceManager(address="your-resource-manager-host", port=8088)
all_nodes = rm.cluster_nodes().data["nodes"]["node"]

yarn_worker_config = []
for node in all_nodes:
    # 仅筛选在线运行的工作节点
    if node["nodeState"] == "RUNNING":
        yarn_worker_config.append({
            "node_id": node["id"],
            "node_host": node["nodeHostName"],
            "total_cores": node["availableVirtualCores"] + node["usedVirtualCores"],
            "total_memory_mb": node["availableMemoryMB"] + node["usedMemoryMB"],
            "used_cores": node["usedVirtualCores"],
            "used_memory_mb": node["usedMemoryMB"]
        })

yarn_worker_df = pd.DataFrame(yarn_worker_config)

3. Kubernetes管理的大数据集群 #

适用K8s搭建的Spark、Flink等集群的节点配置拉取：
首先安装依赖：pip install kubernetes

from kubernetes import client, config
import pandas as pd

# 本地调试用config.load_kube_config()，集群内运行用config.load_incluster_config()
config.load_kube_config()
core_api = client.CoreV1Api()

# 替换为你集群实际的工作节点标签，常用标签还有node.kubernetes.io/worker
worker_nodes = core_api.list_node(label_selector="node-role.kubernetes.io/worker=").items

k8s_worker_config = []
for node in worker_nodes:
    k8s_worker_config.append({
        "node_name": node.metadata.name,
        "total_cpu": node.status.capacity["cpu"],
        "total_memory": node.status.capacity["memory"],
        "instance_spec": node.metadata.labels.get("beta.kubernetes.io/instance-type", "custom-spec"),
        "node_status": node.status.conditions[-1].type
    })

k8s_worker_df = pd.DataFrame(k8s_worker_config)

注意事项 #

• 运行代码的账号需要拥有对应集群的资源查看权限，否则会返回权限类报错
• 可将生成的DataFrame定期写入数表，长期跟踪不同ETL任务的资源占用率，匹配最优工作节点规格

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