Skip to content

[SPARK-31937][SQL] Support processing ArrayType/MapType/StructType data using no-serde mode script transform #30957

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

[SPARK-31937][SQL] Support processing ArrayType/MapType/StructType data using no-serde mode script transform #30957

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
adc9ded
[SPARK-31937][SQL] Support processing array and map type using spark …
AngersZhuuuu Dec 29, 2020
6a7438b
Update CatalystTypeConverters.scala
AngersZhuuuu Dec 29, 2020
d3b9cec
fix failed UT
AngersZhuuuu Dec 29, 2020
fdd5225
Update SparkScriptTransformationSuite.scala
AngersZhuuuu Dec 29, 2020
aa16c8f
Update BaseScriptTransformationSuite.scala
AngersZhuuuu Dec 29, 2020
092c927
Update BaseScriptTransformationExec.scala
AngersZhuuuu Dec 29, 2020
9761c0e
Merge branch 'master' into SPARK-31937
AngersZhuuuu Dec 29, 2020
28ad7fa
Update BaseScriptTransformationSuite.scala
AngersZhuuuu Dec 29, 2020
9ac75fc
Merge branch 'master' into SPARK-31937
AngersZhuuuu Jan 4, 2021
33d8b5b
Update BaseScriptTransformationExec.scala
AngersZhuuuu Jan 4, 2021
63f07eb
follow comment
AngersZhuuuu Feb 4, 2021
b631b70
Update BaseScriptTransformationExec.scala
AngersZhuuuu Feb 4, 2021
b7e7f92
follow comment
AngersZhuuuu Feb 5, 2021
8dec5a1
follow comment
AngersZhuuuu Feb 5, 2021
529d54d
Update BaseScriptTransformationExec.scala
AngersZhuuuu Feb 6, 2021
4f0e78f
Avoid construct JsonToStructs repeated
AngersZhuuuu Feb 6, 2021
ed8c54c
remove unused UT
AngersZhuuuu Feb 6, 2021
520f4b8
Update sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScr…
AngersZhuuuu Apr 16, 2021
97f9d58
Merge branch 'master' into SPARK-31937
AngersZhuuuu Apr 16, 2021
b5a4268
[SPARK-35097][SQL] Add column name to SparkUpgradeException about anc…
AngersZhuuuu Apr 18, 2021
76a746e
Revert "[SPARK-35097][SQL] Add column name to SparkUpgradeException a…
AngersZhuuuu Apr 18, 2021
6aa05fc
fix UT
AngersZhuuuu Apr 19, 2021
9e3f808
Revert "fix UT"
AngersZhuuuu Apr 19, 2021
3f51d27
fix UT
AngersZhuuuu Apr 19, 2021
adf8a66
Update sql-migration-guide.md
AngersZhuuuu Apr 19, 2021
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
13 changes: 11 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ import org.apache.spark.{SparkException, SparkFiles, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Cast, Expression, GenericInternalRow, UnsafeProjection}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Cast, Expression, GenericInternalRow, JsonToStructs, Literal, StructsToJson, UnsafeProjection}
import org.apache.spark.sql.catalyst.plans.logical.ScriptInputOutputSchema
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.catalyst.util.{DateTimeUtils, IntervalUtils}
Expand All @@ -47,7 +47,13 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
def ioschema: ScriptTransformationIOSchema

