“用户未初始化火花上下文”在 SPARK YARN 集群模式下使用 Scala 代码时出错

Question

为了缩小问题的范围，我删除了其他类依赖项，并且我有这个干净的代码：

object LoaderProcessor extends App {

val logger = LoggerFactory.getLogger(this.getClass())
execute()

def execute(): Unit = {

val spark = get_spark()
import spark.implicits._

var df = spark.read
  .format("csv")
  .option("delimiter", ",")
  .option("header", true)
  .option("inferSchema", "true")
  .option("timestampFormat", "yyyy/MM/dd HH:mm:ss")
  .load(args(2))

df = df.withColumn("zs_source", lit(1)) //the only operation on dataframe

val o_file = Config().getString("myapp.dataFolder") + "/8/1/data.csv"
logger.info("Writing output to: {}", o_file)

df.write.mode("overwrite")
.option("header", "true").csv(o_file)

}

def get_spark(): SparkSession = {
val env = System.getenv("MYAPP_ENV")
var spark:SparkSession = null
if (env == null || env == "dev_local") {
  spark = org.apache.spark.sql.SparkSession.builder
    .master("local")
    .appName("MyApp")
    .getOrCreate;
}else{
  spark = org.apache.spark.sql.SparkSession.builder
    .appName("MyApp")
    //.enableHiveSupport()
    .getOrCreate;
}
spark.sparkContext.setCheckpointDir(Config().getString("myapp.rddcp"))
return spark
}
}

它在客户端模式下运行良好。无法弄清楚问题所在。我的集群在 HDInsight 上。

还注意到“写入”操作一直在这样的输出文件夹上写入：

part-00000-3e9566ae-c13c-468a-8732-e7b8a8df5335-c000.csv

---

然后几秒钟后：

part-00000-4f4979a0-d9f9-481b-aac4-115e63b9f59c-c000.csv

---

8/12/01 15:08:53 INFO ApplicationMaster：在单独的线程中启动用户应用程序 18/12/01 15:08:53 INFO ApplicationMaster：等待火花上下文初始化... 2001 年 18 月 12 日 15:08:55 信息配置 $：环境：开发 18/12/01 15:08:55 错误 ApplicationMaster：未捕获的异常： java.lang.IllegalStateException：用户没有初始化火花上下文！ 在 org.apache.spark.deploy.yarn.ApplicationMaster.runDriver(ApplicationMaster.scala:510) 在 org.apache.spark.deploy.yarn.ApplicationMaster.org$apache$spark$deploy$yarn$ApplicationMaster$$runImpl(ApplicationMaster.scala:345) 在 org.apache.spark.deploy.yarn.ApplicationMaster$$anonfun$run$2.apply$mcV$sp(ApplicationMaster.scala:260) 在 org.apache.spark.deploy.yarn.ApplicationMaster$$anonfun$run$2.apply(ApplicationMaster.scala:260) 在 org.apache.spark.deploy.yarn.ApplicationMaster$$anonfun$run$2.apply(ApplicationMaster.scala:260) 在 org.apache.spark.deploy.yarn.ApplicationMaster$$anon$5.run(ApplicationMaster.scala:815) 在 java.security.AccessController.doPrivileged（本机方法） 在 javax.security.auth.Subject.doAs(Subject.java:422) 在 org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1869) 在 org.apache.spark.deploy.yarn.ApplicationMaster.doAsUser(ApplicationMaster.scala:814) 在 org.apache.spark.deploy.yarn.ApplicationMaster.run(ApplicationMaster.scala:259) 在 org.apache.spark.deploy.yarn.ApplicationMaster$.main(ApplicationMaster.scala:839) 在 org.apache.spark.deploy.yarn.ApplicationMaster.main(ApplicationMaster.scala)

spark-submit --master yarn --deploy-mode cluster --jars "wasb://xx@yy/zs/jars/config-1.3.1.jar" --class myapp.LoaderProcessor "wasb://xx@yy/zs/jars/myapp.jar" l 8 /data/8_data.csv 1 , true false -->问题

spark-submit --deploy-mode client --jars "wasb://xx@yy/zs/jars/config-1.3.1.jar" --class myapp.LoaderProcessor "wasb://xx@yy/zs/jars/myapp.jar" l 8 /data/8_data.csv 1 , true false -->工作！！！

Answer 1

编辑：根据我们在 cmets 中的交换更新

问题是您总是使用if (env == null || env == "dev_local") 创建本地上下文（MYAPP_ENV 在分布式环境中为空）