[SPARK-24811][SQL]Avro: add new function from_avro and to_avro #21774

gengliangwang · 2018-07-15T18:54:21Z

What changes were proposed in this pull request?

Add a new function from_avro for parsing a binary column of avro format and converting it into its corresponding catalyst value.

Add a new function to_avro for converting a column into binary of avro format with the specified schema.

How was this patch tested?

Unit test

SparkQA · 2018-07-15T19:13:26Z

Test build #93032 has finished for PR 21774 at commit fd7ad1d.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class AvroDataToCatalyst(child: Expression, avroType: SerializableSchema)
case class CatalystDataToAvro(child: Expression) extends UnaryExpression with CodegenFallback

gengliangwang · 2018-07-18T13:06:43Z

This is ready for review. @cloud-fan

SparkQA · 2018-07-18T13:55:25Z

Test build #93230 has finished for PR 21774 at commit 204a59d.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class AvroFunctionsSuite extends QueryTest with SharedSQLContext

mn-mikke

Are these functions supposed to be used also from textual SQL, Python, etc.?

mn-mikke · 2018-07-18T15:06:09Z

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala

+import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes, Expression, UnaryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.types.{AbstractDataType, BinaryType, DataType}
+


I would add ExpressionDescription and javadoc here.

This is not a function expression like the ones in SQL core, so ExpressionDescription can't apply here. I think we can leave it for now.

mn-mikke · 2018-07-18T15:08:39Z

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala

+
+  override def inputTypes: Seq[AbstractDataType] = Seq(BinaryType)
+
+  override lazy val dataType: DataType =


@transient?

the dataType is needed in executor side to build AvroDeserializer, it's better to serialize it instead of recomputing it at executor side.

Since the schema could be quite complex, I am wondering about sending the same schema to executors twice (dateType, avroType). But yeah, transmission of an extra payload and deserialization of dataType might be faster then the schema conversion.

mn-mikke · 2018-07-18T15:13:50Z

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala

