title: How to yield pandas dataframe rows to spark dataframe
body: <p>Hi I'm making transformation, I have created <code>some_function(iter)</code> generator to  <code>yield Row(id=index, api=row['api'], A=row['A'], B=row['B']</code>  to yield transformed rows from pandas dataframe to rdd and to spark dataframe. I'm getting errors. (I must use pandas to transform data as there is a large amount of legacy code)</p>
  <p>Input Spark DataFrame</p>
  <pre><code>respond_sdf.show()
      +-------------------------------------------------------------------+
      |content                                                            |
      +-------------------------------------------------------------------+
      |{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
      |{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
      +-------------------------------------------------------------------+
  </code></pre>
  <p>Expected Spark Dataframe after transformation</p>
  <pre><code>transform_df.show()
      +-------------------+
      |  api   |  A  |  B |
      +-------------------+
      | api_1  |  1  |  4 |
      | api_1  |  3  |  5 |
      | api_1  |  4  |  6 |
      | api_2  |  7  | 10 |
      | api_2  |  8  | 11 |
      | api_2  |  9  | 12 |
      +-------------------+
  </code></pre>
  <p>Minimum example code</p>
  <pre><code>#### IMPORT PYSPARK ###
  
  import pandas as pd
  import pyspark
  from pyspark.sql import Row
  from pyspark.sql.types import StructType, StructField, IntegerType,StringType
  spark = pyspark.sql.SparkSession.builder.appName(&quot;test&quot;) \
      .master('local[*]') \
      .getOrCreate()
  sc = spark.sparkContext
  
  
  ####### INPUT DATAFRAME WITH LIST OF JSONS ########################
  
  rdd_list = [[&quot;{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }&quot;],
              [&quot;{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }&quot;]]
  
  schema = StructType([StructField('content', StringType(), True)])
  
  jsons = sc.parallelize(rdd_list)
  respond_sdf = spark.createDataFrame(jsons, schema)
  respond_sdf.show(truncate=False)
  
  
  ####### TRANSFORMATION DATAFRAME ########################
  
  # Pandas transformation function returning pandas dataframe
  def pandas_function(url_json):
      # Complex Pandas transformation
      url = url_json[0]
      json = url_json[1]
      df = pd.DataFrame(eval(json))
      return df
  
  # Generator returing Row from pandas dataframe
  def some_function(iter):
    # Pandas generator
    pandas_df = pandas_function(iter)
    for index, row in pandas_df.iterrows():
        ## ERROR COMES FROM THIS ROW
        yield Row(id=index, api=row['api'], A=row['A'], B=row['B'])
  
  # Creating transformation spark dataframe
  schema = StructType([
    StructField('API', StringType(), True),
    StructField('A', IntegerType(), True),
    StructField('B', IntegerType(), True)
    ])
  
  
  rdd = respond_sdf.rdd.map(lambda x: some_function(x))
  transform_df = spark.createDataFrame(rdd,schema)
  transform_df.show()
  </code></pre>
  <p>I'm getting error below:</p>
  <pre><code>raise TypeError(new_msg(&quot;StructType can not accept object %r in type %s&quot;
  TypeError: StructType can not accept object &lt;generator object some_function at 0x7f69b43def90&gt; in type &lt;class 'generator'&gt;
  </code></pre>
  <p>Full error:</p>
  <pre><code>Py4JJavaError: An error occurred while calling o462.showString.
  : org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 37.0 failed 1 times, most recent failure: Lost task 2.0 in stage 37.0 (TID 97, dpc, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
    File &quot;/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py&quot;, line 605, in main
      process()
    File &quot;/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py&quot;, line 597, in process
      serializer.dump_stream(out_iter, outfile)
    File &quot;/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py&quot;, line 271, in dump_stream
      vs = list(itertools.islice(iterator, batch))
    File &quot;/usr/lib/spark/python/lib/pyspark.zip/pyspark/util.py&quot;, line 107, in wrapper
      return f(*args, **kwargs)
    File &quot;/usr/lib/spark/python/pyspark/sql/session.py&quot;, line 612, in prepare
      verify_func(obj)
    File &quot;/usr/lib/spark/python/pyspark/sql/types.py&quot;, line 1408, in verify
      verify_value(obj)
    File &quot;/usr/lib/spark/python/pyspark/sql/types.py&quot;, line 1395, in verify_struct
      raise TypeError(new_msg(&quot;StructType can not accept object %r in type %s&quot;
  TypeError: StructType can not accept object &lt;generator object some_function at 0x7f69b43def90&gt; in type &lt;class 'generator'&gt;
  </code></pre>
  <p>I'm following advice from the link below:
  <a href="https://stackoverflow.com/questions/59262543/pyspark-convert-result-of-mappartitions-to-spark-dataframe">pySpark convert result of mapPartitions to spark DataFrame</a></p>