model_script/script/feature_spark_monitor.py

import argparse
import configparser
import json
from datetime import datetime, timedelta
from typing import Dict, List

from client import YarnClient, ApolloClient
from util import date_util, json_util, feishu_inform_util

yarn_client = YarnClient.YarnClient("192.168.203.16")

table_list = [
    "alg_mid_feature_sharecf",
    "alg_vid_feature_all_share",
    "alg_vid_feature_all_return",
    "alg_vid_feature_share2return",
    "alg_vid_feature_basic_info",
    "alg_recsys_feature_cf_i2i_new",
    "alg_vid_feature_all_exp_v2",
    "alg_vid_feature_exp2share_v2",
    "alg_vid_feature_feed_noflow_exp_v2",
    "alg_vid_feature_feed_noflow_root_share_v2",
    "alg_vid_feature_feed_noflow_root_return_v2",
    "alg_vid_feature_feed_flow_exp_v2",
    "alg_vid_feature_feed_flow_root_share_v2",
    "alg_vid_feature_feed_flow_root_return_v2",
    "alg_vid_feature_feed_province_exp_v2",
    "alg_vid_feature_feed_province_root_share_v2",
    "alg_vid_feature_feed_province_root_return_v2",
    "alg_cid_feature_basic_info",
    "alg_cid_feature_adver_action",
    "alg_cid_feature_cid_action",
    "alg_cid_feature_region_action",
    "alg_cid_feature_app_action",
    "alg_cid_feature_week_action",
    "alg_cid_feature_hour_action",
    "alg_cid_feature_brand_action",
    "alg_cid_feature_weChatVersion_action",
    "alg_cid_feature_vid_cf",
    "alg_cid_feature_vid_cf_rank",
    "alg_mid_feature_ad_action",
    "alg_mid_feature_play",
    "alg_mid_feature_share_and_return",
    "alg_mid_feature_play_tags",
    "alg_mid_feature_return_tags",
    "alg_mid_feature_share_tags",
    "alg_mid_feature_returncf",
    "alg_mid_feature_feed_exp_return_tags_v2",
    "alg_mid_feature_feed_exp_share_tags_v2",
    "alg_vid_global_feature_20250212",
    "alg_vid_recommend_exp_feature_20250212",
    "alg_vid_recommend_flowpool_exp_feature_20250212",
    "alg_vid_apptype_recommend_exp_feature_20250212",
    "alg_vid_province_recommend_exp_feature_20250212",
    "alg_vid_brand_recommend_exp_feature_20250212",
    "alg_vid_hotsencetype_recommend_exp_feature_20250212",
    "scene_type_vid_cf_feature_20250212",
    "vid_click_cf_feature_20250212",
    "alg_recsys_feature_cf_i2i_v2",
    "alg_channel_recommend_exp_feature_20250212",
    "alg_merge_cate1_recommend_exp_feature_20250212",
    "alg_merge_cate2_recommend_exp_feature_20250212",
    "alg_video_unionid_recommend_exp_feature_20250212",
    "mid_merge_cate1_feature_20250212",
    "mid_merge_cate2_feature_20250212",
    "mid_global_feature_20250212",
]

filter_date = datetime(2024, 1, 1)
current_hour = datetime.now().hour


def handle_table(table_name: str, spark_task_list: List[Dict]) -> (bool, str, str):
    filtered_data = [
        item for item in spark_task_list
        if table_name == item['name']
    ]
    if not filtered_data:
        # 如果没有找到，表示近七个小时都没有同步过
        return True, "最近没有四小时同步过数据", "error"

    # 判断最近一次完成时间是否大于两个小时
    filtered_data.sort(key=lambda item: date_util.str_cover_date(item['finishedTime']), reverse=True)
    last_finished_item = filtered_data[0]
    print(f"表: {table_name}, 最后一次完成时间为: {last_finished_item['finishedTime']}")

    finished_time = date_util.str_cover_date(last_finished_item['finishedTime'])
    started_time = date_util.str_cover_date(last_finished_item['startedTime'])

    time_difference = datetime.now() - finished_time
    if time_difference > timedelta(minutes=120):
        return True, f"最近两个小时没有同步完成数据，最近一次完成时间为: {last_finished_item['finishedTime']}", "error"

    # 判断持续时间是否超过一个小时
    elapse = (finished_time - started_time)
    print(f"表: {table_name}, 最后一次任务持续时间为: {date_util.seconds_convert(elapse.seconds)}")

    if elapse > timedelta(minutes=50):
        return True, f"最近一次同步任务持续时间超过50分钟, 持续时间为: {date_util.seconds_convert(elapse.seconds)}", "warn"

    # 判断任务的完成时间是否是当前小时
    finished_hour = finished_time.hour

    print(f"表: {table_name}, 最后一次完成是: {finished_hour} 小时, 当前小时为: {current_hour}")
    if finished_hour != current_hour:
        return True, f"当前小时的任务未完成", "warn"

    return False, ""


