[SPARK-43473][PYTHON] Support struct type in createDataFrame from pandas DataFrame

[ueshin]

What changes were proposed in this pull request?

Supports struct type in createDataFrame from pandas DataFrame.

With Arrow optimization, it works without the fallback:

>>> import pandas as pd
>>> from pyspark.sql.types import Row
>>>
>>> pdf = pd.DataFrame(
...     {"a": [Row(1, "a"), Row(2, "b")], "b": [{"s": 3, "t": "x"}, {"s": 4, "t": "y"}]}
... )
>>> schema = "a struct<x int, y string>, b struct<s int, t string>"
>>>
>>> df = spark.createDataFrame(pdf, schema)
>>> df.show()
+------+------+
|     a|     b|
+------+------+
|{1, a}|{3, x}|
|{2, b}|{4, y}|
+------+------+

and Spark Connect also works.

Why are the changes needed?

In vanilla PySpark without Arrow optimization, Row object or dict can be handled as struct type if the schema is provided:

>>> import pandas as pd
>>> from pyspark.sql.types import *
>>>
>>> pdf = pd.DataFrame(
...     {"a": [Row(1, "a"), Row(2, "b")], "b": [{"s": 3, "t": "x"}, {"s": 4, "t": "y"}]}
... )
>>> schema = "a struct<x int, y string>, b struct<s int, t string>"
>>>
>>> df = spark.createDataFrame(pdf, schema)
>>> df.show()
+------+------+
|     a|     b|
+------+------+
|{1, a}|{3, x}|
|{2, b}|{4, y}|
+------+------+

Whereas with Arrow, it uses a fallback to make it:

>>> spark.conf.set('spark.sql.execution.arrow.pyspark.enabled', True)
>>> spark.createDataFrame(pdf, schema).show()
/.../pyspark/sql/pandas/conversion.py:329: UserWarning: createDataFrame attempted Arrow optimization because 'spark.sql.execution.arrow.pyspark.enabled' is set to true; however, failed by the reason below:
  A field of type StructType expects a pandas.DataFrame, but got: <class 'pandas.core.series.Series'>
Attempting non-optimization as 'spark.sql.execution.arrow.pyspark.fallback.enabled' is set to true.
  warn(msg)
+------+------+
|     a|     b|
+------+------+
|{1, a}|{3, x}|
|{2, b}|{4, y}|
+------+------+

and Spark Connect fails:

>>> df = spark.createDataFrame(pdf, schema)
Traceback (most recent call last):
...
ValueError: A field of type StructType expects a pandas.DataFrame, but got: <class 'pandas.core.series.Series'>

Does this PR introduce any user-facing change?

Row object or dict in pandas DataFrame works as struct type when createDataFrame if the schema is provided.

How was this patch tested?

Added the related test.

[zhengruifeng]

nit, shall we use PySparkValueError here and above?

[HyukjinKwon]

cc @itholic

[itholic]

Yeah, at least here I think we should raise PySparkValueError.
For above errors seems like they're generated from PyArrow internally, so I guess maybe we can't catch them by PySparkxxxError.

[xinrong-meng]

Does that refactoring still conform to UNSUPPORTED_DATA_TYPE_FOR_ARROW_VERSION?

[ueshin]

@xinrong-meng

Does that refactoring still conform to UNSUPPORTED_DATA_TYPE_FOR_ARROW_VERSION?

This PR doesn't change anything related to pyarrow version.

[xinrong-meng]

Sorry I meant UNSUPPORTED_DATA_TYPE_FOR_ARROW_CONVERSION. Do we have plans to remove the constraints?
@ueshin

[xinrong-meng]

Specifically, nested StructType, and MapType with keys/values in StructType/TimestampType?

[ueshin]

Do we have plans to remove the constraints?

I'm not sure if it's planned, but now we can remove the constraints with a bit more work.

[HyukjinKwon]

Merged to master.

[BryanCutler]

This looks great, thanks for doing it @ueshin !