Fix pandas_udf with return type StringType() to handle str type prope…

…rly.

python/pyspark/serializers.py

@@ -230,6 +230,9 @@ def create_array(s, t):
             s = _check_series_convert_timestamps_internal(s.fillna(0), timezone)
             # TODO: need cast after Arrow conversion, ns values cause error with pandas 0.19.2
             return pa.Array.from_pandas(s, mask=mask).cast(t, safe=False)
+        elif t is not None and pa.types.is_string(t) and sys.version < '3':
+            # TODO: need decode before converting to Arrow in Python 2
+            return pa.Array.from_pandas(s.str.decode('utf-8'), mask=mask, type=t)
         return pa.Array.from_pandas(s, mask=mask, type=t)
 
     arrs = [create_array(s, t) for s, t in series]

python/pyspark/sql/tests.py

@@ -3920,6 +3920,14 @@ def test_vectorized_udf_null_string(self):
         res = df.select(str_f(col('str')))
         self.assertEquals(df.collect(), res.collect())
 
+    def test_vectorized_udf_string_in_udf(self):
+        from pyspark.sql.functions import pandas_udf, col
+        import pandas as pd
+        df = self.spark.range(10)
+        str_f = pandas_udf(lambda x: pd.Series(["%s" % i for i in x]), StringType())
+        res = df.select(str_f(col('id')))
+        self.assertEquals(df.select(col('id').cast('string')).collect(), res.collect())
+
     def test_vectorized_udf_datatype_string(self):
         from pyspark.sql.functions import pandas_udf, col
         df = self.spark.range(10).select(