def invoke_feishu_card_mgs(webhook: str, content: str, alarm_level="error"):
    card_json = {
        "config": {},
        "i18n_elements": {
            "zh_cn": [
                {
                    "tag": "markdown",
                    "content": "",
                    "text_align": "left",
                    "text_size": "normal"
                },
                {
                    "tag": "markdown",
                    "content": content,
                    "text_align": "left",
                    "text_size": "normal"
                }
            ]
        },
        "i18n_header": {
            "zh_cn": {
                "title": {
                    "tag": "plain_text",
                    "content": "特征同步延迟告警"
                },
                "subtitle": {
                    "tag": "plain_text",
                    "content": ""
                },
                "template": "red" if alarm_level == "error" else "yellow"
            }
        }
    }
    feishu_inform_util.send_card_msg_to_feishu(webhook, card_json)


def send_error_info(table_name: str, warn_reason: str, alarm_level: str, webhook: str):
    mgs_text = f"\n- 大数据表名: {table_name}" \
               f"\n- 告警原因: {warn_reason}" \
               f"\n- 请关注"
    invoke_feishu_card_mgs(webhook, mgs_text, alarm_level)


def print_config(config_path):
    print(f"配置文件路径: {config_path}")
    config = configparser.ConfigParser()
    config.read(config_path)
    for section in config.sections():
        print(f"[{section}]")
        for key, value in config.items(section):
            print(f"{key} = {value}")


def _main():
    parser = argparse.ArgumentParser(description="feature_spark_task_monitor")
    parser.add_argument("-c", "--config", required=False, help="config file path",
                        default="/home/monitor/model_script/config/config.ini")
    args = parser.parse_args()
    print_config(args.config)

    # 读取配置文件
    config = configparser.ConfigParser()
    config.read(args.config)
    webhook_url = config.get("feishu", "model.webhook")
    # webhook_url = 'https://open.feishu.cn/open-apis/bot/v2/hook/540d4098-367a-4068-9a44-b8109652f07c'

    apollo_meta_url = config.get("apollo", "meta")
    apollo = ApolloClient.ApolloClient(apollo_meta_url)

    # 获取最近4小时的Spark任务
    hours_7_early = int((datetime.now() - timedelta(hours=4)).timestamp()) * 1000
    result = yarn_client.get_apps(finished_time_begin=hours_7_early)
    result = [
        {**item, 'name': item['name'].split(":")[1].strip()}
        for item in result
        if item['finalStatus'] == "SUCCEEDED"

    ]

    if len(result) == 0:
        print("未获取到已完成的任务，跳过")
        return

    j = apollo.get_config("recommend-feature", "default", "application")
    dts_config = json_util.remove_comments(j['configurations']['dts.config'])
    table_config_list = json.loads(dts_config)
    for table_config in table_config_list:
        print("\n\n\n")
        table_name = table_config['odps']['table']
        print(f"当前处理的表为: {table_name}")
        partitions = table_config['odps']['partition']
        if len(partitions) != 2:
            # 只处理有两个分区键的小时级表
            print(f"表: {table_name} 的分区键只有一个，暂不处理")
            continue

        b, warn_reason, alarm_level = handle_table(table_name, result)
        if b:
            print(f"表: {table_name}, 触发告警; 告警原因: {warn_reason}")
            send_error_info(table_name, warn_reason, alarm_level, webhook_url)


if __name__ == '__main__':
    try:
        _main()
    except Exception as e:
        print(f"监控告警发生异常: {e}")
        txt = f"\n- 特征同步异常告警" \
              f"\n- 告警原因: 监控脚本执行异常" \
              f"\n- 请关注"
        url = "https://open.feishu.cn/open-apis/bot/v2/hook/540d4098-367a-4068-9a44-b8109652f07c"
        invoke_feishu_card_mgs(url, txt, "error")