Skip to content

[SPARK-22417][PYTHON] Fix for createDataFrame from pandas.DataFrame with timestamp #19646

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 4 commits into from
Closed

[SPARK-22417][PYTHON] Fix for createDataFrame from pandas.DataFrame with timestamp #19646

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cca6757
added fix for pandas timestamp to convert to microseconds for createD…
BryanCutler Nov 2, 2017
839bb50
did not need a func to apply fix
BryanCutler Nov 2, 2017
3944b5c
moved conversion to internal methods
BryanCutler Nov 3, 2017
355edfc
use datetime64 instead of M8, minor fixes
BryanCutler Nov 6, 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
32 changes: 31 additions & 1 deletion python/pyspark/sql/session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -512,9 +512,39 @@ def createDataFrame(self, data, schema=None, samplingRatio=None, verifySchema=Tr
except Exception:
has_pandas = False
if has_pandas and isinstance(data, pandas.DataFrame):
import numpy as np

# Convert pandas.DataFrame to list of numpy records
np_records = data.to_records(index=False)

# Check if any columns need to be fixed for Spark to infer properly
record_type_list = None
if schema is None and len(np_records) > 0:
cur_dtypes = np_records[0].dtype
col_names = cur_dtypes.names
record_type_list = []
has_rec_fix = False
for i in xrange(len(cur_dtypes)):

@viirya viirya Nov 3, 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.

oh, session.py didn't define xrange for version > 3.

@BryanCutler BryanCutler Nov 3, 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.

Ooops, I forgot about that. thx!

curr_type = cur_dtypes[i]
# If type is a datetime64 timestamp, convert to microseconds
# NOTE: if dtype is M8[ns] then np.record.tolist() will output values as longs,
# this conversion will lead to an output of py datetime objects, see SPARK-22417
if curr_type == np.dtype('M8[ns]'):
curr_type = 'M8[us]'
has_rec_fix = True
record_type_list.append((str(col_names[i]), curr_type))
if not has_rec_fix:
record_type_list = None

@viirya viirya Nov 3, 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.

Shall we put this conversion into an internal method?

@BryanCutler BryanCutler Nov 3, 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.

Yeah, probably a good idea. I'll see if I can clean it up some. Thanks @viirya !


# If no schema supplied by user then get the names of columns only
if schema is None:
schema = [str(x) for x in data.columns]
data = [r.tolist() for r in data.to_records(index=False)]

@cloud-fan cloud-fan Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

seems r.tolist is the problem, how about r[i] for i in xrange(r.size)? Then we can get numpy.datatype64

>>> pd.DataFrame({"ts": [datetime(2017, 10, 31, 1, 1, 1)]}).to_records(index=False)[0].tolist()[0]
1509411661000000000L
>>> pd.DataFrame({"ts": [datetime(2017, 10, 31, 1, 1, 1)]}).to_records(index=False)[0][0]
numpy.datetime64('2017-10-31T02:01:01.000000000+0100')
>>>

@HyukjinKwon HyukjinKwon Nov 3, 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.

but ... numpy.datetime64 is not supported in createDataFrame IIUC:

import pandas as pd
from datetime import datetime
pdf = pd.DataFrame({"ts": [datetime(2017, 10, 31, 1, 1, 1)]})
print [[v for v in r] for r in pdf.to_records(index=False)]
spark.createDataFrame([[v for v in r] for r in pdf.to_records(index=False)])
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/.../spark/python/pyspark/sql/session.py", line 591, in createDataFrame
    rdd, schema = self._createFromLocal(map(prepare, data), schema)
  File "/.../spark/python/pyspark/sql/session.py", line 404, in _createFromLocal
    struct = self._inferSchemaFromList(data)
  File "/.../spark/python/pyspark/sql/session.py", line 336, in _inferSchemaFromList
    schema = reduce(_merge_type, map(_infer_schema, data))
  File "/.../spark/python/pyspark/sql/types.py", line 1095, in _infer_schema
    fields = [StructField(k, _infer_type(v), True) for k, v in items]
  File "/.../spark/python/pyspark/sql/types.py", line 1072, in _infer_type
    raise TypeError("not supported type: %s" % type(obj))
TypeError: not supported type: <type 'numpy.datetime64'>

@HyukjinKwon HyukjinKwon Nov 3, 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.

(It reminds of me SPARK-6857 BTW)

@cloud-fan cloud-fan Nov 3, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

according to the ticket, seems we need to convert numpy.datetime64 to python datetime manually.

@BryanCutler BryanCutler Nov 6, 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.

the problem is that nanosecond values can not be converted to a python datetime object, which only has microsecond resolution, so numpy converts it to long. Numpy will convert microseconds and above to python datetime objects, which Spark will correctly infer.

according to the ticket, seems we need to convert numpy.datetime64 to python datetime manually.

This fix is just meant to convert nanosecond timestamps to microseconds so that calling tolist() can fit them in a python object. Does it seem ok to you guys to leave it at that scope for now?


# Convert list of numpy records to python lists
if record_type_list is not None:
data = [r.astype(record_type_list).tolist() for r in np_records]
else:
data = [r.tolist() for r in np_records]

if isinstance(schema, StructType):
verify_func = _make_type_verifier(schema) if verifySchema else lambda _: True
Expand Down
10 changes: 10 additions & 0 deletions python/pyspark/sql/tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2592,6 +2592,16 @@ def test_create_dataframe_from_array_of_long(self):
df = self.spark.createDataFrame(data)
self.assertEqual(df.first(), Row(longarray=[-9223372036854775808, 0, 9223372036854775807]))

@unittest.skipIf(not _have_pandas, "Pandas not installed")

@dongjoon-hyun dongjoon-hyun Nov 9, 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.

Hi, @cloud-fan and @BryanCutler .
This seems to break branch-2.2, but has been hidden behind another SQL error. (cc @gatorsmile , @henryr)
Please see this.

cc @felixcheung since he is RM for 2.2.1.

@BryanCutler BryanCutler Nov 9, 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.

Thanks @dongjoon-hyun for tracking this down. It looks like sql/tests.py for branch-2.2 is just missing the following


_have_pandas = False
try:
    import pandas
    _have_pandas = True
except:
    # No Pandas, but that's okay, we'll skip those tests
    pass

This was probably added from an earlier PR in master and wasn't included when this one was cherry-picked.

@BryanCutler BryanCutler Nov 9, 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.

I can add a patch a little bit later tonight unless someone is able to get to it first.

@ueshin ueshin Nov 9, 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.

I can take it over. I'll submit a pr soon.

@dongjoon-hyun dongjoon-hyun Nov 9, 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.

Thank you, @BryanCutler !

@dongjoon-hyun dongjoon-hyun Nov 9, 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.

Great, @ueshin ! :)