protected lazy val inputExpressionsWithoutSerde: Seq[Expression] = {
input.map(Cast(_, StringType).withTimeZone(conf.sessionLocalTimeZone))
input.map { in =>
Comment thread
AngersZhuuuu marked this conversation as resolved.
in.dataType match {
case _: ArrayType | _: MapType | _: StructType =>
new StructsToJson(in).withTimeZone(conf.sessionLocalTimeZone)

@maropu maropu Feb 5, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it okay to follow the default behaviour w/o options?

@AngersZhuuuu AngersZhuuuu Feb 5, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it okay to follow the default behaviour w/o options?

I am not so familiar about this part's properties. (I am checking these properties)
Just a quick thought, how about we add inputSerdeProps as StructTJson's properties and add outputSerdeProps as JsonToStruct 's perperties.

@maropu maropu Feb 5, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a quick thought, how about we add inputSerdeProps as StructTJson's properties and add outputSerdeProps as JsonToStruct 's perperties.

Ah, it seems fine. Could you try it?

@AngersZhuuuu AngersZhuuuu Feb 6, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a quick thought, how about we add inputSerdeProps as StructTJson's properties and add outputSerdeProps as JsonToStruct 's perperties.

Ah, it seems fine. Could you try it?

Updated.

@viirya viirya Feb 5, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How do users know we serialize complex type data using json format? This requires the external process know to parse json back to complext data, isn't? Is it well known to use json for similar feature in Hive?

@AngersZhuuuu AngersZhuuuu Feb 6, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How do users know we serialize complex type data using json format? This requires the external process know to parse json back to complext data, isn't? Is it well known to use json for similar feature in Hive?

In the first I implement it in same way as hive, then change to use json serde, you can see discussion in this part of comment #30957 (comment)

@AngersZhuuuu AngersZhuuuu Feb 6, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, it's a new feature, so we can have a migration guide or release note make user know the new feature and difference.

@maropu maropu Feb 10, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hm. Actually, this feature is not relevant to users using pre-built spark (w/ -Phive enabled)? IIUC there is no way to use this feature when enabling -Phive. I'm not sure about how important this feature is, but I think we need to document it somewhere if we will merge it.

@AngersZhuuuu AngersZhuuuu Mar 26, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hm. Actually, this feature is not relevant to users using pre-built spark (w/ -Phive enabled)? IIUC there is no way to use this feature when enabling -Phive. I'm not sure about how important this feature is, but I think we need to document it somewhere if we will merge it.

That's why I implement this in hive's way first and support make json as a serde. If we implement this in json, then some origin hive SQL still can't run when directly upgrade to spark sql (w/ -Phive)。 How about add a configuration such as
spark.sql.sccriptTransform.useSparkFirst and add document?

case _ => Cast(in, StringType).withTimeZone(conf.sessionLocalTimeZone)
}
}
}

override def producedAttributes: AttributeSet = outputSet -- inputSet
Expand Down Expand Up @@ -220,6 +226,9 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
case CalendarIntervalType => wrapperConvertException(
data => IntervalUtils.stringToInterval(UTF8String.fromString(data)),
converter)
case _: ArrayType | _: MapType | _: StructType =>
wrapperConvertException(data => JsonToStructs(attr.dataType, Map.empty[String, String],

@maropu maropu Feb 5, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This can cause much overhead cuz this make a new object (JsonToStructs ) for each call. Could you avoid it?

@AngersZhuuuu AngersZhuuuu Feb 5, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This can cause much overhead cuz this make a new object (JsonToStructs ) for each call. Could you avoid it?

This problem also happen in input side's Cast and StructToJson.
To avoid it maybe we need to extract common method from these expression or just write some thing for ScriptTransform. WDYT @cloud-fan

@maropu maropu Feb 5, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This problem also happen in input side's Cast and StructToJson.

Really? I think the input side does not have the issue. Why do you think so?

@AngersZhuuuu AngersZhuuuu Feb 6, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This problem also happen in input side's Cast and StructToJson.

Really? I think the input side does not have the issue. Why do you think so?

Hmmm, I got your point. I misunderstood you point. Updated, how about current? Test with df.repartition(1) then transform. It only create JsonToStruct once for one partition.

Literal(data), Some(conf.sessionLocalTimeZone)).eval(), any => any)
Comment thread
maropu marked this conversation as resolved.
Outdated
case udt: UserDefinedType[_] =>
wrapperConvertException(data => udt.deserialize(data), converter)
case dt =>
Expand Down
130 changes: 126 additions & 4 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,14 +302,16 @@ abstract class BaseScriptTransformationSuite extends SparkPlanTest with SQLTestU
script = "cat",
output = Seq(
AttributeReference("a", CalendarIntervalType)(),
AttributeReference("b", StringType)(),
AttributeReference("c", StringType)(),
AttributeReference("d", StringType)(),
AttributeReference("b", ArrayType(IntegerType))(),
AttributeReference("c", MapType(StringType, IntegerType))(),
AttributeReference("d", StructType(
Array(StructField("_1", IntegerType),
StructField("_2", IntegerType))))(),
AttributeReference("e", new SimpleTupleUDT)()),
child = child,
ioschema = defaultIOSchema
),
df.select('a, 'b.cast("string"), 'c.cast("string"), 'd.cast("string"), 'e).collect())
df.select('a, 'b, 'c, 'd, 'e).collect())
}
}

