Apache Spark를 1.5계에 올리면 UnknownHostException이 나오게 되어 버렸을 때의 대처법

경위

최근 Apache Spark를 1.4系에서 1.5.1로 업그레이드했습니다.
그런데, 작업을 만들어 jar로 해 spark-submit 로부터 실행했는데, 아래와 같은 에러가 나와 fail 하게 되어 버렸습니다.

15/10/21 15:22:12 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, spark003.example.com): java.lang.IllegalArgumentException: java.net.UnknownHostException: nameservice1
        at org.apache.hadoop.security.SecurityUtil.buildTokenService(SecurityUtil.java:374)
        at org.apache.hadoop.hdfs.NameNodeProxies.createNonHAProxy(NameNodeProxies.java:312)
        at org.apache.hadoop.hdfs.NameNodeProxies.createProxy(NameNodeProxies.java:178)
        at org.apache.hadoop.hdfs.DFSClient.<init>(DFSClient.java:665)
        at org.apache.hadoop.hdfs.DFSClient.<init>(DFSClient.java:601)
        at org.apache.hadoop.hdfs.DistributedFileSystem.initialize(DistributedFileSystem.java:148)
        at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2596)
        at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91)
        at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2630)
        at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2612)
        at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:370)
        at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:169)
        at org.apache.hadoop.mapred.JobConf.getWorkingDirectory(JobConf.java:656)
        at org.apache.hadoop.mapred.FileInputFormat.setInputPaths(FileInputFormat.java:436)
        at org.apache.hadoop.mapred.FileInputFormat.setInputPaths(FileInputFormat.java:409)
        at org.apache.spark.SparkContext$$anonfun$hadoopFile$1$$anonfun$32.apply(SparkContext.scala:1016)
        at org.apache.spark.SparkContext$$anonfun$hadoopFile$1$$anonfun$32.apply(SparkContext.scala:1016)
        at org.apache.spark.rdd.HadoopRDD$$anonfun$getJobConf$6.apply(HadoopRDD.scala:176)
        at org.apache.spark.rdd.HadoopRDD$$anonfun$getJobConf$6.apply(HadoopRDD.scala:176)
        at scala.Option.map(Option.scala:145)
        at org.apache.spark.rdd.HadoopRDD.getJobConf(HadoopRDD.scala:176)
        at org.apache.spark.rdd.HadoopRDD$$anon$1.<init>(HadoopRDD.scala:220)
        at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:216)
        at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:101)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
        at org.apache.spark.scheduler.Task.run(Task.scala:88)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.UnknownHostException: nameservice1
        ... 41 more

nameservice1는 HDFS의 HA 클러스터에 붙은 논리 서비스 이름입니다.
(내 환경에서는 HDFS HighAvailability를 사용하도록 설정했습니다.)

했던 일

같은 내용의 작업을 Spark1.4系로 실행.
→ 특히 문제없이 작업이 완료.
(작업을 만들 때 Spark 종속성이 1.5.1에서 1.4.1로 변경되었습니다.)

작업과 동일한 내용을 spark-shell에서 실행
→ 문제없이 실행 가능.

HDFS HA 서비스 이름을 확인할 수 없는 것 같기 때문에 HDFS HA 사용할 수 없습니다.
→ 특히 문제없이 작업이 완료.

Spark의 JIRA에 issue를 올린다.
htps : // 일단 s. 아파치. 오 rg/지라/b로 w세/S파 RK-11227
→ HDFS 설정 파일을 검토하자. 해결되었습니다

◇ 실행 명령

/opt/spark/bin/spark-submit \
  --class com.example.Job /jobs/job-assembly-1.0.0.jar

◇ 직업 내용

import org.apache.spark.sql.{SaveMode, SQLContext}

object Job {
  val sparkConfig = ...

  def main( args: Array[String] ): Unit = {
    val sc = new SparkContext( sparkConfig )
    implicit val sqlContext = new SQLContext( sc )
    import sqlContext.implicits._

    val df = sqlContext.read.format( "com.databricks.spark.csv" ).option( "header", "true" ).load( input )
    df.write.format( "json" ).mode( SaveMode.Overwrite ).save( output )
  }
}

CSV 파일을 읽고 내보내기 만하면됩니다

◇ 환경

실행 환경은 다음과 같습니다.

OS ... CentOS6.6

HDFS ... CDH5.4.0 (ClouderaManager로 구축)

Cluster화 ... Zookeeper + Mesos 0.22.0

결론

spark-shell와의 차이점을 찾았는데 HiveContext를 사용하고 있는지 여부의 차이가 있었으므로 시도에 spark-shell에서 implicit val sqlContext = new SQLContext( sc )를 실행해 보았습니다.

그러면 같은 에러가 나오게 되었으므로, 뭐 이것일까라고.

작업 내에서 기본 SQLContext 대신 HiveContext를 사용하면 오류가 해결되었습니다

◇ 변경 후의 작업

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.hive.HiveContext

object Job {
  val sparkConfig = ...

  def main( args: Array[String] ): Unit = {
    val sc = new SparkContext( sparkConfig )
    implicit val sqlContext = new HiveContext( sc )
    import sqlContext.implicits._

    val df = sqlContext.read.format( "com.databricks.spark.csv" ).option( "header", "true" ).load( input )
    df.write.format( "json" ).mode( SaveMode.Overwrite ).save( output )
  }
}

SQLContext 버그가 있다고 생각합니다. . .