convert foreach to foreachpartition in spark
<p>I am working with foreach on spark RDD. Now I wanted to convert the same logic to foreachpartition.</p>

<p><strong>Here is my logic:</strong> </p>

<pre><code>batch = []
max = rdd.count()
processed = 0

def writeBatch(b):
    ///Push this batch to a database

def collect(row):
    dict={'name':row[0],'age':row[1}
    batch.append(dict)
    global processed
    processed += 1
    if len(batch) &gt;= 500 or processed &gt;= max:
        writeBatch(batch)



rdd.foreach(collect)
</code></pre>

<p>Here, I am using rdd.count() in the condition. How can I implement this using foreachPartition() effectively.</p>

<p>Problem is it is not a valid logic. It may work in some special cases but in general you'll loose data.</p>

<p>Inside <code>foreach</code> you mutate two objects <code>batch</code>, <code>processed</code> where neither one is truly global (for a whole cluster) or isolated per partition (unless you configured Spark to not to reuse interpreters).</p>

<p>Without using external dependencies a correct approach can be expressed like this:</p>

<pre><code>max_size = ...

def write_partition(items, max_size):
    batch = []
    size = 0

    # Process items in batches
    for item in items:
        batch.append(item)
        size += 1

        # Reached desired number of partitions
        if size == max_size:
            write_batch(batch)

            # Reset batch
            batch = []
            size = 0

    # If partition has been completed and there are still objects in batch 
    if batch:
        write_batch(batch)

rdd.foreachPartition(lambda items: write_partition(items, max_size))
</code></pre>

python apache-spark pyspark