Skip to content

[SPARK-5738] [SQL] Reuse mutable row for each record at jsonStringToRow #4527

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
wants to merge 6 commits into from
Closed

[SPARK-5738] [SQL] Reuse mutable row for each record at jsonStringToRow #4527

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
18e4ddc
[SQL] Reuse mutable row for each record at jsonStringToRow
Feb 11, 2015
2f001ba
Use mutable rows for inner structures
Feb 12, 2015
837785a
Use mutable row arrays for inner arrays
Feb 15, 2015
d97d7db
keep scala style
Feb 15, 2015
2d45c68
fix array reuse issue
Feb 15, 2015
2286ac5
keep scala style
Feb 15, 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
48 changes: 36 additions & 12 deletions sql/core/src/main/scala/org/apache/spark/sql/json/JsonRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,11 @@ private[sql] object JsonRDD extends Logging {
json: RDD[String],
schema: StructType,
columnNameOfCorruptRecords: String): RDD[Row] = {
parseJson(json, columnNameOfCorruptRecords).map(parsed => asRow(parsed, schema))
// Reuse the mutable row for each record and all innner nested structures
parseJson(json, columnNameOfCorruptRecords).mapPartitions {
val row = new GenericMutableRow(schema.fields.length)
iter => iter.map(parsed => asRow(parsed, schema, row))
}
}

private[sql] def inferSchema(
Expand Down Expand Up @@ -401,7 +405,8 @@ private[sql] object JsonRDD extends Logging {
}
}

private[json] def enforceCorrectType(value: Any, desiredType: DataType): Any ={
private[json] def enforceCorrectType(
value: Any, desiredType: DataType, slot: Any = null): Any = {
if (value == null) {
null
} else {
Expand All @@ -414,22 +419,41 @@ private[sql] object JsonRDD extends Logging {
case DecimalType() => toDecimal(value)
case BooleanType => value.asInstanceOf[BooleanType.JvmType]
case NullType => null
case ArrayType(elementType, _) =>
value.asInstanceOf[Seq[Any]].map(enforceCorrectType(_, elementType))
case struct: StructType => asRow(value.asInstanceOf[Map[String, Any]], struct)
case ArrayType(elementType, _) => {
val arrayLength = value.asInstanceOf[Seq[Any]].length
val arraySlot = if (slot != null && slot.asInstanceOf[Seq[Any]].size == arrayLength) {
slot.asInstanceOf[Seq[Any]]
} else {
(new Array[Any](arrayLength)).toSeq
}
value.asInstanceOf[Seq[Any]].zip(arraySlot).map {
case (v, s) => enforceCorrectType(v, elementType,s)
}.toList
}
case struct: StructType =>
asRow(value.asInstanceOf[Map[String, Any]], struct, slot.asInstanceOf[GenericMutableRow])
case DateType => toDate(value)
case TimestampType => toTimestamp(value)
}
}
}

private def asRow(json: Map[String,Any], schema: StructType): Row = {
// TODO: Reuse the row instead of creating a new one for every record.
val row = new GenericMutableRow(schema.fields.length)
schema.fields.zipWithIndex.foreach {
case (StructField(name, dataType, _, _), i) =>
row.update(i, json.get(name).flatMap(v => Option(v)).map(
enforceCorrectType(_, dataType)).orNull)
private def asRow(
json: Map[String,Any],
schema: StructType,
mutable: GenericMutableRow = null): Row = {

val row = if (mutable != null && mutable.length == schema.fields.length) {
mutable
} else {
new GenericMutableRow(schema.fields.length)
}

for(i <- 0 until schema.fields.length) {
val fieldName = schema.fields(i).name
val fieldType = schema.fields(i).dataType
row.update(i, json.get(fieldName).flatMap(v => Option(v)).map(
enforceCorrectType(_, fieldType, row(i))).orNull)
}

row
Expand Down