Use arrow_column.flatten().

Author: ueshin

python/pyspark/serializers.py

@@ -390,13 +390,8 @@ def arrow_to_pandas(self, arrow_column, data_type):
 
         if self._df_for_struct and type(data_type) == StructType:
             import pandas as pd
-            import pyarrow as pa
-            column_arrays = zip(*[[chunk.field(i)
-                                   for i in range(chunk.type.num_children)]
-                                  for chunk in arrow_column.data.iterchunks()])
-            series = [_arrow_column_to_pandas(pa.column(field.name, pa.chunked_array(arrays)),
-                                              field.dataType)
-                      for arrays, field in zip(column_arrays, data_type)]
+            series = [_arrow_column_to_pandas(column, field.dataType).rename(field.name)
+                      for column, field in zip(arrow_column.flatten(), data_type)]
             s = _check_dataframe_localize_timestamps(pd.concat(series, axis=1), self._timezone)
         else:
             s = super(ArrowStreamPandasUDFSerializer, self).arrow_to_pandas(arrow_column, data_type)

python/pyspark/sql/tests/test_pandas_udf_scalar.py

@@ -294,7 +294,7 @@ def func(id):
 
         struct_f = pandas_udf(lambda x: x, return_type)
         actual = df.select(struct_f(struct(col('id'), col('id').cast('string').alias('str'))))
-        if LooseVersion(pa.__version__) < LooseVersion("0.11.0"):
+        if LooseVersion(pa.__version__) < LooseVersion("0.10.0"):
             with QuietTest(self.sc):
                 from py4j.protocol import Py4JJavaError
                 with self.assertRaisesRegexp(

python/pyspark/sql/types.py

@@ -1675,10 +1675,10 @@ def from_arrow_type(at):
             raise TypeError("Unsupported type in conversion from Arrow: " + str(at))
         spark_type = ArrayType(from_arrow_type(at.value_type))
     elif types.is_struct(at):
-        # TODO: remove version check once minimum pyarrow version is 0.11.0
-        if LooseVersion(pa.__version__) < LooseVersion("0.11.0"):
+        # TODO: remove version check once minimum pyarrow version is 0.10.0
+        if LooseVersion(pa.__version__) < LooseVersion("0.10.0"):
             raise TypeError("Unsupported type in conversion from Arrow: " + str(at) +
-                            "\nPlease install pyarrow >= 0.11.0 for StructType support.")
+                            "\nPlease install pyarrow >= 0.10.0 for StructType support.")
         if any(types.is_struct(field.type) for field in at):
             raise TypeError("Nested StructType not supported in conversion from Arrow: " + str(at))
         return StructType(