Expand Down Expand Up @@ -471,6 +473,126 @@ abstract class BaseScriptTransformationSuite extends SparkPlanTest with SQLTestU
}
}

test("SPARK-31936: Script transform support ArrayType/MapType/StructType (no serde)") {
assume(TestUtils.testCommandAvailable("python"))
withTempView("v") {
val df = Seq(
(Array(0, 1, 2), Array(Array(0, 1), Array(2)),
Map("a" -> 1), Map("b" -> Array("a", "b"))),
(Array(3, 4, 5), Array(Array(3, 4), Array(5)),
Map("b" -> 2), Map("c" -> Array("c", "d"))),
(Array(6, 7, 8), Array(Array(6, 7), Array(8)),
Map("c" -> 3), Map("d" -> Array("e", "f")))
).toDF("a", "b", "c", "d")
.select('a, 'b, 'c, 'd,
struct('a, 'b).as("e"),
struct('a, 'd).as("f"),
struct(struct('a, 'b), struct('a, 'd)).as("g")
)

checkAnswer(
df,
(child: SparkPlan) => createScriptTransformationExec(
input = Seq(
df.col("a").expr,
df.col("b").expr,
df.col("c").expr,
df.col("d").expr,
df.col("e").expr,
df.col("f").expr,
df.col("g").expr),
script = "cat",
output = Seq(
AttributeReference("a", ArrayType(IntegerType))(),
AttributeReference("b", ArrayType(ArrayType(IntegerType)))(),
AttributeReference("c", MapType(StringType, IntegerType))(),
AttributeReference("d", MapType(StringType, ArrayType(StringType)))(),
AttributeReference("e", StructType(
Array(StructField("a", ArrayType(IntegerType)),
StructField("b", ArrayType(ArrayType(IntegerType))))))(),
AttributeReference("f", StructType(
Array(StructField("a", ArrayType(IntegerType)),
StructField("d", MapType(StringType, ArrayType(StringType))))))(),
AttributeReference("g", StructType(
Array(StructField("col1", StructType(
Array(StructField("a", ArrayType(IntegerType)),
StructField("b", ArrayType(ArrayType(IntegerType)))))),
StructField("col2", StructType(
Array(StructField("a", ArrayType(IntegerType)),
StructField("d", MapType(StringType, ArrayType(StringType)))))))))()),
child = child,
ioschema = defaultIOSchema
),
df.select('a, 'b, 'c, 'd, 'e, 'f, 'g).collect())
}
}

test("SPARK-31936: Script transform support nested complex type (no serde)") {

@maropu maropu Feb 5, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We still need this test now? On second thought, only the test "SPARK-31936: Script transform support ArrayType/MapType/StructType (no serde)" looks fine.

@AngersZhuuuu AngersZhuuuu Feb 5, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We still need this test now? On second thought, only the test "SPARK-31936: Script transform support ArrayType/MapType/StructType (no serde)" looks fine.

Emmm, add a more complex test is more better, how about combine this two test?

@maropu maropu Feb 5, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's the purpose of this tests? It seems this test does not improve test coverage.

@AngersZhuuuu AngersZhuuuu Feb 6, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's the purpose of this tests? It seems this test does not improve test coverage.

Want to show we can support more complex case. Ok, I will remove this.

assume(TestUtils.testCommandAvailable("python"))
withTempView("v") {
val df = Seq(
(Array(Array(Array(Array(Array(Array(1, 2, 3)))))),
Array(Array(Array(Array(Array(Array(1, 2, 3))), Array(Array(Array(1, 2, 3)))))),
Map("a" -> Map("c" -> Map("d" -> Array(1, 2, 3))),
"b" -> Map("c" -> Map("d" -> Array(1, 2, 3))))
)
).toDF("a", "b", "c").select('a, 'b, 'c,
struct('a, 'b, 'c).as("d")
).select('a, 'b, 'c, 'd,
struct('c, 'd).as("e")
)

checkAnswer(
df,
(child: SparkPlan) => createScriptTransformationExec(
input = Seq(
df.col("a").expr,
df.col("b").expr,
df.col("c").expr,
df.col("d").expr,
df.col("e").expr),
script = "cat",
output = Seq(
AttributeReference("a",
ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(IntegerType)))))))(),
AttributeReference("b",
ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(IntegerType)))))))(),
AttributeReference("c",
MapType(StringType, MapType(StringType,
MapType(StringType, ArrayType(IntegerType)))))(),
AttributeReference("d",
StructType(Array(
StructField("a",
ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(IntegerType))))))),
StructField("b",
ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(IntegerType))))))),
StructField("c",
MapType(StringType, MapType(StringType,
MapType(StringType, ArrayType(IntegerType))))))))(),
AttributeReference("e",
StructType(Array(
StructField("c",
MapType(StringType, MapType(StringType,
MapType(StringType, ArrayType(IntegerType))))),
StructField("d",
StructType(Array(
StructField("a",
ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(
ArrayType(IntegerType))))))),
StructField("b",
ArrayType(ArrayType(ArrayType(ArrayType(ArrayType(
ArrayType(IntegerType))))))),
StructField("c",
MapType(StringType, MapType(StringType,
MapType(StringType, ArrayType(IntegerType)))))))))))()
),
child = child,
ioschema = defaultIOSchema
),
df.select('a, 'b, 'c, 'd, 'e).collect())
}
}

