title: Spark: How to correctly transform dataframe by mapInPandas
body: <p>I'm trying to transform spark dataframe with 10k rows by latest spark 3.0.1 function <a href="https://spark.apache.org/docs/latest/sql-pyspark-pandas-with-arrow.html" rel="nofollow noreferrer">mapInPandas</a>.</p>
  <p>Expected output: mapped pandas_function() transforms one row to three, so output transformed_df should have 30k rows</p>
  <p>Current output: I'm getting 3 rows with 1 core and 24 rows with 8 cores.</p>
  <p>INPUT: respond_sdf has 10k rows</p>
  <pre><code>+-----+-------------------------------------------------------------------+     
  |url  |content                                                            |
  +-----+-------------------------------------------------------------------+
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  |api_1|{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }   |
  |api_2|{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }|
  +-----+-------------------------------------------------------------------+
  only showing top 20 rows
  Input respond_sdf has 10000 rows
  </code></pre>
  <p>OUTPUT A) 3 rows - with 1 core - .master('local [<a href="https://spark.apache.org/docs/latest/sql-pyspark-pandas-with-arrow.html" rel="nofollow noreferrer">1</a>]')</p>
  <pre><code>{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }    (0 + 1) / 1]
  +-----+---+---+                                                                 
  |  api|  A|  B|
  +-----+---+---+
  |api_1|  1|  4|
  |api_1|  2|  5|
  |api_1|  3|  6|
  +-----+---+---+
  
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
   Output transformed_df has 3 rows
  </code></pre>
  <p>OUTPUT B) 24 rows - with 8 cores - .master('local[8]')</p>
  <pre><code>{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }    (0 + 1) / 1]
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }                
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  +-----+---+---+
  |  api|  A|  B|
  +-----+---+---+
  |api_1|  1|  4|
  |api_1|  2|  5|
  |api_1|  3|  6|
  |api_1|  1|  4|
  |api_1|  2|  5|
  |api_1|  3|  6|
  |api_1|  1|  4|
  |api_1|  2|  5|
  |api_1|  3|  6|
  |api_1|  1|  4|
  |api_1|  2|  5|
  |api_1|  3|  6|
  |api_1|  1|  4|
  |api_1|  2|  5|
  |api_1|  3|  6|
  |api_1|  1|  4|
  |api_1|  2|  5|
  |api_1|  3|  6|
  |api_1|  1|  4|
  |api_1|  2|  5|
  +-----+---+---+
  only showing top 20 rows
  
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }    (3 + 5) / 8]
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
  {'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }
   Output transformed_df has 24 rows    
  </code></pre>
  <p>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[1]') \
      .getOrCreate()
  sc = spark.sparkContext
  
  ####### INPUT DATAFRAME WITH LIST OF JSONS ########################
  
  # Create list with 10k nested tuples(url,content)
  rdd_list = [('api_1',&quot;{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }&quot;),
              (' api_2', &quot;{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }&quot;)]*5000
  
  schema = StructType([
    StructField('url', StringType(), True),
    StructField('content', StringType(), True)
    ])
  
  #Create input dataframe with 10k rows
  jsons = sc.parallelize(rdd_list)
  respond_sdf = spark.createDataFrame(jsons, schema)
  respond_sdf.show(truncate=False)
  
  print(f'Input respond_sdf has {respond_sdf.count()} rows')
  
  ####### TRANSFORMATION DATAFRAME ########################
  
  # Pandas transformation function returning pandas dataframe
  def pandas_function(iter):
      for df in iter:
          print(df['content'][0])
          yield pd.DataFrame(eval(df['content'][0]))
  
  transformed_df = respond_sdf.mapInPandas(pandas_function, &quot;api string, A int, B int&quot;)
  transformed_df.show()
  print(f' Output transformed_df has {transformed_df.count()} rows')
  print(f'Expected output dataframe should has 30k rows')
  </code></pre>
  <p>Link to related discussion:
  <a href="https://stackoverflow.com/questions/65412532/how-to-yield-pandas-dataframe-rows-to-spark-dataframe">How to yield pandas dataframe rows to spark dataframe</a></p>