从脚本到自动化用Python和Shell封装YARN应用管理实现一键终止与巡检在大规模数据处理场景中YARN集群往往同时运行着数百个应用实例。当某个ETL任务因代码缺陷陷入死循环或是某个临时查询占用了生产环境关键资源时传统的手动操作方式显得力不从心。本文将通过实战案例展示如何构建一个智能化的YARN应用管控系统实现三个核心能力实时状态巡检、精准条件筛选和批量操作执行。1. 构建YARN应用管理的基础工具链1.1 环境准备与API接入YARN ResourceManager的REST API是我们所有自动化操作的基础入口点。首先需要确保操作主机能够访问ResourceManager服务并安装必要的命令行工具# 验证网络连通性 ping resourcemanager.example.com telnet resourcemanager.example.com 8088 # 安装基础工具包 sudo yum install -y curl jq python3-pip # CentOS/RHEL sudo apt-get install -y curl jq python3-pip # Ubuntu/Debian pip install requests pandas提示生产环境建议配置Kerberos认证可通过kinit命令预先获取票据1.2 应用状态查询的原型实现通过组合curl和jq工具可以快速构建应用列表查询功能#!/bin/bash RM_HOSTresourcemanager.example.com API_URLhttp://$RM_HOST:8088/ws/v1/cluster/apps # 获取所有RUNNING状态的应用 curl -s $API_URL | jq .apps.app[] | select(.state RUNNING)对应的Python实现版本更具扩展性import requests def fetch_yarn_apps(stateRUNNING): api_url http://resourcemanager.example.com:8088/ws/v1/cluster/apps params {states: state} if state else None response requests.get(api_url, paramsparams) response.raise_for_status() return response.json().get(apps, {}).get(app, []) if __name__ __main__: running_apps fetch_yarn_apps() print(fFound {len(running_apps)} running applications)2. 高级筛选策略的实现2.1 多维度过滤条件设计在实际运维中我们需要根据多种条件组合筛选目标应用。下表列出了常见的过滤维度及其实现方式筛选维度数据类型示例值实现方法用户字符串etl_user.user etl_user队列字符串production.queue production运行时长整数3600 (秒).elapsedTime 3600000资源占用浮点数50.0 (vcore占比).allocatedVCores 50应用类型枚举SPARK.applicationType SPARK2.2 动态过滤器的Python实现通过封装过滤逻辑可以构建灵活的筛选系统from datetime import datetime, timedelta class AppFilter: staticmethod def by_user(apps, username): return [app for app in apps if app[user] username] staticmethod def by_runtime(apps, hours): threshold hours * 3600 * 1000 # 转为毫秒 return [app for app in apps if app[elapsedTime] threshold] staticmethod def by_resource(apps, min_vcores, min_memory): return [ app for app in apps if app[allocatedVCores] min_vcores and app[allocatedMB] min_memory * 1024 ] # 使用示例 apps fetch_yarn_apps() long_running AppFilter.by_runtime(apps, hours2) overloaded AppFilter.by_resource(apps, min_vcores8, min_memory32)3. 安全终止机制的实现3.1 批量终止的防护措施直接执行批量终止操作存在风险需要建立防护机制二次确认机制显示将被终止的应用详情要求人工确认模拟执行模式仅输出将要执行的操作而不实际调用API操作白名单限制可操作的应用类型和用户范围操作日志记录详细记录每次终止操作的元数据3.2 带防护的终止脚本实现import logging from typing import List, Dict logging.basicConfig( filenameyarn_operations.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) class YarnTerminator: def __init__(self, dry_runFalse): self.dry_run dry_run def safe_kill(self, app_ids: List[str], reason: str ): results [] for app_id in app_ids: if self.dry_run: logging.info(fDRY RUN: Would kill {app_id}) results.append({id: app_id, status: simulated}) continue try: url fhttp://resourcemanager.example.com:8088/ws/v1/cluster/apps/{app_id}/state payload {state: KILLED} headers {Content-Type: application/json} response requests.put(url, jsonpayload, headersheaders) if response.status_code 200: logging.info(fKilled {app_id} - Reason: {reason}) results.append({id: app_id, status: success}) else: logging.error(fFailed to kill {app_id}: {response.text}) results.append({id: app_id, status: failed}) except Exception as e: logging.exception(fError killing {app_id}) results.append({id: app_id, status: error}) return results # 使用示例 terminator YarnTerminator(dry_runTrue) apps_to_kill [app[id] for app in overloaded] terminator.safe_kill(apps_to_kill, reasonResource overcommit)4. 系统集成与自动化调度4.1 与监控系统对接将YARN管理脚本集成到Prometheus监控系统中可以通过Textfile Collector暴露指标#!/bin/bash OUTPUT_FILE/var/lib/node_exporter/textfile_collector/yarn_metrics.prom # 获取运行超时的应用数量 TIMEOUT_APPS$(python3 yarn_manager.py --filter runtime2h --count-only) cat EOF $OUTPUT_FILE # HELP yarn_timeout_apps Number of applications running over 2 hours # TYPE yarn_timeout_apps gauge yarn_timeout_apps $TIMEOUT_APPS EOF4.2 定时巡检的Crontab配置设置定期执行的巡检任务# 每30分钟检查一次长时间运行的应用 */30 * * * * /usr/local/bin/yarn_manager.py --filter runtime4h --notify # 每天凌晨清理测试环境残留应用 0 1 * * * /usr/local/bin/yarn_manager.py --queue test --older-than 24h --kill4.3 完整的CLI工具封装通过argparse模块创建功能完善的命令行工具import argparse def main(): parser argparse.ArgumentParser(descriptionYARN Application Manager) parser.add_argument(--user, helpFilter by user) parser.add_argument(--queue, helpFilter by queue) parser.add_argument(--runtime, helpFilter by runtime (e.g. 2h, 30m)) parser.add_argument(--kill, actionstore_true, helpTerminate matched apps) parser.add_argument(--dry-run, actionstore_true, helpSimulate operations) parser.add_argument(--notify, actionstore_true, helpSend alert notifications) args parser.parse_args() # 应用过滤逻辑 filters {} if args.user: filters[user] args.user if args.queue: filters[queue] args.queue if args.runtime: filters[runtime] parse_runtime(args.runtime) apps fetch_yarn_apps() matched apply_filters(apps, **filters) if args.kill: terminator YarnTerminator(dry_runargs.dry_run) results terminator.safe_kill([app[id] for app in matched]) if args.notify and matched: send_notification(matched) def parse_runtime(time_str): # 实现时间字符串解析逻辑 pass在实际项目中这套系统将运维人员从重复的手动操作中解放出来。通过组合不同的过滤条件可以精确控制操作范围比如仅终止测试环境中运行超过8小时的Spark作业或是清理特定用户提交的异常任务。
从脚本到自动化：用Python和Shell封装YARN应用管理，实现一键终止与巡检
从脚本到自动化用Python和Shell封装YARN应用管理实现一键终止与巡检在大规模数据处理场景中YARN集群往往同时运行着数百个应用实例。当某个ETL任务因代码缺陷陷入死循环或是某个临时查询占用了生产环境关键资源时传统的手动操作方式显得力不从心。本文将通过实战案例展示如何构建一个智能化的YARN应用管控系统实现三个核心能力实时状态巡检、精准条件筛选和批量操作执行。1. 构建YARN应用管理的基础工具链1.1 环境准备与API接入YARN ResourceManager的REST API是我们所有自动化操作的基础入口点。首先需要确保操作主机能够访问ResourceManager服务并安装必要的命令行工具# 验证网络连通性 ping resourcemanager.example.com telnet resourcemanager.example.com 8088 # 安装基础工具包 sudo yum install -y curl jq python3-pip # CentOS/RHEL sudo apt-get install -y curl jq python3-pip # Ubuntu/Debian pip install requests pandas提示生产环境建议配置Kerberos认证可通过kinit命令预先获取票据1.2 应用状态查询的原型实现通过组合curl和jq工具可以快速构建应用列表查询功能#!/bin/bash RM_HOSTresourcemanager.example.com API_URLhttp://$RM_HOST:8088/ws/v1/cluster/apps # 获取所有RUNNING状态的应用 curl -s $API_URL | jq .apps.app[] | select(.state RUNNING)对应的Python实现版本更具扩展性import requests def fetch_yarn_apps(stateRUNNING): api_url http://resourcemanager.example.com:8088/ws/v1/cluster/apps params {states: state} if state else None response requests.get(api_url, paramsparams) response.raise_for_status() return response.json().get(apps, {}).get(app, []) if __name__ __main__: running_apps fetch_yarn_apps() print(fFound {len(running_apps)} running applications)2. 高级筛选策略的实现2.1 多维度过滤条件设计在实际运维中我们需要根据多种条件组合筛选目标应用。下表列出了常见的过滤维度及其实现方式筛选维度数据类型示例值实现方法用户字符串etl_user.user etl_user队列字符串production.queue production运行时长整数3600 (秒).elapsedTime 3600000资源占用浮点数50.0 (vcore占比).allocatedVCores 50应用类型枚举SPARK.applicationType SPARK2.2 动态过滤器的Python实现通过封装过滤逻辑可以构建灵活的筛选系统from datetime import datetime, timedelta class AppFilter: staticmethod def by_user(apps, username): return [app for app in apps if app[user] username] staticmethod def by_runtime(apps, hours): threshold hours * 3600 * 1000 # 转为毫秒 return [app for app in apps if app[elapsedTime] threshold] staticmethod def by_resource(apps, min_vcores, min_memory): return [ app for app in apps if app[allocatedVCores] min_vcores and app[allocatedMB] min_memory * 1024 ] # 使用示例 apps fetch_yarn_apps() long_running AppFilter.by_runtime(apps, hours2) overloaded AppFilter.by_resource(apps, min_vcores8, min_memory32)3. 安全终止机制的实现3.1 批量终止的防护措施直接执行批量终止操作存在风险需要建立防护机制二次确认机制显示将被终止的应用详情要求人工确认模拟执行模式仅输出将要执行的操作而不实际调用API操作白名单限制可操作的应用类型和用户范围操作日志记录详细记录每次终止操作的元数据3.2 带防护的终止脚本实现import logging from typing import List, Dict logging.basicConfig( filenameyarn_operations.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) class YarnTerminator: def __init__(self, dry_runFalse): self.dry_run dry_run def safe_kill(self, app_ids: List[str], reason: str ): results [] for app_id in app_ids: if self.dry_run: logging.info(fDRY RUN: Would kill {app_id}) results.append({id: app_id, status: simulated}) continue try: url fhttp://resourcemanager.example.com:8088/ws/v1/cluster/apps/{app_id}/state payload {state: KILLED} headers {Content-Type: application/json} response requests.put(url, jsonpayload, headersheaders) if response.status_code 200: logging.info(fKilled {app_id} - Reason: {reason}) results.append({id: app_id, status: success}) else: logging.error(fFailed to kill {app_id}: {response.text}) results.append({id: app_id, status: failed}) except Exception as e: logging.exception(fError killing {app_id}) results.append({id: app_id, status: error}) return results # 使用示例 terminator YarnTerminator(dry_runTrue) apps_to_kill [app[id] for app in overloaded] terminator.safe_kill(apps_to_kill, reasonResource overcommit)4. 系统集成与自动化调度4.1 与监控系统对接将YARN管理脚本集成到Prometheus监控系统中可以通过Textfile Collector暴露指标#!/bin/bash OUTPUT_FILE/var/lib/node_exporter/textfile_collector/yarn_metrics.prom # 获取运行超时的应用数量 TIMEOUT_APPS$(python3 yarn_manager.py --filter runtime2h --count-only) cat EOF $OUTPUT_FILE # HELP yarn_timeout_apps Number of applications running over 2 hours # TYPE yarn_timeout_apps gauge yarn_timeout_apps $TIMEOUT_APPS EOF4.2 定时巡检的Crontab配置设置定期执行的巡检任务# 每30分钟检查一次长时间运行的应用 */30 * * * * /usr/local/bin/yarn_manager.py --filter runtime4h --notify # 每天凌晨清理测试环境残留应用 0 1 * * * /usr/local/bin/yarn_manager.py --queue test --older-than 24h --kill4.3 完整的CLI工具封装通过argparse模块创建功能完善的命令行工具import argparse def main(): parser argparse.ArgumentParser(descriptionYARN Application Manager) parser.add_argument(--user, helpFilter by user) parser.add_argument(--queue, helpFilter by queue) parser.add_argument(--runtime, helpFilter by runtime (e.g. 2h, 30m)) parser.add_argument(--kill, actionstore_true, helpTerminate matched apps) parser.add_argument(--dry-run, actionstore_true, helpSimulate operations) parser.add_argument(--notify, actionstore_true, helpSend alert notifications) args parser.parse_args() # 应用过滤逻辑 filters {} if args.user: filters[user] args.user if args.queue: filters[queue] args.queue if args.runtime: filters[runtime] parse_runtime(args.runtime) apps fetch_yarn_apps() matched apply_filters(apps, **filters) if args.kill: terminator YarnTerminator(dry_runargs.dry_run) results terminator.safe_kill([app[id] for app in matched]) if args.notify and matched: send_notification(matched) def parse_runtime(time_str): # 实现时间字符串解析逻辑 pass在实际项目中这套系统将运维人员从重复的手动操作中解放出来。通过组合不同的过滤条件可以精确控制操作范围比如仅终止测试环境中运行超过8小时的Spark作业或是清理特定用户提交的异常任务。