test("SPARK-33934: Add SparkFile's root dir to env property PATH") {
assume(TestUtils.testCommandAvailable("python"))
val scriptFilePath = copyAndGetResourceFile("test_script.py", ".py").getAbsoluteFile
Expand Down
42 changes: 1 addition & 41 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/SparkScriptTransformationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.sql.execution

import org.apache.spark.{SparkException, TestUtils}
import org.apache.spark.TestUtils
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.test.SharedSparkSession
Expand Down Expand Up @@ -59,44 +59,4 @@ class SparkScriptTransformationSuite extends BaseScriptTransformationSuite with
assert(e.contains("TRANSFORM with serde is only supported in hive mode"))
}
}

test("SPARK-32106: TRANSFORM doesn't support ArrayType/MapType/StructType " +
"as output data type (no serde)") {
assume(TestUtils.testCommandAvailable("/bin/bash"))
// check for ArrayType
val e1 = intercept[SparkException] {
sql(
"""
|SELECT TRANSFORM(a)
|USING 'cat' AS (a array<int>)
|FROM VALUES (array(1, 1), map('1', 1), struct(1, 'a')) t(a, b, c)
""".stripMargin).collect()
}.getMessage
assert(e1.contains("SparkScriptTransformation without serde does not support" +
" ArrayType as output data type"))

// check for MapType
val e2 = intercept[SparkException] {
sql(
"""
|SELECT TRANSFORM(b)
|USING 'cat' AS (b map<int, string>)
|FROM VALUES (array(1, 1), map('1', 1), struct(1, 'a')) t(a, b, c)
""".stripMargin).collect()
}.getMessage
assert(e2.contains("SparkScriptTransformation without serde does not support" +
" MapType as output data type"))

// check for StructType
val e3 = intercept[SparkException] {
sql(
"""
|SELECT TRANSFORM(c)
|USING 'cat' AS (c struct<col1:int, col2:string>)
|FROM VALUES (array(1, 1), map('1', 1), struct(1, 'a')) t(a, b, c)
""".stripMargin).collect()
}.getMessage
assert(e3.contains("SparkScriptTransformation without serde does not support" +
" StructType as output data type"))
}
}