寻求Ambari集群组件/服务状态查询工具或Python脚本

我之前维护Hadoop集群时，也碰到过bash脚本处理Ambari API返回结果太繁琐的问题，后来改用Python封装API调用，灵活性和可读性提升了不少。下面给你几个实用的脚本示例，刚好覆盖你需要的「获取组件状态」的需求：

基础示例：获取单个组件的状态 #

这个脚本会调用Ambari的API，指定集群名称和组件名称（比如DataNode），返回该组件所有实例的运行状态：

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning

# 禁用SSL证书验证警告（如果Ambari使用自签证书）
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# Ambari配置信息，根据你的集群修改
AMBARI_HOST = "your-ambari-server-host"
AMBARI_PORT = 8080  # 默认HTTP端口是8080，HTTPS一般用8443
USERNAME = "admin"
PASSWORD = "your-ambari-password"
CLUSTER_NAME = "your-cluster-name"
COMPONENT_NAME = "DATANODE"  # 可替换为NAMENODE、RESOURCEMANAGER等

def get_component_status():
    # 构建API请求URL
    api_url = f"http://{AMBARI_HOST}:{AMBARI_PORT}/api/v1/clusters/{CLUSTER_NAME}/components/{COMPONENT_NAME}"
    
    # 发送带身份验证的GET请求
    response = requests.get(
        api_url,
        auth=(USERNAME, PASSWORD),
        verify=False  # HTTPS+自签证书时设为False，否则可移除
    )
    
    # 处理响应结果
    if response.status_code == 200:
        component_data = response.json()
        instances = component_data['host_components']
        print(f"=== {COMPONENT_NAME} 组件状态 ===")
        for instance in instances:
            host_name = instance['HostRoles']['host_name']
            state = instance['HostRoles']['state']
            print(f"主机 {host_name}: {state}")
    else:
        print(f"请求失败，状态码: {response.status_code}")
        print(f"错误信息: {response.text}")

if __name__ == "__main__":
    get_component_status()

扩展：获取所有服务及组件的状态 #

如果需要一次性排查整个集群的服务健康情况，可以用这个脚本批量获取所有服务和组件的状态：

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

AMBARI_HOST = "your-ambari-server-host"
AMBARI_PORT = 8080
USERNAME = "admin"
PASSWORD = "your-ambari-password"
CLUSTER_NAME = "your-cluster-name"

def get_all_service_status():
    api_url = f"http://{AMBARI_HOST}:{AMBARI_PORT}/api/v1/clusters/{CLUSTER_NAME}/services"
    response = requests.get(api_url, auth=(USERNAME, PASSWORD), verify=False)
    
    if response.status_code == 200:
        services_data = response.json()['items']
        for service in services_data:
            service_name = service['ServiceInfo']['service_name']
            service_state = service['ServiceInfo']['state']
            print(f"\n=== 服务 {service_name} 状态: {service_state} ===")
            
            # 获取当前服务下的所有组件
            components_url = f"{api_url}/{service_name}/components"
            components_response = requests.get(components_url, auth=(USERNAME, PASSWORD), verify=False)
            if components_response.status_code == 200:
                components = components_response.json()['items']
                for component in components:
                    component_name = component['ServiceComponentInfo']['component_name']
                    # 获取组件的所有实例状态
                    instances_url = f"{components_url}/{component_name}/host_components"
                    instances_response = requests.get(instances_url, auth=(USERNAME, PASSWORD), verify=False)
                    if instances_response.status_code == 200:
                        instances = instances_response.json()['items']
                        for instance in instances:
                            host = instance['HostRoles']['host_name']
                            state = instance['HostRoles']['state']
                            print(f"  组件 {component_name} - 主机 {host}: {state}")
    else:
        print(f"请求失败，状态码: {response.status_code}")
        print(f"错误信息: {response.text}")

if __name__ == "__main__":
    get_all_service_status()

实用小技巧 #

• 异常过滤：可以在脚本里加判断逻辑，只输出状态不是STARTED的实例，快速定位异常节点。
• 定时监控：结合schedule库设置定时任务（比如每5分钟检查一次），发现异常时触发邮件或企业微信告警。
• 格式化输出：把结果转为JSON格式，方便导入Prometheus、Grafana等监控系统做可视化分析。

使用前记得确认Ambari服务器开启了API访问权限，且用户名密码拥有足够的操作权限（一般admin权限即可）。如果用HTTPS协议，记得把URL里的http改成https，端口换成对应的HTTPS端口（比如8443）。

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