[SPARK-52433][PYTHON] Unify the string coercion in createDataFrame

python/pyspark/sql/conversion.py

@@ -255,7 +255,7 @@ def convert_string(value: Any) -> Any:
                     return None
                 else:
                     if isinstance(value, bool):
-                        # To match the PySpark which convert bool to string in
+                        # To match the PySpark Classic which convert bool to string in
                         # the JVM side (python.EvaluatePython.makeFromJava)
                         return str(value).lower()
                     else:

python/pyspark/sql/session.py

@@ -54,6 +54,7 @@
     DataType,
     StructField,
     StructType,
+    VariantVal,
     _make_type_verifier,
     _infer_schema,
     _has_nulltype,
@@ -1194,18 +1195,23 @@ def _createFromLocal(
         if not isinstance(data, list):
             data = list(data)
 
+        if any(isinstance(d, VariantVal) for d in data):
+            raise PySparkValueError("Rows cannot be of type VariantVal")
+
+        tupled_data: Iterable[Tuple]
         if schema is None or isinstance(schema, (list, tuple)):
             struct = self._inferSchemaFromList(data, names=schema)
             converter = _create_converter(struct)
-            tupled_data: Iterable[Tuple] = map(converter, data)
+            tupled_data = map(converter, data)
             if isinstance(schema, (list, tuple)):
                 for i, name in enumerate(schema):
                     struct.fields[i].name = name
                     struct.names[i] = name
 
         elif isinstance(schema, StructType):
             struct = schema
-            tupled_data = data
+            converter = _create_converter(struct)
+            tupled_data = map(converter, data)
 
         else:
             raise PySparkTypeError(

python/pyspark/sql/tests/test_creation.py

@@ -24,6 +24,9 @@
 from pyspark.sql import Row
 import pyspark.sql.functions as F
 from pyspark.sql.types import (
+    StructType,
+    StructField,
+    StringType,
     DateType,
     TimestampType,
     TimestampNTZType,
@@ -42,6 +45,21 @@
 
 
 class DataFrameCreationTestsMixin:
+    def test_create_str_from_dict(self):
+        data = [
+            {"broker": {"teamId": 3398, "contactEmail": "abc.xyz@123.ca"}},
+        ]
+
+        for schema in [
+            StructType([StructField("broker", StringType())]),
+            "broker: string",
+        ]:
+            df = self.spark.createDataFrame(data, schema=schema)
+            self.assertEqual(
+                df.first().broker,
+                """{'teamId': 3398, 'contactEmail': 'abc.xyz@123.ca'}""",
+            )
+
     def test_create_dataframe_from_array_of_long(self):
         import array
 

python/pyspark/sql/types.py

@@ -2510,6 +2510,9 @@ def _need_converter(dataType: DataType) -> bool:
         return _need_converter(dataType.keyType) or _need_converter(dataType.valueType)
     elif isinstance(dataType, NullType):
         return True
+    elif isinstance(dataType, StringType):
+        # Coercion to StringType is allowed, e.g. dict -> str
+        return True
     else:
         return False
 
@@ -2521,16 +2524,35 @@ def _create_converter(dataType: DataType) -> Callable:
 
     if isinstance(dataType, ArrayType):
         conv = _create_converter(dataType.elementType)
-        return lambda row: [conv(v) for v in row]
+        return lambda row: [conv(v) for v in row] if row is not None else None
 
     elif isinstance(dataType, MapType):
         kconv = _create_converter(dataType.keyType)
         vconv = _create_converter(dataType.valueType)
-        return lambda row: dict((kconv(k), vconv(v)) for k, v in row.items())
+        return (
+            lambda row: dict((kconv(k), vconv(v)) for k, v in row.items())
+            if row is not None
+            else None
+        )
 
     elif isinstance(dataType, NullType):
         return lambda x: None
 
+    elif isinstance(dataType, StringType):
+
+        def convert_string(value: Any) -> Any:
+            if value is None:
+                return None
+            else:
+                if isinstance(value, bool):
+                    # To match the Classic behavior
+                    return str(value).lower()
+                else:
+                    # Coercion to StringType is allowed, e.g. dict -> str
+                    return str(value)
+
+        return convert_string
+
     elif not isinstance(dataType, StructType):
         return lambda x: x