Skip to content

[SPARK-21375][PYSPARK][SQL] Add Date and Timestamp support to ArrowConverters for toPandas() Conversion #18664

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
BryanCutler wants to merge 39 commits into from
Closed

[SPARK-21375][PYSPARK][SQL] Add Date and Timestamp support to ArrowConverters for toPandas() Conversion #18664

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
39 commits
Select commit Hold shift + click to select a range
5aa8b9e
added date type and started test, still some issue with time difference
BryanCutler Jul 13, 2017
20313f9
DateTimeUtils forces defaultTimeZone
BryanCutler Jul 18, 2017
69e1e21
fix style checks
BryanCutler Jul 18, 2017
dbfbef3
date type java tests passing
BryanCutler Jul 18, 2017
436afff
timestamp type java tests passing
BryanCutler Jul 18, 2017
78119ca
adding date and timestamp data to python tests, not passing
BryanCutler Jul 19, 2017
b709d78
TimestampType is correctly inferred as datetime64[ns]
BryanCutler Jul 19, 2017
399e527
Merge remote-tracking branch 'upstream/master' into arrow-date-timest…
BryanCutler Jul 24, 2017
e6d8590
Adding DateType and TimestampType to ArrowUtils conversions
BryanCutler Jul 24, 2017
719e77c
using default timezone, fixed tests
BryanCutler Jul 24, 2017
3585520
fixed scala tests for timestamp
BryanCutler Jul 25, 2017
f977d0b
Adding sync between Python and Java default timezones
BryanCutler Jul 26, 2017
b826445
Merge remote-tracking branch 'upstream/master' into arrow-date-timest…
BryanCutler Jul 27, 2017
3b83d7a
added date timestamp writers, fixed tests
BryanCutler Jul 27, 2017
a6009a5
Modify ArrowUtils to have timeZoneId when convert schema to Arrow sch…
ueshin Jul 28, 2017
2ec98cc
fixed python test tearDownClass
BryanCutler Aug 1, 2017
c29018c
using Date.valueOf for tests instead
BryanCutler Aug 2, 2017
7dbdb1f
Made timezone id required for TimestampType
BryanCutler Aug 14, 2017
c3f4e4d
added test for TimestampType without specifying timezone id
BryanCutler Aug 14, 2017
ddbea24
added date and timestamp to ArrowWriter and tests
BryanCutler Aug 15, 2017
c6b597d
removed unused import
BryanCutler Aug 16, 2017
874f104
Merge remote-tracking branch 'upstream/master' into arrow-date-timest…
BryanCutler Oct 10, 2017
d8bae0b
added Python timezone converions for working with Pandas
BryanCutler Oct 10, 2017
36f58b1
Merge remote-tracking branch 'upstream/master' into arrow-date-timest…
BryanCutler Oct 11, 2017
c4fd5ae
fix compilation
BryanCutler Oct 11, 2017
d1617fd
fixed test comp
BryanCutler Oct 11, 2017
d7d9b47
add conversion to Python system local timezone before localize
BryanCutler Oct 11, 2017
efe3e27
timestamps with Arrow almost working for pandas_udfs
BryanCutler Oct 11, 2017
9894519
added workaround for Series to_pandas with timestamps, store os.envir…
BryanCutler Oct 17, 2017
a3ba4ac
change use of xrange for py3
BryanCutler Oct 17, 2017
7266304
remove check for valid timezone in vector for ArrowWriter
BryanCutler Oct 17, 2017
e428cbe
added note for 'us' conversion
BryanCutler Oct 17, 2017
cade921
changed python api for is_datetime64
BryanCutler Oct 19, 2017
f512deb
remove Option for timezoneId
BryanCutler Oct 19, 2017
171d9e1
Merge remote-tracking branch 'upstream/master' into arrow-date-timest…
BryanCutler Oct 20, 2017
79bb93f
added pandas_udf test for date
BryanCutler Oct 23, 2017
c555207
added workaround for date casting, put back check for timestamp conve…
BryanCutler Oct 24, 2017
4d40893
added fillna for null timestamp values
BryanCutler Oct 25, 2017
addd35f
added check for pandas_udf return is a timestamp with tz, added comme…
BryanCutler Oct 26, 2017
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
3 changes: 2 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import org.apache.arrow.vector._
import org.apache.arrow.vector.BaseValueVector.BaseMutator
import org.apache.arrow.vector.file._
import org.apache.arrow.vector.schema.{ArrowFieldNode, ArrowRecordBatch}
import org.apache.arrow.vector.types.FloatingPointPrecision
import org.apache.arrow.vector.types.{DateUnit, FloatingPointPrecision}
import org.apache.arrow.vector.types.pojo.{ArrowType, Field, FieldType, Schema}
import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel

Expand Down Expand Up @@ -84,6 +84,7 @@ private[sql] object ArrowConverters {
case ByteType => new ArrowType.Int(8, true)
case StringType => ArrowType.Utf8.INSTANCE
case BinaryType => ArrowType.Binary.INSTANCE
case DateType => new ArrowType.Date(DateUnit.DAY)
case _ => throw new UnsupportedOperationException(s"Unsupported data type: $dataType")
}
}
Expand Down
75 changes: 71 additions & 4 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowConvertersSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import org.scalatest.BeforeAndAfterAll

