[HADOOP] CDH에서 스파크 코어의 의존성 제외

CDH에서 스파크 코어의 의존성 제외

Kafka에서 오는 HBase 데이터에 쓰기 위해 Structured Spark Streaming을 사용하고 있습니다.

내 클러스터 배포는 Hadoop 3.0.0-cdh6.2.0이며 Spark 2.4.0을 사용하고 있습니다.

내 코드는 다음과 같습니다.

val df = spark
 .readStream
 .format("kafka")
 .option("kafka.bootstrap.servers", bootstrapServers)
 .option("subscribe", topic)
 .option("failOnDataLoss", false)
 .load()
 .selectExpr("CAST(key AS STRING)" , "CAST(value AS STRING)")
 .as(Encoders.STRING)

df.writeStream
  .foreachBatch { (batchDF: Dataset[Row], batchId: Long) =>
     batchDF.write
           .options(Map(HBaseTableCatalog.tableCatalog->catalog, HBaseTableCatalog.newTable -> "6"))
          .format("org.apache.spark.sql.execution.datasources.hbase").save()
     }
     .option("checkpointLocation", checkpointDirectory)
     .start()
     .awaitTermination()

HBaseTableCatalog는 json4s-jackson_2.11 라이브러리를 사용합니다. 이 라이브러리는 Spark Core에 포함되어 있지만 버전이 잘못되어 충돌이 발생합니다.

이 문제를 해결하기 위해 스파크 코어에서 json4s-jackson_2.11 라이브러리를 제외하고 pom에 다운 그레이드 된 버전을 추가합니다.

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.4.0-cdh6.2.0</version>
  <exclusions>
    <exclusion>
      <groupId>org.json4s</groupId>
      <artifactId>json4s-jackson_2.11</artifactId>
    </exclusion>
  </exclusions>
</dependency>
<dependency>
  <groupId>org.json4s</groupId>
  <artifactId>json4s-jackson_2.11</artifactId>
  <version>3.2.11</version>
</dependency>

로케일 머신에서 코드를 실행할 때 완벽하게 작동하지만 문제는 cloudera 클러스터에 제출할 때 라이브러리 충돌의 첫 번째 오류가 있다는 것입니다.

Caused by: java.lang.NoSuchMethodError: org.json4s.jackson.JsonMethods$.parse(Lorg/json4s/JsonInput;Z)Lorg/json4s/JsonAST$JValue;
        at org.apache.spark.sql.execution.datasources.hbase.HBaseTableCatalog$.apply(HBaseTableCatalog.scala:257)
        at org.apache.spark.sql.execution.datasources.hbase.HBaseRelation.<init>(HBaseRelation.scala:80)
        at org.apache.spark.sql.execution.datasources.hbase.DefaultSource.createRelation(HBaseRelation.scala:59)
        at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
        at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
        at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
        at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
        at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
        at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
        at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
        at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
        at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
        at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
        at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:668)
        at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:276)
        at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:270)
        at com.App$$anonfun$main$1.apply(App.scala:129)
        at com.App$$anonfun$main$1.apply(App.scala:126)

클러스터에 자체 hadoop 및 spark 라이브러리가 있고이를 사용한다는 것을 알고 있으므로 spark 제출에서 confs spark.driver.userClassPathFirst 및 spark.executor.userClassPathFirst를 true로 설정하지만 다른 오류가 있습니다. 나는 그것을 이해하지 못한다 :

Exception in thread "main" java.lang.ExceptionInInitializerError
        at org.apache.spark.deploy.yarn.YarnSparkHadoopUtil$.<init>(YarnSparkHadoopUtil.scala:48)
        at org.apache.spark.deploy.yarn.YarnSparkHadoopUtil$.<clinit>(YarnSparkHadoopUtil.scala)
        at org.apache.spark.deploy.yarn.Client$$anonfun$1.apply$mcJ$sp(Client.scala:83)
        at org.apache.spark.deploy.yarn.Client$$anonfun$1.apply(Client.scala:83)
        at org.apache.spark.deploy.yarn.Client$$anonfun$1.apply(Client.scala:83)
        at scala.Option.getOrElse(Option.scala:121)
        at org.apache.spark.deploy.yarn.Client.<init>(Client.scala:82)
        at org.apache.spark.deploy.yarn.YarnClusterApplication.start(Client.scala:1603)
        at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:851)
        at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:167)
        at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:195)
        at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)
        at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:926)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:935)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassCastException: org.apache.hadoop.yarn.api.records.impl.pb.PriorityPBImpl cannot be cast to org.apache.hadoop.yarn.api.records.Priority
        at org.apache.hadoop.yarn.api.records.Priority.newInstance(Priority.java:39)
        at org.apache.hadoop.yarn.api.records.Priority.<clinit>(Priority.java:34)
        ... 15 more

마지막으로, 내가 원하는 것은 Spark 코어의 것이 아니라 내 pom에서 json4s-jackson_2.11을 사용하여 Spark를 만드는 것입니다.

해결법

1. ==============================

   1.이 문제를 해결하려면 spark.driver.userClassPathFirst 및 spark.executor.userClassPathFirst를 사용하지 말고 spark.driver.extraClassPath 및 spark.executor.extraClassPath를 사용하십시오.

   이 문제를 해결하려면 spark.driver.userClassPathFirst 및 spark.executor.userClassPathFirst를 사용하지 말고 spark.driver.extraClassPath 및 spark.executor.extraClassPath를 사용하십시오.

   공식 문서에서 정의 : "드라이버의 클래스 경로 앞에 추가 클래스 클래스 항목."

   예 :

   이것은 내 문제를 해결했습니다 (사용하려는 Jackson 버전과 하나의 스파크가 사용 중입니다).

   도움이 되길 바랍니다.

from https://stackoverflow.com/questions/57329060/exclusion-of-dependency-of-spark-core-in-cdh by cc-by-sa and MIT license