AWS Glue（Spark引擎）任务状态通知与动态重启方案咨询

一、任务成功/失败状态通知的多种方式 #

1. CloudWatch Events + Lambda 触发Kafka事件 #

• 配置CloudWatch Events规则，匹配Glue任务的SUCCEEDED和FAILED状态事件（事件源为aws.glue，事件类型为JobRunStateChange）
• 触发Lambda函数，在函数中调用Kafka Producer API，将任务名称、状态、结束时间、错误信息（失败时）等封装成JSON消息发送到指定Kafka Topic
• 示例Lambda代码（Python）：

from kafka import KafkaProducer
import json

def lambda_handler(event, context):
    job_details = event['detail']
    message = {
        "job_name": job_details['jobName'],
        "status": job_details['state'],
        "run_id": job_details['jobRunId'],
        "timestamp": event['time'],
        "error_message": job_details.get('errorMessage', '')
    }
    # 初始化Kafka生产者
    producer = KafkaProducer(
        bootstrap_servers=['your-kafka-broker-1:9092', 'your-kafka-broker-2:9092'],
        value_serializer=lambda m: json.dumps(m).encode('utf-8')
    )
    producer.send('glue-job-status-topic', value=message)
    producer.flush()
    return {'statusCode': 200}

2. Glue原生SNS通知 + Lambda转Kafka #

• 在Glue任务配置页的「Job notifications」中，勾选成功/失败状态通知，选择已创建的SNS主题作为接收方
• 给该SNS主题添加Lambda订阅，Lambda函数将SNS推送的任务状态消息转换为Kafka兼容格式后发送到指定Topic
• 优势：无需手动配置CloudWatch事件规则，直接复用Glue原生通知能力，减少配置复杂度

3. Spark作业内埋点发送Kafka事件 #

• 在Spark代码中嵌入状态判断逻辑，在作业正常完成或捕获到异常时，主动发送Kafka消息
• 适用于需要自定义通知内容（如包含作业执行指标、数据处理量）的场景
• 示例Spark代码（Scala）：

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import java.util.Properties

// 初始化Kafka生产者配置
val kafkaProps = new Properties()
kafkaProps.put("bootstrap.servers", "your-kafka-broker:9092")
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

val producer = new KafkaProducer[String, String](kafkaProps)
val jobName = spark.conf.get("spark.job.name", "unknown-glue-job")

try {
    // 核心Spark作业逻辑
    val sourceDF = spark.read.parquet("s3://input-path")
    val processedDF = sourceDF.filter("status = 'valid'")
    processedDF.write.parquet("s3://output-path")
    
    // 发送成功状态消息
    val successMsg = s"""{"job_name":"$jobName","status":"SUCCEEDED","record_count":${processedDF.count()},"timestamp":${System.currentTimeMillis()}}"""
    producer.send(new ProducerRecord[String, String]("glue-job-status-topic", successMsg))
} catch {
    case e: Exception =>
        // 发送失败状态消息
        val failMsg = s"""{"job_name":"$jobName","status":"FAILED","error":"${e.getMessage.replaceAll("\"", "\\\"")}","timestamp":${System.currentTimeMillis()}}"""
        producer.send(new ProducerRecord[String, String]("glue-job-status-topic", failMsg))
        throw e // 抛出异常确保Glue任务状态标记为失败
} finally {
    producer.close()
}

二、网络故障导致失败的延迟动态重启方案 #

1. 基于CloudWatch Events + Lambda的动态延迟重启 #

• 实现步骤：
  1. 配置CloudWatch Events规则，捕获Glue任务的FAILED状态事件，并通过事件内容过滤出网络故障（如错误信息包含NetworkError、SocketTimeout、ConnectionRefused等关键词）
  2. 触发Lambda函数，在函数中根据任务的重试次数动态计算延迟时间（例如：第1次延迟5分钟，第2次10分钟，最大延迟不超过30分钟）
  3. 通过CloudWatch Events的定时触发功能，在指定延迟时间后调用Glue API重启任务
• 示例Lambda代码（Python）：

import boto3
import json
from datetime import datetime, timedelta

glue_client = boto3.client('glue')
events_client = boto3.client('events')

def lambda_handler(event, context):
    job_name = event['detail']['jobName']
    job_run_id = event['detail']['jobRunId']
    
    # 获取当前任务的重试次数
    run_info = glue_client.get_job_run(JobName=job_name, RunId=job_run_id)
    attempt_num = run_info['JobRun']['Attempt']
    
    # 动态计算延迟时间（指数退避）
    delay_minutes = min(5 * (2 ** (attempt_num - 1)), 30)
    trigger_time = datetime.utcnow() + timedelta(minutes=delay_minutes)
    
    # 创建一次性定时规则触发重启
    rule_name = f"glue-restart-{job_name}-{job_run_id}"
    events_client.put_rule(
        Name=rule_name,
        ScheduleExpression=f"at({trigger_time.strftime('%Y-%m-%dT%H:%M:%SZ')})",
        State='ENABLED',
        Description=f"Restart glue job {job_name} after {delay_minutes} minutes due to network failure"
    )
    
    # 添加重启任务的Lambda作为目标
    events_client.put_targets(
        Rule=rule_name,
        Targets=[{
            'Id': '1',
            'Arn': context.function_arn,
            'Input': json.dumps({'action': 'restart', 'job_name': job_name})
        }]
    )
    
    # 处理重启动作
    if event.get('action') == 'restart':
        glue_client.start_job_run(JobName=job_name)
        # 删除临时定时规则
        events_client.delete_rule(Name=rule_name)
        events_client.remove_targets(Rule=rule_name, Ids=['1'])
    
    return {'statusCode': 200, 'delay_minutes': delay_minutes, 'next_attempt': attempt_num + 1}

2. Spark作业内网络异常捕获与重试 #

• 在Spark代码中针对网络敏感操作（如读取外部JDBC数据源、调用外部API）捕获网络异常，实现局部逻辑的延迟重试
• 适用于不需要重启整个任务，仅需重试特定阶段的场景
• 示例Scala代码：

import java.net.SocketTimeoutException
import scala.util.control.Breaks._

val maxRetries = 3
val baseDelayMs = 300000 // 5分钟
var retryCount = 0
var success = false

breakable {
    while (retryCount < maxRetries) {
        try {
            // 可能触发网络故障的操作
            val jdbcDF = spark.read
                .format("jdbc")
                .option("url", "jdbc:postgresql://db-host:5432/your-db")
                .option("dbtable", "your-table")
                .load()
            jdbcDF.write.parquet("s3://output-path")
            success = true
            break
        } catch {
            case e: SocketTimeoutException =>
                retryCount += 1
                val delayMs = baseDelayMs * retryCount
                println(s"Network timeout, retrying after ${delayMs/60000} minutes...")
                Thread.sleep(delayMs)
        }
    }
}

if (!success) {
    throw new RuntimeException("Failed after maximum retries due to network issues")
}

3. Glue任务内置重试+自定义错误过滤 #

• 在Glue任务配置中设置「Number of retries」和「Retry delay (seconds)」，同时通过任务的「Job parameters」添加自定义错误过滤逻辑
• 结合CloudWatch日志过滤，仅当故障为网络类时触发重试，避免无意义的重试
• 注意：内置重试的延迟时间是固定值，无法实现动态调整，适合对延迟要求不高的场景

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