import org.apache.spark.SparkException
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.test.SharedSQLContext
import org.apache.spark.sql.types.{BinaryType, StructField, StructType}
import org.apache.spark.util.Utils
Expand Down Expand Up @@ -792,6 +793,76 @@ class ArrowConvertersSuite extends SharedSQLContext with BeforeAndAfterAll {
collectAndValidate(df, json, "binaryData.json")
}

test("date type conversion") {
val json =
s"""
|{
| "schema" : {
| "fields" : [ {
| "name" : "date",
| "type" : {
| "name" : "date",
| "unit" : "DAY"
| },
| "nullable" : true,
| "children" : [ ],
| "typeLayout" : {
| "vectors" : [ {
| "type" : "VALIDITY",
| "typeBitWidth" : 1
| }, {
| "type" : "DATA",
| "typeBitWidth" : 32
| } ]
| }
| } ]
| },
| "batches" : [ {
| "count" : 4,
| "columns" : [ {
| "name" : "date",
| "count" : 4,
| "VALIDITY" : [ 1, 1, 1, 1 ],
| "DATA" : [ -1, 0, 16533, 16930 ]
| } ]
| } ]
|}
""".stripMargin

val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS z", Locale.US)
val d1 = new Date(-1) // "1969-12-31 13:10:15.000 UTC"
val d2 = new Date(0) // "1970-01-01 13:10:15.000 UTC"
val d3 = new Date(sdf.parse("2015-04-08 13:10:15.000 UTC").getTime)
val d4 = new Date(sdf.parse("2016-05-09 12:01:01.000 UTC").getTime)

// Date is created unaware of timezone, but DateTimeUtils force defaultTimeZone()
assert(DateTimeUtils.toJavaDate(DateTimeUtils.fromJavaDate(d2)).getTime == d2.getTime)

@BryanCutler BryanCutler Jul 18, 2017 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

@holdenk @cloud-fan I'm trying out the DateType conversion and ran into this problem. The Dataset encoder uses DateTimeUtils.toJavaDate and fromJavaDate similar to above (which fails), and this forces a defaultTimeZone() when working with the data. So a value new Date(0) should be the epoch, but in my timezone it forces it to be the day before and the test here will not pass.

What are your thoughts on this, should the conversion to Arrow assume the defaultTimeZone()? is this something that should be fixed first in Spark? Thanks!

@kiszk kiszk Jul 18, 2017

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.

cc @ueshin

@ueshin ueshin Jul 18, 2017 •
edited
Loading

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 handle DateType value as number of days from 1970-01-01 internally.

When converting from/to Date to/from internal value, we assume the Date instance contains the timestamp of 00:00:00 time of the day in TimeZone.getDefault() timezone, which is the offset of the timezone. e.g. in JST (GMT+09:00):

scala> TimeZone.setDefault(TimeZone.getTimeZone("JST"))

scala> Date.valueOf("1970-01-01").getTime()
res6: Long = -32400000

whereas in PST (GMT-08:00):

scala> TimeZone.setDefault(TimeZone.getTimeZone("PST"))

scala> Date.valueOf("1970-01-01").getTime()
res8: Long = 28800000

We use DateTimeUtils.defaultTimeZone() to adjust the offset.

@BryanCutler BryanCutler Jul 18, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Thanks @ueshin! I think I got it


val df = Seq(d1, d2, d3, d4).toDF("date")

collectAndValidate(df, json, "dateData.json")
}

ignore("timestamp conversion") {
/*
val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS z", Locale.US)
val ts1 = new Timestamp(sdf.parse("2013-04-08 01:10:15.567 UTC").getTime)
val ts2 = new Timestamp(sdf.parse("2013-04-08 13:10:10.789 UTC").getTime)
val data = Seq(ts1, ts2)

val schema = new JSONSchema(Seq(new TimestampType("timestamp")))
val us_data = data.map(_.getTime * 1000) // convert to microseconds
val columns = Seq(
new PrimitiveColumn("timestamp", data.length, data.map(_ => true), us_data))
val batch = new JSONRecordBatch(data.length, columns)
val json = new JSONFile(schema, Seq(batch))

val df = data.toDF("timestamp")

collectAndValidate(df, json, "timestampData.json")
*/
}

test("floating-point NaN") {
val json =
s"""
Expand Down Expand Up @@ -1044,10 +1115,6 @@ class ArrowConvertersSuite extends SharedSQLContext with BeforeAndAfterAll {
runUnsupported { complexData.toArrowPayload.collect() }

val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS z", Locale.US)
val d1 = new Date(sdf.parse("2015-04-08 13:10:15.000 UTC").getTime)
val d2 = new Date(sdf.parse("2016-05-09 13:10:15.000 UTC").getTime)
runUnsupported { Seq(d1, d2).toDF("date").toArrowPayload.collect() }

val ts1 = new Timestamp(sdf.parse("2013-04-08 01:10:15.567 UTC").getTime)
val ts2 = new Timestamp(sdf.parse("2013-04-08 13:10:10.789 UTC").getTime)
runUnsupported { Seq(ts1, ts2).toDF("timestamp").toArrowPayload.collect() }
Expand Down