You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by db...@apache.org on 2019/05/21 17:34:36 UTC
[spark] branch master updated: [SPARK-27762][SQL] Support user
provided avro schema for writing fields with different ordering
This is an automated email from the ASF dual-hosted git repository.
dbtsai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 808d9d0 [SPARK-27762][SQL] Support user provided avro schema for writing fields with different ordering
808d9d0 is described below
commit 808d9d05fcf63595bb490ed10a9b8652972fe137
Author: DB Tsai <d_...@apple.com>
AuthorDate: Tue May 21 17:34:19 2019 +0000
[SPARK-27762][SQL] Support user provided avro schema for writing fields with different ordering
## What changes were proposed in this pull request?
Spark Avro reader supports reading avro files with provided schema with different field orderings. However, the avro writer doesn't support this feature. This PR enables the Spark avro writer with this feature.
## How was this patch tested?
New test is added.
Closes #24635 from dbtsai/avroFix.
Lead-authored-by: DB Tsai <d_...@apple.com>
Co-authored-by: Brian Lindblom <bl...@apple.com>
Signed-off-by: DB Tsai <d_...@apple.com>
---
.../org/apache/spark/sql/avro/AvroSerializer.scala | 22 ++++++---
.../org/apache/spark/sql/avro/AvroSuite.scala | 55 ++++++++++++++++++++--
2 files changed, 67 insertions(+), 10 deletions(-)
diff --git a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
index e902b4c..8f895c8 100644
--- a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
+++ b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
@@ -205,18 +205,28 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
throw new IncompatibleSchemaException(s"Cannot convert Catalyst type $catalystStruct to " +
s"Avro type $avroStruct.")
}
- val fieldConverters = catalystStruct.zip(avroStruct.getFields.asScala).map {
- case (f1, f2) => newConverter(f1.dataType, resolveNullableType(f2.schema(), f1.nullable))
- }
+
+ val (avroIndices: Array[Int], fieldConverters: Array[Converter]) =
+ catalystStruct.map { catalystField =>
+ val avroField = avroStruct.getField(catalystField.name)
+ if (avroField == null) {
+ throw new IncompatibleSchemaException(
+ s"Cannot convert Catalyst type $catalystStruct to Avro type $avroStruct.")
+ }
+ val converter = newConverter(catalystField.dataType, resolveNullableType(
+ avroField.schema(), catalystField.nullable))
+ (avroField.pos(), converter)
+ }.toArray.unzip
+
val numFields = catalystStruct.length
- (row: InternalRow) =>
+ row: InternalRow =>
val result = new Record(avroStruct)
var i = 0
while (i < numFields) {
if (row.isNullAt(i)) {
- result.put(i, null)
+ result.put(avroIndices(i), null)
} else {
- result.put(i, fieldConverters(i).apply(row, i))
+ result.put(avroIndices(i), fieldConverters(i).apply(row, i))
}
i += 1
}
diff --git a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
index 81a5cb7..57e4f63 100644
--- a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
+++ b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
@@ -692,9 +692,9 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
| "type" : "record",
| "name" : "test_schema",
| "fields" : [{
- | "name": "enum",
+ | "name": "Suit",
| "type": [{ "type": "enum",
- | "name": "Suit",
+ | "name": "SuitEnumType",
| "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
| }, "null"]
| }]
@@ -734,9 +734,9 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
| "type" : "record",
| "name" : "test_schema",
| "fields" : [{
- | "name": "enum",
+ | "name": "Suit",
| "type": { "type": "enum",
- | "name": "Suit",
+ | "name": "SuitEnumType",
| "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
| }
| }]
@@ -883,6 +883,53 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
}
}
+ test("support user provided avro schema for writing / reading fields with different ordering") {
+ withTempPath { tempDir =>
+ val avroSchema =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [
+ | {"name": "Age", "type": "int"},
+ | {"name": "Name", "type": "string"}
+ | ]
+ |}
+ """.stripMargin
+
+ val avroSchemaReversed =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [
+ | {"name": "Name", "type": "string"},
+ | {"name": "Age", "type": "int"}
+ | ]
+ |}
+ """.stripMargin
+
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(Row(2, "Aurora"))),
+ StructType(Seq(
+ StructField("Age", IntegerType, false),
+ StructField("Name", StringType, false))))
+
+ val tempSaveDir = s"$tempDir/save/"
+
+ // Writing avro file with reversed field ordering
+ df.write.format("avro").option("avroSchema", avroSchemaReversed).save(tempSaveDir)
+
+ // Reading reversed avro file
+ checkAnswer(df.select("Name", "Age"), spark.read.format("avro").load(tempSaveDir))
+ checkAvroSchemaEquals(avroSchemaReversed, getAvroSchemaStringFromFiles(tempSaveDir))
+
+ // Reading reversed avro file with provided original schema
+ val avroDf = spark.read.format("avro").option("avroSchema", avroSchema).load(tempSaveDir)
+ checkAnswer(df, avroDf)
+ assert(avroDf.schema.fieldNames.sameElements(Array("Age", "Name")))
+ }
+ }
+
test("error handling for unsupported Interval data types") {
withTempDir { dir =>
val tempDir = new File(dir, "files").getCanonicalPath
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org