[我有一个方法以Parquet形式编写我的一个类Foo
,该类定义为Thrift。
import Foo
import org.apache.spark.rdd.RDD
import org.apache.thrift.TBase
import org.apache.hadoop.mapreduce.Job
import org.apache.parquet.hadoop.ParquetOutputFormat
import org.apache.parquet.hadoop.thrift.ParquetThriftOutputFormat
def writeThriftParquet(rdd: RDD[Foo], outputPath: String): Unit = {
val job = Job.getInstance()
ParquetThriftOutputFormat.setThriftClass(job, classOf[Foo])
ParquetOutputFormat.setWriteSupportClass(job, classOf[Foo])
rdd
.map(x => (null, x))
.saveAsNewAPIHadoopFile(
outputPath,
classOf[Void],
classOf[Foo],
classOf[ParquetThriftOutputFormat[Foo]],
job.getConfiguration)
}
这很好,但是我更愿意编写一个更通用的方法。我尝试了(相对)简单的方法:
def writeThriftParquetGeneral[A <: TBase[_, _]](rdd: RDD[A], outputPath: String): Unit = {
val job = Job.getInstance()
ParquetThriftOutputFormat.setThriftClass(job, classOf[A])
ParquetOutputFormat.setWriteSupportClass(job, classOf[A])
rdd
.map(x => (null, x))
.saveAsNewAPIHadoopFile(
outputPath,
classOf[Void],
classOf[A],
classOf[ParquetThriftOutputFormat[A]],
job.getConfiguration)
}
但是失败,并显示以下错误:
class type required but A found ParquetThriftOutputFormat.setThriftClass(job, classOf[A])
class type required but A found ParquetOutputFormat.setWriteSupportClass(job, classOf[A])
为了解决这个问题,我使用了ClassTag
,但是还没有编译的内容。
import scala.reflect._
implicit val ct = ClassTag[Foo](classOf[Foo])
def writeThriftParquetGeneral[A <: TBase[_, _]](rdd: RDD[A], outputPath: String)(
implicit tag: ClassTag[A]): Unit = {
val job = Job.getInstance()
// The problem line
ParquetThriftOutputFormat.setThriftClass(job, tag.runtimeClass)
// Seems OK from here
ParquetOutputFormat.setWriteSupportClass(job, tag.runtimeClass)
rdd
.map(x => (null, x))
.saveAsNewAPIHadoopFile(
outputPath,
classOf[Void],
tag.runtimeClass,
classOf[ParquetThriftOutputFormat[A]],
job.getConfiguration)
}
此行失败:ParquetThriftOutputFormat.setThriftClass(job, tag.runtimeClass)
[error] found : Class[_$1] where type _$1
[error] required: Class[_ <: org.apache.thrift.TBase[_, _]]
[令我惊讶的是,编译器(Scala 2.11)没有意识到tag.runtimeClass
必须是classOf[A]
,并且A
满足了定义所定义的类型。
[ClassTag#runtimeClass
仅返回一个Class[_]
https://github.com/scala/scala/blob/2.13.x/src/library/scala/reflect/ClassTag.scala#L55
[Class[_ <: TBase[_, _]]
是不同于Class[_]
的存在类型(实际上是其子类型)
implicitly[Class[_ <: TBase[_, _]] <:< Class[_]]
尝试用替换问题行>>
ParquetThriftOutputFormat.setThriftClass(job, classTag.runtimeClass.asSubclass(classOf[TBase[_, _]]))