Spark任务执行报错：HDFS指定Jar文件不存在求助

我通过Bash运行Spark任务，目标是写入数据库表而非HDFS路径。已上传Jar包并检查.conf和shell文件，但执行时抛出如下异常：

22/09/07 21:50:42 INFO yarn.Client: Deleted staging directory hdfs://nn/user/srv_remozo_equip/.sparkStaging/application_1661633254168_93772
Exception in thread "main" java.io.FileNotFoundException: File does not exist: hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/logistica_carga_input_simpledata_2.11-0.1.jar
        at org.apache.hadoop.hdfs.DistributedFileSystem$29.doCall(DistributedFileSystem.java:1749)
        at org.apache.hadoop.hdfs.DistributedFileSystem$29.doCall(DistributedFileSystem.java:1742)
        at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
        at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1757)
        at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:386)

任务核心逻辑代码：

// ================= INICIO LÓGICA DE PROCESO =================
//Se crea df a partir de archivo diario input de acuerdo a la ruta indicada
val df_csv = spark.read.format("csv").option("header","true").option("sep",";").option("mode","dropmalformed").load("/applications/recup_remozo_equipos/equipos_por_recuperar/output/agendamientos_sin_pet_2")

val df_final = df_csv.select($"RutSinDV".as("RUT_SIN_DV"),
    $"dv".as("DV"),
    $"Agendado".as("AGENDADO"),
    to_date(col("Dia_Agendado"), "yyyyMMdd").as("DIA_AGENDADO"),
    $"Horario_Agendado".as("HORARIO_AGENDADO"),
    $"Nombre_Agendamiento".as("NOMBRE_AGENDAMIENTO"),
    $"Telefono_Agendamiento".as("TELEFONO_AGENDAMIENTO"),
    $"Email".substr(0,49).as("EMAIL"),
    $"Region_Agendamiento".substr(0,29).as("REGION_AGENDAMIENTO"),
    $"Comuna_Agendamiento".as("COMUNA_AGENDAMIENTO"),
    $"Direccion_Agendamiento".as("DIRECCION_AGENDAMIENTO"),
    $"Numero_Agendamiento".substr(0,5).as("NUMERO_AGENDAMIENTO"),
    $"Depto_Agendamiento".substr(0,9).as("DEPTO_AGENDAMIENTO"),
    to_timestamp(col("fecha_registro")).as("FECHA_REGISTRO"),
    to_timestamp(col("Fecha_Proceso")).as("FECHA_PROCESO")
  )
// ================== FIN LÓGICA DE PROCESO ==================

// Limpieza en EXADATA
println("[INFO] Se inicia la limpieza por reproceso en EXADATA")

val query_particiones = "(SELECT * FROM (WITH DATA AS (select table_name,partition_name,to_date(trim(''' " +
    "from regexp_substr(extractvalue(dbms_xmlgen.getxmltype('select high_value from all_tab_partitions " +
    "where table_name='''|| table_name|| ''' and table_owner = '''|| table_owner|| ''' and partition_name = '''" +
    "|| partition_name|| ''''),'//text()'),'''.*?''')),'syyyy-mm-dd hh24:mi:ss') high_value_in_date_format " +
    "FROM all_tab_partitions WHERE table_name = '" + table_name + "' AND table_owner = '" + table_owner + "')" +
    "SELECT partition_name FROM DATA WHERE high_value_in_date_format > DATE '" + startDateYear + "-" + startDateMonth + "-" + startDateDay + "' " +
    "AND high_value_in_date_format <= DATE '" + endDateYear + "-" + endDateMonth + "-" + endDateDay + "') A)"

Class.forName(driver_jdbc)
val db = DriverManager.getConnection(url_jdbc, user_jdbc, pass_jdbc)
val st = db.createStatement()

try {
    val consultaParticiones = spark.read.format("jdbc")
      .option("url", url_jdbc)
      .option("driver", driver_jdbc)
      .option("dbTable", query_particiones)
      .option("user", user_jdbc)
      .option("password", pass_jdbc)
      .load()
      .collect()
    for (partition <- consultaParticiones) {
      st.executeUpdate("call " + table_owner + ".DO_THE_TRUNCATE_PARTITION('" + table + "','" + partition.getString(0) + "')")
    }
} catch {
    case e: Exception =>
      println("[ERROR TRUNCATE] " + e)
}

st.close()
db.close()

println("[INFO] Se inicia la inserción en EXADATA")

df_final.filter($"DIA_AGENDADO" >= "2022-08-01")
    .repartition(repartition).write.mode("append")
    .jdbc(url_jdbc, table, utils.jdbcProperties(driver_jdbc, user_jdbc, pass_jdbc))

println("[INFO] Inserción en EXADATA completada con éxito")
println("[INFO] Proceso Logistica Carga Input SimpleData")

异常原因分析 #

该异常和业务代码逻辑无关，核心问题是Spark提交YARN任务时，找不到指定路径下的Jar包：

1. Spark提交任务时，需要将依赖Jar包上传到HDFS的staging目录供YARN集群使用，但配置中指定的Jar包路径hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/logistica_carga_input_simpledata_2.11-0.1.jar不存在。
2. 可能的触发场景：
   • Jar包上传路径与shell脚本/conf配置中的路径不一致
   • Jar包上传中断，导致文件不完整或未成功上传
   • HDFS路径权限不足，执行任务的用户srv_remozo_equip无法访问该路径

解决方法 #

1. 验证Jar包实际路径 #

执行HDFS命令检查Jar包是否存在：

hdfs dfs -ls hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/

若未找到目标Jar包，重新上传：

hdfs dfs -put ./logistica_carga_input_simpledata_2.11-0.1.jar hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/

2. 检查任务提交脚本中的Jar包路径 #

打开Spark提交shell脚本，找到--jar或--class参数对应的Jar包路径，确保与HDFS实际路径完全一致，注意大小写、版本号（如2.11-0.1）的匹配。

3. 确认HDFS路径权限 #

检查执行任务的用户srv_remozo_equip对Jar包所在路径的访问权限：

hdfs dfs -ls -d hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/

若权限不足，修改路径权限：

hdfs dfs -chmod 755 hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/
hdfs dfs -chown srv_remozo_equip:srv_remozo_equip hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/logistica_carga_input_simpledata_2.11-0.1.jar

4. 检查Spark staging目录配置 #

若以上步骤无问题，检查Spark配置中spark.yarn.stagingDir是否设置正确，确保该目录有足够空间和权限，避免任务提交时无法复制Jar包到staging目录。

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