How many Iterators are there in Spark mapInPandas?

user_10236974 0
Viewed 583

I am trying to understand how "mapInPandas" works in Spark. The example quoted on the Databricks blog is:

from typing import Iterator
import pandas as pd

df = spark.createDataFrame([(1, 21), (2, 30)], ("id", "age"))

def pandas_filter(iterator: Iterator[pd.DataFrame]) -> Iterator[pd.DataFrame]:
    for pdf in iterator:
        yield pdf[pdf.id == 1]

df.mapInPandas(pandas_filter, schema=df.schema).show()

Question is, how many "pdf" are going to be in the iterator? I guessed that perhaps they would be as many as the number of partitions but when I further tested the code it seemed like they were far too many (on a different dataset with ~100 m records)

So is there a way to know how the number of iterations is determined and if there is a way to make it equal to the number of partitions?

pyspark azure-databricks databricks apache-spark
1 Answers

You can find that in documentation:

Data partitions in Spark are converted into Arrow record batches, which can temporarily lead to high memory usage in the JVM. To avoid possible out of memory exceptions, the size of the Arrow record batches can be adjusted by setting the conf “spark.sql.execution.arrow.maxRecordsPerBatch” to an integer that will determine the maximum number of rows for each batch. The default value is 10,000 records per batch. If the number of columns is large, the value should be adjusted accordingly. Using this limit, each data partition will be made into 1 or more record batches for processing.

so if you have 10M records, the you will have ~10,000 iterators

Alex_Ott_186270
Related