Github user HyukjinKwon commented on a diff in the pull request: https://github.com/apache/spark/pull/14102#discussion_r74194642 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JacksonParser.scala --- @@ -35,184 +34,337 @@ import org.apache.spark.util.Utils private[json] class SparkSQLJsonProcessingException(msg: String) extends RuntimeException(msg) -object JacksonParser extends Logging { +class JacksonParser( + schema: StructType, + columnNameOfCorruptRecord: String, + options: JSONOptions) extends Logging { - def parse( - input: RDD[String], - schema: StructType, - columnNameOfCorruptRecords: String, - configOptions: JSONOptions): RDD[InternalRow] = { + import com.fasterxml.jackson.core.JsonToken._ - input.mapPartitions { iter => - parseJson(iter, schema, columnNameOfCorruptRecords, configOptions) + // A `ValueConverter` is responsible for converting a value from `JsonParser` + // to a value in a field for `InternalRow`. + private type ValueConverter = (JsonParser) => Any + + // `ValueConverter`s for the root schema for all fields in the schema + private val rootConverter: ValueConverter = makeRootConverter(schema) + + private val factory = new JsonFactory() + options.setJacksonOptions(factory) + + private def failedConversion( + parser: JsonParser, + dataType: DataType): Any = parser.getCurrentToken match { + case _ if parser.getTextLength < 1 => + // If conversion is failed, this produces `null` rather than + // returning empty string. This will protect the mismatch of types. + null + + case token => + // We cannot parse this token based on the given data type. So, we throw a + // SparkSQLJsonProcessingException and this exception will be caught by + // parseJson method. + throw new SparkSQLJsonProcessingException( + s"Failed to parse a value for data type $dataType (current token: $token).") + } + + private def failedRecord(record: String): Seq[InternalRow] = { + // create a row even if no corrupt record column is present + if (options.failFast) { + throw new RuntimeException(s"Malformed line in FAILFAST mode: $record") + } + if (options.dropMalformed) { + logWarning(s"Dropping malformed line: $record") + Nil + } else { + val row = new GenericMutableRow(schema.length) + for (corruptIndex <- schema.getFieldIndex(columnNameOfCorruptRecord)) { + require(schema(corruptIndex).dataType == StringType) + row.update(corruptIndex, UTF8String.fromString(record)) + } + Seq(row) } } /** - * Parse the current token (and related children) according to a desired schema - * This is a wrapper for the method `convertField()` to handle a row wrapped - * with an array. + * Create a converter which converts the JSON documents held by the `JsonParser` + * to a value according to a desired schema. This is a wrapper for the method + * `makeConverter()` to handle a row wrapped with an array. */ - def convertRootField( - factory: JsonFactory, - parser: JsonParser, - schema: DataType): Any = { - import com.fasterxml.jackson.core.JsonToken._ - (parser.getCurrentToken, schema) match { - case (START_ARRAY, st: StructType) => - // SPARK-3308: support reading top level JSON arrays and take every element - // in such an array as a row - convertArray(factory, parser, st) - - case (START_OBJECT, ArrayType(st, _)) => + def makeRootConverter(dataType: DataType): ValueConverter = dataType match { + case st: StructType => + val elementConverter = makeConverter(st) + val fieldConverters = st.map(_.dataType).map(makeConverter) + (parser: JsonParser) => parser.getCurrentToken match { + case START_OBJECT => convertObject(parser, st, fieldConverters) + // SPARK-3308: support reading top level JSON arrays and take every element + // in such an array as a row + // + // For example, we support, the JSON data as below: + // + // [{"a":"str_a_1"}] + // [{"a":"str_a_2"}, {"b":"str_b_3"}] + // + // resulting in: + // + // List([str_a_1,null]) + // List([str_a_2,null], [null,str_b_3]) + // + case START_ARRAY => convertArray(parser, elementConverter) + case _ => failedConversion(parser, st) + } + + case ArrayType(st: StructType, _) => + val elementConverter = makeConverter(st) + val fieldConverters = st.map(_.dataType).map(makeConverter) + (parser: JsonParser) => parser.getCurrentToken match { // the business end of SPARK-3308: - // when an object is found but an array is requested just wrap it in a list - convertField(factory, parser, st) :: Nil + // when an object is found but an array is requested just wrap it in a list. + // This is being wrapped in `JacksonParser.parse`. + case START_OBJECT => convertObject(parser, st, fieldConverters) + case START_ARRAY => convertArray(parser, elementConverter) + case _ => failedConversion(parser, st) + } - case _ => - convertField(factory, parser, schema) - } + case _ => makeConverter(dataType) } - private def convertField( - factory: JsonFactory, - parser: JsonParser, - schema: DataType): Any = { - import com.fasterxml.jackson.core.JsonToken._ - (parser.getCurrentToken, schema) match { - case (null | VALUE_NULL, _) => - null - - case (FIELD_NAME, _) => - parser.nextToken() - convertField(factory, parser, schema) - - case (VALUE_STRING, StringType) => - UTF8String.fromString(parser.getText) - - case (VALUE_STRING, _) if parser.getTextLength < 1 => - // guard the non string type - null - - case (VALUE_STRING, BinaryType) => - parser.getBinaryValue - - case (VALUE_STRING, DateType) => - val stringValue = parser.getText - if (stringValue.contains("-")) { - // The format of this string will probably be "yyyy-mm-dd". - DateTimeUtils.millisToDays(DateTimeUtils.stringToTime(parser.getText).getTime) - } else { - // In Spark 1.5.0, we store the data as number of days since epoch in string. - // So, we just convert it to Int. - stringValue.toInt + /** + * Create a converter which converts the JSON documents held by the `JsonParser` + * to a value according to a desired schema. + */ + private def makeConverter(dataType: DataType): ValueConverter = dataType match { + case BooleanType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_TRUE => true + case VALUE_FALSE => false + case _ => failedConversion(parser, dataType) } + } - case (VALUE_STRING, TimestampType) => - // This one will lose microseconds parts. - // See https://issues.apache.org/jira/browse/SPARK-10681. - DateTimeUtils.stringToTime(parser.getText).getTime * 1000L + case ByteType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_NUMBER_INT => parser.getByteValue + case _ => failedConversion(parser, dataType) + } + } - case (VALUE_NUMBER_INT, TimestampType) => - parser.getLongValue * 1000000L + case ShortType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_NUMBER_INT => parser.getShortValue + case _ => failedConversion(parser, dataType) + } + } - case (_, StringType) => - val writer = new ByteArrayOutputStream() - Utils.tryWithResource(factory.createGenerator(writer, JsonEncoding.UTF8)) { - generator => generator.copyCurrentStructure(parser) + case IntegerType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_NUMBER_INT => parser.getIntValue + case _ => failedConversion(parser, dataType) } - UTF8String.fromBytes(writer.toByteArray) - - case (VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT, FloatType) => - parser.getFloatValue - - case (VALUE_STRING, FloatType) => - // Special case handling for NaN and Infinity. - val value = parser.getText - val lowerCaseValue = value.toLowerCase() - if (lowerCaseValue.equals("nan") || - lowerCaseValue.equals("infinity") || - lowerCaseValue.equals("-infinity") || - lowerCaseValue.equals("inf") || - lowerCaseValue.equals("-inf")) { - value.toFloat - } else { - throw new SparkSQLJsonProcessingException(s"Cannot parse $value as FloatType.") + } + + case LongType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_NUMBER_INT => parser.getLongValue + case _ => failedConversion(parser, dataType) } + } - case (VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT, DoubleType) => - parser.getDoubleValue - - case (VALUE_STRING, DoubleType) => - // Special case handling for NaN and Infinity. - val value = parser.getText - val lowerCaseValue = value.toLowerCase() - if (lowerCaseValue.equals("nan") || - lowerCaseValue.equals("infinity") || - lowerCaseValue.equals("-infinity") || - lowerCaseValue.equals("inf") || - lowerCaseValue.equals("-inf")) { - value.toDouble - } else { - throw new SparkSQLJsonProcessingException(s"Cannot parse $value as DoubleType.") + case FloatType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT => + parser.getFloatValue + + case VALUE_STRING => + // Special case handling for NaN and Infinity. + val value = parser.getText + val lowerCaseValue = value.toLowerCase + if (lowerCaseValue.equals("nan") || + lowerCaseValue.equals("infinity") || + lowerCaseValue.equals("-infinity") || + lowerCaseValue.equals("inf") || + lowerCaseValue.equals("-inf")) { + value.toFloat + } else { + throw new SparkSQLJsonProcessingException(s"Cannot parse $value as FloatType.") + } + + case _ => failedConversion(parser, dataType) } + } - case (VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT, dt: DecimalType) => - Decimal(parser.getDecimalValue, dt.precision, dt.scale) + case DoubleType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT => + parser.getDoubleValue + + case VALUE_STRING => + // Special case handling for NaN and Infinity. + val value = parser.getText + val lowerCaseValue = value.toLowerCase + if (lowerCaseValue.equals("nan") || + lowerCaseValue.equals("infinity") || + lowerCaseValue.equals("-infinity") || + lowerCaseValue.equals("inf") || + lowerCaseValue.equals("-inf")) { + value.toDouble + } else { + throw new SparkSQLJsonProcessingException(s"Cannot parse $value as DoubleType.") + } - case (VALUE_NUMBER_INT, ByteType) => - parser.getByteValue + case _ => failedConversion(parser, dataType) + } + } - case (VALUE_NUMBER_INT, ShortType) => - parser.getShortValue + case StringType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_STRING => + UTF8String.fromString(parser.getText) - case (VALUE_NUMBER_INT, IntegerType) => - parser.getIntValue + case _ => + val writer = new ByteArrayOutputStream() + Utils.tryWithResource(factory.createGenerator(writer, JsonEncoding.UTF8)) { + generator => generator.copyCurrentStructure(parser) + } + UTF8String.fromBytes(writer.toByteArray) + } + } - case (VALUE_NUMBER_INT, LongType) => - parser.getLongValue + case TimestampType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_STRING => + // This one will lose microseconds parts. + // See https://issues.apache.org/jira/browse/SPARK-10681. + DateTimeUtils.stringToTime(parser.getText).getTime * 1000L - case (VALUE_TRUE, BooleanType) => - true + case VALUE_NUMBER_INT => + parser.getLongValue * 1000000L - case (VALUE_FALSE, BooleanType) => - false + case _ => failedConversion(parser, dataType) + } + } - case (START_OBJECT, st: StructType) => - convertObject(factory, parser, st) + case DateType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_STRING => + val stringValue = parser.getText + if (stringValue.contains("-")) { + // The format of this string will probably be "yyyy-mm-dd". + DateTimeUtils.millisToDays(DateTimeUtils.stringToTime(parser.getText).getTime) + } else { + // In Spark 1.5.0, we store the data as number of days since epoch in string. + // So, we just convert it to Int. + stringValue.toInt + } - case (START_ARRAY, ArrayType(st, _)) => - convertArray(factory, parser, st) + case _ => failedConversion(parser, dataType) + } + } - case (START_OBJECT, MapType(StringType, kt, _)) => - convertMap(factory, parser, kt) + case BinaryType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case VALUE_STRING => parser.getBinaryValue + case _ => failedConversion(parser, dataType) + } + } + + case dt: DecimalType => + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case (VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT) => + Decimal(parser.getDecimalValue, dt.precision, dt.scale) + + case _ => failedConversion(parser, dt) + } + } + + case st: StructType => + val fieldConverters = st.map(_.dataType).map(makeConverter) + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case START_OBJECT => convertObject(parser, st, fieldConverters) + case _ => failedConversion(parser, st) + } + } + + case at: ArrayType => + val elementConverter = makeConverter(at.elementType) + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case START_ARRAY => convertArray(parser, elementConverter) + case _ => failedConversion(parser, at) + } + } + + case mt: MapType => + val valueConverter = makeConverter(mt.valueType) + (parser: JsonParser) => convertField(parser) { + parser.getCurrentToken match { + case START_OBJECT => convertMap(parser, valueConverter) + case _ => failedConversion(parser, mt) + } + } + + case udt: UserDefinedType[_] => + makeConverter(udt.sqlType) + + case _ => + (parser: JsonParser) => + failedConversion(parser, dataType) + } + + /** + * This converts a field. If this is called after `START_OBJECT`, then, the next token can be + * `FIELD_NAME`. Since the names are kept in `JacksonParser.convertObject`, this `FIELD_NAME` + * token can be skipped as below. When this is called after `START_ARRAY`, the tokens become + * ones about values until `END_ARRAY`. In this case, we don't have to skip. + */ + private def convertField(parser: JsonParser)(f: => Any): Any = { --- End diff -- Sure, I will merge both `convertValue` and `convertField` but additional checking should be added for merging https://github.com/apache/spark/pull/14102/files#r74034675 to be called ahead, because in `StringType`, it does not call `failedConversion` when it fails. For `StringType`, it should produce empty string when `parser.getTextLength < 1`. This is also partly because I guess the failed cases are relatively low.
--- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org