Airflow通过SSHHook执行EMR集群Spark任务时无法识别错误的处理方案咨询

一、先解决Spark的IndexError错误 #

从你贴的报错栈来看，IndexError: list index out of range出现在x["_c0"].split("::")[1]这一行，这里有两个核心问题需要修复：

1. 数据格式不统一，部分行分割后元素不足 #

有些数据行的_c0字段用::分割后，无法拿到第2个元素（索引1），说明要么这些行没有::分隔符，要么分割后只有1段内容。

你可以先加个调试逻辑排查异常数据：

# 在处理RDD前先打印前10行数据，确认格式问题
sample_rows = movies.take(10)
for row in sample_rows:
    raw_content = row["_c0"]
    split_parts = raw_content.split("::")
    print(f"原始内容: {raw_content} | 分割后长度: {len(split_parts)} | 内容: {split_parts}")

然后修改处理逻辑，过滤掉不符合格式的行，或者给异常行补充默认值：

def safe_process_movie(row):
    split_parts = row["_c0"].split("::")
    # 注意：你要生成3列（movie_id, title, genre），所以至少需要3个分割部分
    if len(split_parts) >= 3:
        return [int(split_parts[0]), split_parts[1], split_parts[2]]
    else:
        # 要么跳过这条异常数据，要么补默认值
        return None

# 过滤掉处理后为None的异常行
cleaned_rdd = movies.rdd.map(safe_process_movie).filter(lambda x: x is not None)
# 转成DataFrame
movies_df = cleaned_rdd.toDF(["movie_id","title","genre"])

2. 修正列数不匹配的低级错误 #

你当前的map操作只返回了2个元素，但toDF却指定了3个列名，这本身就会导致数据结构不匹配！就算没有IndexError，后续也会报错，一定要确保返回的元素数量和列名数量一致。

二、让Airflow感知到SSH执行的任务失败 #

Airflow用SSHHook执行命令时没标记失败，核心原因是没检查命令的退出状态码——Spark任务报错但spark-submit返回的退出码没被Airflow捕获，导致任务被误判为成功。

1. 使用SSHExecuteOperator时开启状态检查 #

如果你用的是SSHExecuteOperator，只需添加check_exit_code=True配置，它会自动检查命令的退出码，非0就标记任务失败：

from airflow.providers.ssh.operators.ssh import SSHExecuteOperator

run_spark_task = SSHExecuteOperator(
    task_id="execute_spark_job",
    ssh_conn_id="emr_master_ssh",  # 替换成你的SSH连接ID
    command="spark-submit /root/movie_data_analysis.py",
    check_exit_code=True,  # 关键配置：开启退出码检查
    do_xcom_push=True,  # 可选：推送命令输出到XCom方便排查问题
)

2. 直接用SSHHook时手动判断退出码 #

如果是自己写PythonOperator调用SSHHook，需要手动获取命令的退出码，非0就抛出Airflow异常：

from airflow.providers.ssh.hooks.ssh import SSHHook
from airflow.exceptions import AirflowFailException
from airflow.decorators import task

@task
def run_spark_job():
    ssh_hook = SSHHook(ssh_conn_id="emr_master_ssh")
    with ssh_hook.get_conn() as ssh_client:
        stdin, stdout, stderr = ssh_client.exec_command("spark-submit /root/movie_data_analysis.py")
        # 获取命令的退出码
        exit_code = stdout.channel.recv_exit_status()
        # 打印日志方便后续排查
        print("命令输出:\n", stdout.read().decode())
        print("错误输出:\n", stderr.read().decode())
        # 非0退出码则标记任务失败
        if exit_code != 0:
            raise AirflowFailException(f"Spark任务执行失败，退出码: {exit_code}")

3. 确保Spark脚本返回正确的退出码 #

最后要确认你的Spark脚本在失败时会返回非0退出码，可以在脚本末尾加个try-except块主动控制：

if __name__ == "__main__":
    try:
        # 你的所有Spark处理逻辑放在这里
        ...
        print("Spark任务执行完成")
        exit(0)  # 成功时返回0
    except Exception as e:
        print(f"Spark任务失败: {str(e)}")
        exit(1)  # 失败时返回非0

这样一套操作下来，Spark的错误能被修复，Airflow也能正确捕获任务失败状态，不会让下游任务继续执行了。

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