@dongjoon-hyun dongjoon-hyun Nov 9, 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.

BTW, @ueshin .
branch-2.2 Jenkins will fail due to #19701 .
Could you review and merge #19701 first?

@ueshin ueshin Nov 9, 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.

Ah, in that case, maybe we need to revert one of the two original patches and fix one by one, or merge the two follow-ups into one as a hot-fix pr. cc @gatorsmile @cloud-fan

def test_create_dataframe_from_pandas_with_timestamp(self):
import pandas as pd
from datetime import datetime
pdf = pd.DataFrame({"ts": [datetime(2017, 10, 31, 1, 1, 1)],
"d": [pd.Timestamp.now().date()]})
df = self.spark.createDataFrame(pdf)

@ueshin ueshin Nov 3, 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.

What if we specify the schema? For example:

df = self.spark.createDataFrame(pdf, "ts timestamp, d date")

@HyukjinKwon HyukjinKwon Nov 3, 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.

I was checking this PR and ran this for my curiosity. I got:

import pandas as pd
from datetime import datetime

pdf = pd.DataFrame({"ts": [datetime(2017, 10, 31, 1, 1, 1)], "d": [pd.Timestamp.now().date()]})
spark.createDataFrame(pdf, "d date, ts timestamp")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/.../spark/python/pyspark/sql/session.py", line 587, in createDataFrame
    rdd, schema = self._createFromLocal(map(prepare, data), schema)
  File "/.../spark/python/pyspark/sql/session.py", line 401, in _createFromLocal
    data = list(data)
  File "/.../spark/python/pyspark/sql/session.py", line 567, in prepare
    verify_func(obj)
  File "/.../spark/python/pyspark/sql/types.py", line 1411, in verify
    verify_value(obj)
  File "/.../spark/python/pyspark/sql/types.py", line 1392, in verify_struct
    verifier(v)
  File "/.../spark/python/pyspark/sql/types.py", line 1411, in verify
    verify_value(obj)
  File "/.../spark/python/pyspark/sql/types.py", line 1405, in verify_default
    verify_acceptable_types(obj)
  File "/.../spark/python/pyspark/sql/types.py", line 1300, in verify_acceptable_types
    % (dataType, obj, type(obj))))
TypeError: field ts: TimestampType can not accept object 1509411661000000000L in type <type 'long'>

Seems we should fix this one too.

@BryanCutler BryanCutler Nov 6, 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.

Yes, looks like that needs to be fixed also. I thought it was working when schema was supplied, but I'll double-check and add that into the tests.

self.assertTrue(isinstance(df.schema['ts'].dataType, TimestampType))
self.assertTrue(isinstance(df.schema['d'].dataType, DateType))


class HiveSparkSubmitTests(SparkSubmitTests):

Expand Down