Skip to content

Commit

Permalink
Rename useJsonRDD2 to useJacksonStreamingAPI
Browse files Browse the repository at this point in the history
  • Loading branch information
Nathan Howell committed May 4, 2015
1 parent 15c5d1b commit b31917b
Show file tree
Hide file tree
Showing 3 changed files with 8 additions and 8 deletions.
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/SQLConf.scala
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ private[spark] object SQLConf {

val USE_SQL_SERIALIZER2 = "spark.sql.useSerializer2"

val USE_JSONRDD2 = "spark.sql.json.useJsonRDD2"
val USE_JACKSON_STREAMING_API = "spark.sql.json.useJacksonStreamingAPI"

object Deprecated {
val MAPRED_REDUCE_TASKS = "mapred.reduce.tasks"
Expand Down Expand Up @@ -162,7 +162,7 @@ private[sql] class SQLConf extends Serializable {

private[spark] def useSqlSerializer2: Boolean = getConf(USE_SQL_SERIALIZER2, "true").toBoolean

private[spark] def useJsonRDD2: Boolean = getConf(USE_JSONRDD2, "true").toBoolean
private[spark] def useJacksonStreamingAPI: Boolean = getConf(USE_JACKSON_STREAMING_API, "true").toBoolean

/**
* Upper bound on the sizes (in bytes) of the tables qualified for the auto conversion to
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala
Original file line number Diff line number Diff line change
Expand Up @@ -615,7 +615,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
*/
@Experimental
def jsonRDD(json: RDD[String], schema: StructType): DataFrame = {
if (conf.useJsonRDD2) {
if (conf.useJacksonStreamingAPI) {
baseRelationToDataFrame(new JSONRelation(json, None, 1.0, Some(schema))(this))
} else {
val columnNameOfCorruptJsonRecord = conf.columnNameOfCorruptRecord
Expand Down Expand Up @@ -649,7 +649,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
*/
@Experimental
def jsonRDD(json: RDD[String], samplingRatio: Double): DataFrame = {
if (conf.useJsonRDD2) {
if (conf.useJacksonStreamingAPI) {
baseRelationToDataFrame(new JSONRelation(json, None, samplingRatio, None)(this))
} else {
val columnNameOfCorruptJsonRecord = conf.columnNameOfCorruptRecord
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -125,12 +125,12 @@ private[sql] class JSONRelation(
samplingRatio,
userSpecifiedSchema)(sqlContext)

private val useJsonRDD2: Boolean = sqlContext.conf.useJsonRDD2
private val useJacksonStreamingAPI: Boolean = sqlContext.conf.useJacksonStreamingAPI

override val needConversion: Boolean = false

override lazy val schema = userSpecifiedSchema.getOrElse {
if (useJsonRDD2) {
if (useJacksonStreamingAPI) {
JsonRDD2.nullTypeToStringType(
JsonRDD2.inferSchema(
baseRDD,
Expand All @@ -146,7 +146,7 @@ private[sql] class JSONRelation(
}

override def buildScan(): RDD[Row] = {
if (useJsonRDD2) {
if (useJacksonStreamingAPI) {
JsonRDD2.jsonStringToRow(
baseRDD,
schema,
Expand All @@ -160,7 +160,7 @@ private[sql] class JSONRelation(
}

override def buildScan(requiredColumns: Seq[Attribute], filters: Seq[Expression]): RDD[Row] = {
if (useJsonRDD2) {
if (useJacksonStreamingAPI) {
JsonRDD2.jsonStringToRow(
baseRDD,
StructType.fromAttributes(requiredColumns),
Expand Down

0 comments on commit b31917b

Please sign in to comment.