You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by li...@apache.org on 2017/11/22 17:21:31 UTC
spark git commit: [SPARK-17920][SPARK-19580][SPARK-19878][SQL]
Backport PR 19779 to branch-2.2 - Support writing to Hive table which uses
Avro schema url 'avro.schema.url'
Repository: spark
Updated Branches:
refs/heads/branch-2.2 df9228b49 -> b17f4063c
[SPARK-17920][SPARK-19580][SPARK-19878][SQL] Backport PR 19779 to branch-2.2 - Support writing to Hive table which uses Avro schema url 'avro.schema.url'
## What changes were proposed in this pull request?
> Backport https://github.com/apache/spark/pull/19779 to branch-2.2
SPARK-19580 Support for avro.schema.url while writing to hive table
SPARK-19878 Add hive configuration when initialize hive serde in InsertIntoHiveTable.scala
SPARK-17920 HiveWriterContainer passes null configuration to serde.initialize, causing NullPointerException in AvroSerde when using avro.schema.url
Support writing to Hive table which uses Avro schema url 'avro.schema.url'
For ex:
create external table avro_in (a string) stored as avro location '/avro-in/' tblproperties ('avro.schema.url'='/avro-schema/avro.avsc');
create external table avro_out (a string) stored as avro location '/avro-out/' tblproperties ('avro.schema.url'='/avro-schema/avro.avsc');
insert overwrite table avro_out select * from avro_in; // fails with java.lang.NullPointerException
WARN AvroSerDe: Encountered exception determining schema. Returning signal schema to indicate problem
java.lang.NullPointerException
at org.apache.hadoop.fs.FileSystem.getDefaultUri(FileSystem.java:182)
at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:174)
## Changes proposed in this fix
Currently 'null' value is passed to serializer, which causes NPE during insert operation, instead pass Hadoop configuration object
## How was this patch tested?
Added new test case in VersionsSuite
Author: vinodkc <vi...@gmail.com>
Closes #19795 from vinodkc/br_Fix_SPARK-17920_branch-2.2.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/b17f4063
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/b17f4063
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/b17f4063
Branch: refs/heads/branch-2.2
Commit: b17f4063cdf52c101ae562ac2a885918acd172ac
Parents: df9228b
Author: vinodkc <vi...@gmail.com>
Authored: Wed Nov 22 09:21:26 2017 -0800
Committer: gatorsmile <ga...@gmail.com>
Committed: Wed Nov 22 09:21:26 2017 -0800
----------------------------------------------------------------------
.../sql/hive/execution/HiveFileFormat.scala | 4 +-
.../spark/sql/hive/client/VersionsSuite.scala | 53 +++++++++++++++++++-
2 files changed, 54 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/b17f4063/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
----------------------------------------------------------------------
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
index ac735e8..4a7cd69 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
@@ -116,7 +116,7 @@ class HiveOutputWriter(
private val serializer = {
val serializer = tableDesc.getDeserializerClass.newInstance().asInstanceOf[Serializer]
- serializer.initialize(null, tableDesc.getProperties)
+ serializer.initialize(jobConf, tableDesc.getProperties)
serializer
}
@@ -130,7 +130,7 @@ class HiveOutputWriter(
private val standardOI = ObjectInspectorUtils
.getStandardObjectInspector(
- tableDesc.getDeserializer.getObjectInspector,
+ tableDesc.getDeserializer(jobConf).getObjectInspector,
ObjectInspectorCopyOption.JAVA)
.asInstanceOf[StructObjectInspector]
http://git-wip-us.apache.org/repos/asf/spark/blob/b17f4063/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala
----------------------------------------------------------------------
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala
index 7aff49c..d48a23f 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.hive.client
-import java.io.{ByteArrayOutputStream, File, PrintStream}
+import java.io.{ByteArrayOutputStream, File, PrintStream, PrintWriter}
import java.net.URI
import org.apache.hadoop.conf.Configuration
@@ -697,6 +697,57 @@ class VersionsSuite extends SparkFunSuite with Logging {
assert(versionSpark.table("t1").collect() === Array(Row(2)))
}
}
+
+ test(s"$version: SPARK-17920: Insert into/overwrite avro table") {
+ withTempDir { dir =>
+ val path = dir.getAbsolutePath
+ val schemaPath = s"""$path${File.separator}avroschemadir"""
+ val destTableName = "tab1"
+
+ new File(schemaPath).mkdir()
+
+ val avroSchema =
+ """{
+ |"type": "record",
+ | "name": "test_Record",
+ | "namespace": "ns.avro",
+ | "fields" : [
+ | {"name": "f1", "type": "string"},
+ | {"name": "f2", "type": ["null", "string"]}
+ | ]
+ |}
+ """.stripMargin
+
+ withTable(destTableName) {
+ val schemaUrl = s"""$schemaPath${File.separator}avroSchema.avsc"""
+ val schemaFile = new File(schemaPath, "avroSchema.avsc")
+ val writer = new PrintWriter(schemaFile)
+ writer.write(avroSchema)
+ writer.close()
+
+ versionSpark.sql(
+ s"""CREATE TABLE $destTableName
+ |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
+ |STORED AS
+ | INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
+ | OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
+ |TBLPROPERTIES ('avro.schema.url' = '$schemaUrl')
+ """.stripMargin
+ )
+ val insertStmt = s"INSERT OVERWRITE TABLE $destTableName SELECT 'ABC', 'DEF'"
+ if (version == "0.12") {
+ // Hive 0.12 throws AnalysisException
+ intercept[AnalysisException](versionSpark.sql(insertStmt))
+ } else {
+ val result = versionSpark.sql("SELECT 'ABC', 'DEF'").collect()
+ versionSpark.sql(insertStmt)
+ assert(versionSpark.table(destTableName).collect() === result)
+ versionSpark.sql(s"INSERT INTO TABLE $destTableName SELECT 'ABC', 'DEF'")
+ assert(versionSpark.table(destTableName).collect() === result ++ result)
+ }
+ }
+ }
+ }
// TODO: add more tests.
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org