+import org.apache.spark.sql.types.{AbstractDataType, BinaryType, DataType}
+
+case class AvroDataToCatalyst(child: Expression, avroType: SerializableSchema)
+  extends UnaryExpression with CodegenFallback with ExpectsInputTypes {


see the comment about CodegenFallback

good point. Since the implementation is short, I think it should be easy to codegen it.

mn-mikke · 2018-07-18T15:22:20Z

external/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala

+
+case class CatalystDataToAvro(child: Expression) extends UnaryExpression with CodegenFallback {
+
+  override lazy val dataType: DataType = BinaryType


just def?

mn-mikke · 2018-07-18T15:23:45Z

external/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala

+import org.apache.spark.sql.catalyst.expressions.{Expression, UnaryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.types.{BinaryType, DataType}
+


ExpressionDescription and javadoc?

mn-mikke · 2018-07-18T15:26:50Z

external/avro/src/main/scala/org/apache/spark/sql/avro/package.scala

@@ -36,4 +40,27 @@ package object avro {
    @scala.annotation.varargs
    def avro(sources: String*): DataFrame = reader.format("avro").load(sources: _*)
  }
+


Why these two functions are not a part of org.apache.spark.sql.functions?

because avro data source is an external package like kafka data source. It's not available in org.apache.spark.sql.functions

Yes, having a function depending on external package in org.apache.spark.sql.functions would be weird.

Thanks guys for your explanation!

cloud-fan · 2018-07-19T05:45:02Z

external/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala

+       """.stripMargin
+
+    // When read float data as double, avro reader fails(trying to read 8 bytes while the data have
+    // only 4 bytes), `AvroDataToCatalyst` catches the exception and returns null.


remove AvroDataToCatalyst catches the exception and returns null.

cloud-fan · 2018-07-19T05:47:39Z

Since Spark doesn't have a persistent UDF API like Hive UDF, I think this is the best we can do now. In the future we should migrate this to UDF API so that we can register it with a name and use it in SQL.

gengliangwang · 2018-07-19T16:42:31Z

external/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala

+
+  override def sql: String = simpleString
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {


I tried complicated code like

nullSafeCodeGen(ctx, ev, binary => { val avroReader = ctx.addReferenceObj("reader", reader) val avroDeserializer = ctx.addReferenceObj("deserializer", deserializer) val avroDecoder = ctx.addMutableState(classOf[BinaryDecoder].getName, "avroDecoder") val catalystResult = ctx.addMutableState(classOf[Any].getName, "catalystResult") val decoderFactory = classOf[DecoderFactory].getName s""" $avroDecoder = $decoderFactory.get().binaryDecoder($binary, $avroDecoder); try { $catalystResult = $avroReader.read($catalystResult, $avroDecoder); } catch (java.io.IOException e) { org.apache.spark.unsafe.Platform.throwException(e); } ${ev.value} = (${CodeGenerator.boxedType(dataType)}) $avroDeserializer.deserialize($catalystResult); """ })

But eventually find it hard to avoid serializing some objects.

cloud-fan · 2018-07-19T17:14:04Z

external/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala

+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val expr = ctx.addReferenceObj("this", this)
+    defineCodeGen(ctx, ev, input =>
+      s"(${CodeGenerator.boxedType(dataType)})$expr.nullSafeEval($input)")


we know output is binary type, we can just write (byte[])

cloud-fan · 2018-07-19T17:15:14Z

external/avro/src/test/scala/org/apache/spark/sql/avro/AvroFunctionsSuite.scala

+    val avroTypeLong = new Schema.Parser().parse(
+      s"""
+         |{
+         |  "type": "long",


the test name says int and string?

cloud-fan · 2018-07-19T17:17:08Z

LGTM

SparkQA · 2018-07-19T20:47:49Z

Test build #93288 has finished for PR 21774 at commit 833dccc.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class CatalystDataToAvro(child: Expression) extends UnaryExpression

SparkQA · 2018-07-19T22:48:34Z

Test build #93292 has finished for PR 21774 at commit a6dd08f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2018-07-20T02:11:47Z

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala

+  }
+
+  override def simpleString: String = {
+    s"from_avro(${child.sql}, ${dataType.simpleString})"


Shall we use catalogString for datatype?

IIRC simpleString will be used in the plan string and should not be too long.

but this is being used in sql though. Do we prefer truncated string form in sql too?

It's not used in sql, we can override sql here and use the untruncated version.

HyukjinKwon · 2018-07-20T02:12:32Z

external/avro/src/main/scala/org/apache/spark/sql/avro/CatalystDataToAvro.scala

+  }
+
+  override def simpleString: String = {
+    s"to_avro(${child.sql}, ${child.dataType.simpleString})"


ditto for catalogString

HyukjinKwon · 2018-07-20T02:14:59Z

external/avro/src/main/scala/org/apache/spark/sql/avro/package.scala

+   * @param avroType the avro type.
+   */
+  @Experimental
+  def from_avro(data: Column, avroType: Schema): Column = {


Why don't we just expose String version (json) in case we expose it in SQL syntax (also R or Python maybe)? I am sure we won't want to have many variants in the future as well.

Column can support arbitrary expressions. Personally I don't like these string-based overloads...

Since it's in an external module, I think we can only add python/R/SQL version when we have a persistent UDF API.

Yup, I was just worried if we happened to have many overloaded version of this again like json function family. We happened to have many overloaded versions of it, which seems an actual concern (just roughly given my few interaction with other guys like Reynold).

ah sorry i thought you are talking about the data parameter.

Yes, for avroType parameter, we should have a string version

I am actually +1 with @HyukjinKwon .
As the Avro schema require serialization, I decide to make the API accepts JSON format string only. And then parse the string schema in executor side.

HyukjinKwon · 2018-07-20T02:19:32Z

external/avro/src/main/scala/org/apache/spark/sql/avro/package.scala

+   * Converts a binary column of avro format into its corresponding catalyst value. The specified
+   * schema must match the read data, otherwise the behavior is undefined: it may fail or return
+   * arbitrary result.
+   *


Shall we add @since?

HyukjinKwon

LGTM too otherwise

gengliangwang · 2018-07-20T08:20:31Z

I will create another separate PR to totally remove SerializableSchema.

SparkQA · 2018-07-20T11:44:45Z

Test build #93331 has finished for PR 21774 at commit 7179e85.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-07-20T11:50:19Z

Test build #93328 has finished for PR 21774 at commit 0a6dd0b.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class AvroDataToCatalyst(child: Expression, jsonFormatSchema: String)

cloud-fan · 2018-07-20T12:10:53Z

retest this please

SparkQA · 2018-07-20T16:16:04Z

Test build #93345 has finished for PR 21774 at commit 7179e85.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2018-07-20T16:19:13Z

Thanks! Merged to master.

gatorsmile · 2018-07-20T19:54:00Z

Need to revert this PR since it breaks the build. https://amplab.cs.berkeley.edu/jenkins/view/Spark%20QA%20Compile/job/spark-master-compile-maven-hadoop-2.6/7902/

## What changes were proposed in this pull request? 1. Add a new function from_avro for parsing a binary column of avro format and converting it into its corresponding catalyst value. 2. Add a new function to_avro for converting a column into binary of avro format with the specified schema. I created #21774 for this, but it failed the build https://amplab.cs.berkeley.edu/jenkins/view/Spark%20QA%20Compile/job/spark-master-compile-maven-hadoop-2.6/7902/ Additional changes In this PR: 1. Add `scalacheck` dependency in pom.xml to resolve the failure. 2. Update the `log4j.properties` to make it consistent with other modules. ## How was this patch tested? Unit test Compile with different commands: ``` ./build/mvn --force -DzincPort=3643 -DskipTests -Phadoop-2.6 -Phive-thriftserver -Pkinesis-asl -Pspark-ganglia-lgpl -Pmesos -Pyarn compile test-compile ./build/mvn --force -DzincPort=3643 -DskipTests -Phadoop-2.7 -Phive-thriftserver -Pkinesis-asl -Pspark-ganglia-lgpl -Pmesos -Pyarn compile test-compile ./build/mvn --force -DzincPort=3643 -DskipTests -Phadoop-3.1 -Phive-thriftserver -Pkinesis-asl -Pspark-ganglia-lgpl -Pmesos -Pyarn compile test-compile ``` Author: Gengliang Wang <[email protected]> Closes #21838 from gengliangwang/from_and_to_avro.

1. Add a new function from_avro for parsing a binary column of avro format and converting it into its corresponding catalyst value. 2. Add a new function to_avro for converting a column into binary of avro format with the specified schema. I created apache#21774 for this, but it failed the build https://amplab.cs.berkeley.edu/jenkins/view/Spark%20QA%20Compile/job/spark-master-compile-maven-hadoop-2.6/7902/ Additional changes In this PR: 1. Add `scalacheck` dependency in pom.xml to resolve the failure. 2. Update the `log4j.properties` to make it consistent with other modules. Unit test Compile with different commands: ``` ./build/mvn --force -DzincPort=3643 -DskipTests -Phadoop-2.6 -Phive-thriftserver -Pkinesis-asl -Pspark-ganglia-lgpl -Pmesos -Pyarn compile test-compile ./build/mvn --force -DzincPort=3643 -DskipTests -Phadoop-2.7 -Phive-thriftserver -Pkinesis-asl -Pspark-ganglia-lgpl -Pmesos -Pyarn compile test-compile ./build/mvn --force -DzincPort=3643 -DskipTests -Phadoop-3.1 -Phive-thriftserver -Pkinesis-asl -Pspark-ganglia-lgpl -Pmesos -Pyarn compile test-compile ``` Author: Gengliang Wang <[email protected]> Closes apache#21838 from gengliangwang/from_and_to_avro. (cherry picked from commit 8817c68)

gengliangwang added 3 commits July 16, 2018 21:17

add from_avro and to_avro

2b78cff

add test suite

9c185d5

add AvroFunctionsSuite

204a59d

gengliangwang force-pushed the from_and_to_avro branch from fd7ad1d to 204a59d Compare July 18, 2018 09:50

gengliangwang changed the title ~~[WIP][SPARK-24811][SQL]Avro: add new function from_avro and to_avro~~ [SPARK-24811][SQL]Avro: add new function from_avro and to_avro Jul 18, 2018

mn-mikke reviewed Jul 18, 2018

View reviewed changes

cloud-fan reviewed Jul 19, 2018

View reviewed changes

gengliangwang added 2 commits July 19, 2018 15:28

address some comments

fb8e50f

codegen

833dccc

gengliangwang commented Jul 19, 2018

View reviewed changes

cloud-fan reviewed Jul 19, 2018

View reviewed changes

address comments

a6dd08f

HyukjinKwon reviewed Jul 20, 2018

View reviewed changes

HyukjinKwon approved these changes Jul 20, 2018

View reviewed changes

address comments

0a6dd0b

override sql

7179e85

asfgit closed this in 244bcff Jul 20, 2018

gengliangwang mentioned this pull request Jul 22, 2018

[SPARK-24811][SQL]Avro: add new function from_avro and to_avro #21838

Closed


		override def inputTypes: Seq[AbstractDataType] = Seq(BinaryType)

		override lazy val dataType: DataType =


		case class CatalystDataToAvro(child: Expression) extends UnaryExpression with CodegenFallback {

		override lazy val dataType: DataType = BinaryType


		override def sql: String = simpleString

		override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {

[SPARK-24811][SQL]Avro: add new function from_avro and to_avro #21774

[SPARK-24811][SQL]Avro: add new function from_avro and to_avro #21774

Conversation

gengliangwang commented Jul 15, 2018 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Jul 15, 2018

gengliangwang commented Jul 18, 2018

SparkQA commented Jul 18, 2018

mn-mikke left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Jul 19, 2018

gengliangwang Jul 19, 2018 • edited Loading

Choose a reason for hiding this comment

cloud-fan Jul 19, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Jul 19, 2018

SparkQA commented Jul 19, 2018

SparkQA commented Jul 19, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Jul 20, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HyukjinKwon Jul 20, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gengliangwang Jul 20, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HyukjinKwon left a comment

Choose a reason for hiding this comment

gengliangwang commented Jul 20, 2018

SparkQA commented Jul 20, 2018

SparkQA commented Jul 20, 2018

cloud-fan commented Jul 20, 2018

SparkQA commented Jul 20, 2018

gatorsmile commented Jul 20, 2018

gatorsmile commented Jul 20, 2018 • edited Loading

gengliangwang commented Jul 15, 2018 •

edited

Loading

gengliangwang Jul 19, 2018 •

edited

Loading

cloud-fan Jul 19, 2018 •

edited

Loading

cloud-fan Jul 20, 2018 •

edited

Loading

HyukjinKwon Jul 20, 2018 •

edited

Loading

gengliangwang Jul 20, 2018 •

edited

Loading

gatorsmile commented Jul 20, 2018 •

edited

Loading