1
getRecordWriter 메서드를 덮어 쓰려면 사용자 정의 ParquetOutputFormat (클래스는 org.apache.parquet.hadoop)을 작성했습니다. getRecordWriter 메서드 내에서 IllegalAccessError의 원인이되는 CodecFactory에 액세스합니다. 문제를 해결하기 위해 내 자신의 클래스 로더를 만들려고 시도했지만 도움이되지 않았습니다. 나는 다음과 같은 나는 CustomParquetOutputFormat 사용 된 사용자 정의 클래스 로더를 생성하기 전에 나는이 블로그 게시물 http://techblog.applift.com/upgrading-spark#advanced-case-parquet-writer동적 클래스 로더에 대한 IllegalAccessError
을 따라 :
override def createOutputFormat: OutputFormat[Void, InternalRow] with Ext = new CustomParquetOutputFormat[InternalRow]() with Ext {
...
}
CustomParquetOutputFormat 시도 라인 (274)에 CodecFactory에 액세스 할 때 getRecordWriter를 호출 할 때 문제가 발생합니다
CodecFactory codecFactory = new CodecFactory(conf);
는
(이는 액세스 CustomParquetOutputFormat ParquetOutputFormat 라인 (274)이다)CodecFactory은 package-private입니다.
사용자 정의 클래스 로더 :이
class CustomClassLoader(urls: Array[URL], parent: ClassLoader, whiteList: List[String])
extends ChildFirstURLClassLoader(urls, parent) {
override def loadClass(name: String) = {
if (whiteList.exists(name.startsWith)) {
super.loadClass(name)
} else {
parent.loadClass(name)
}
}
}
사용법 :이
val sc: SparkContext = SparkContext.getOrCreate()
val cl: CustomClassLoader = new CustomClassLoader(sc.jars.map(new URL(_)).toArray,
Thread.currentThread.getContextClassLoader, List(
"org.apache.parquet.hadoop.CustomParquetOutputFormat",
"org.apache.parquet.hadoop.CodecFactory",
"org.apache.parquet.hadoop.ParquetFileWriter",
"org.apache.parquet.hadoop.ParquetRecordWriter",
"org.apache.parquet.hadoop.InternalParquetRecordWriter",
"org.apache.parquet.hadoop.ColumnChunkPageWriteStore",
"org.apache.parquet.hadoop.MemoryManager"
))
cl.loadClass("org.apache.parquet.hadoop.CustomParquetOutputFormat")
.getConstructor(classOf[String], classOf[TaskAttemptContext])
.newInstance(fullPathWithoutExt, taskAttemptContext)
.asInstanceOf[OutputFormat[Void, InternalRow] with ProvidesExtension]
오류 :
java.lang.IllegalAccessError: tried to access class org.apache.parquet.hadoop.CodecFactory from class org.apache.parquet.hadoop.customParquetOutputFormat
at org.apache.parquet.hadoop.CustomParquetOutputFormat.getRecordWriter(CustomParquetOutputFormat.scala:40)
at org.apache.parquet.hadoop.ParquetOutputFormat.getRecordWriter(ParquetOutputFormat.java:262)
at org.apache.spark.custom.hadoop.HadoopWriter.<init>(HadoopWriter.scala:35)
at org.apache.spark.sql.execution.datasources.parquet.ParquetWriter.<init>(ParquetWriter.scala:16)
at org.apache.spark.sql.execution.datasources.parquet.ParquetWriterFactory.createWriter(ParquetWriterFactory.scala:71)
at com.abden.custom.index.IndexBuilder$$anonfun$4.apply(IndexBuilder.scala:55)
at com.abden.custom.index.IndexBuilder$$anonfun$4.apply(IndexBuilder.scala:54)
at scala.collection.immutable.Stream.map(Stream.scala:418)
at com.abden.custom.index.IndexBuilder.generateTiles(IndexBuilder.scala:54)
at com.abden.custom.index.IndexBuilder.generateLayer(IndexBuilder.scala:155)
at com.abden.custom.index.IndexBuilder.appendLayer(IndexBuilder.scala:184)
at com.abden.custom.index.IndexBuilder$$anonfun$appendLayers$1$$anonfun$apply$1.apply(IndexBuilder.scala:213)
at com.abden.custom.index.IndexBuilder$$anonfun$appendLayers$1$$anonfun$apply$1.apply(IndexBuilder.scala:210)
at scala.collection.Iterator$class.foreach(Iterator.scala:742)
at com.abden.custom.util.SplittingByKeyIterator.foreach(SplittingByKeyIterator.scala:3)
at com.abden.custom.index.IndexBuilder$$anonfun$appendLayers$1.apply(IndexBuilder.scala:210)
at com.abden.custom.index.IndexBuilder$$anonfun$appendLayers$1.apply(IndexBuilder.scala:209)
at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$33.apply(RDD.scala:920)
at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$33.apply(RDD.scala:920)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:89)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
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)
오류가 발생 getRecordWriter이 줄에서 :
val codecFactory = new CodecFactory(conf)
CodecFactory 그것이 자사의 패키지로 제한되도록 더 수정이 없습니다. 동일한 클래스 로더에서 모든 클래스를로드하는 동적 클래스 로더를 사용하더라도 여전히 얻을 수 있습니다. IllegalAccessError
그것은 오류 메시지가'customParquetOutputFormat'을 표시하는 것이 이상하다 (소문자 c) 다른 모든 것은'CustomParquetOutputFormat' (대문자 C)를 참조합니다. 그 외에도'super.loadClass (name)'도 부모 로더를 먼저 검사하고 부모가 찾지 못하면 클래스를 로컬에서 해결하려고 시도한다는 것을 알아야합니다. 음, 그리고 다른 클래스 로더에 의해로드 된 클래스는 이름에 관계없이 항상 다른 (런타임) 패키지로 간주됩니다. – Holger
죄송합니다, 오류 메시지가 해결되었습니다. 이 질문에 대한 클래스 이름을 변경하고 우연히 소문자를 사용했습니다. – abden003
안녕하세요, 이전에 갖고 있던 문제를 이해하기 위해 맞춤 클래스 로더 이전에 코드를 공유 할 수 있습니까? 자신의 클래스 로더를 구현하는 것이 여기에 과도한 것 같습니다. – loicmathieu