Skip to content

[SPARK-6505][SQL]Remove the reflection call in HiveFunctionWrapper #5660

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
baishuo wants to merge 4 commits into from
Closed

[SPARK-6505][SQL]Remove the reflection call in HiveFunctionWrapper #5660

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a5ff9c7
Remove the reflection call in HiveFunctionWrapper
baishuo Apr 23, 2015
0b522a7
modify code style
baishuo Apr 23, 2015
78d9fa3
modify code style
baishuo Apr 24, 2015
ae61ec4
modify code style
baishuo Apr 24, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 22 additions & 22 deletions sql/hive/v0.13.1/src/main/scala/org/apache/spark/sql/hive/Shim13.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,15 @@ package org.apache.spark.sql.hive

import java.rmi.server.UID
import java.util.{Properties, ArrayList => JArrayList}
import java.io.{OutputStream, InputStream}

import scala.collection.JavaConversions._
import scala.language.implicitConversions
import scala.reflect.ClassTag

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.Input
import com.esotericsoftware.kryo.io.Output
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.hive.common.StatsSetupConst
Expand All @@ -46,6 +50,7 @@ import org.apache.hadoop.{io => hadoopIo}

import org.apache.spark.Logging
import org.apache.spark.sql.types.{Decimal, DecimalType, UTF8String}
import org.apache.spark.util.Utils._

/**
* This class provides the UDF creation and also the UDF instance serialization and
Expand All @@ -61,39 +66,34 @@ private[hive] case class HiveFunctionWrapper(var functionClassName: String)
// for Serialization
def this() = this(null)

import org.apache.spark.util.Utils._

@transient
private val methodDeSerialize = {
val method = classOf[Utilities].getDeclaredMethod(
"deserializeObjectByKryo",
classOf[Kryo],
classOf[java.io.InputStream],
classOf[Class[_]])
method.setAccessible(true)

method
def deserializeObjectByKryo[T: ClassTag](
kryo: Kryo,
in: InputStream,
clazz: Class[_]): T = {
val inp = new Input(in)
val t: T = kryo.readObject(inp,clazz).asInstanceOf[T]
inp.close()
t
}

@transient
private val methodSerialize = {
val method = classOf[Utilities].getDeclaredMethod(
"serializeObjectByKryo",
classOf[Kryo],
classOf[Object],
classOf[java.io.OutputStream])
method.setAccessible(true)

method
def serializeObjectByKryo(
kryo: Kryo,
plan: Object,
out: OutputStream ) {
val output: Output = new Output(out)
kryo.writeObject(output, plan)
output.close()
}

def deserializePlan[UDFType](is: java.io.InputStream, clazz: Class[_]): UDFType = {
methodDeSerialize.invoke(null, Utilities.runtimeSerializationKryo.get(), is, clazz)
deserializeObjectByKryo(Utilities.runtimeSerializationKryo.get(), is, clazz)
.asInstanceOf[UDFType]
}

def serializePlan(function: AnyRef, out: java.io.OutputStream): Unit = {
methodSerialize.invoke(null, Utilities.runtimeSerializationKryo.get(), function, out)
serializeObjectByKryo(Utilities.runtimeSerializationKryo.get(), function, out)
}

private var instance: AnyRef = null
Expand Down