How to run independent transformations in parallel using PySpark?

by

Just use threads and make sure that cluster have enough resources to process both tasks at the same time.

from threading import Thread
import time

def process(rdd, f):
    def delay(x):
        time.sleep(1)
        return f(x)
    return rdd.map(delay).sum()


rdd = sc.parallelize(range(100), int(sc.defaultParallelism / 2))

t1 = Thread(target=process, args=(rdd, lambda x: x * 2))
t2  = Thread(target=process, args=(rdd, lambda x: x + 1))
t1.start(); t2.start()

Arguably this is not that often useful in practice but otherwise should work just fine.

You can further use in-application scheduling with FAIR scheduler and scheduler pools for a better control over execution strategy.

You can also try pyspark-asyncactions (disclaimer – the author of this answer is also the author of the package) which provides a set of wrappers around Spark API and concurrent.futures:

import asyncactions
import concurrent.futures

f1 = rdd.filter(lambda x: x % 3 == 0).countAsync()
f2 = rdd.filter(lambda x: x % 11 == 0).countAsync()

[x.result() for x in concurrent.futures.as_completed([f1, f2])]

More Related Contents:

  1. Pyspark filter dataframe by columns of another dataframe
  2. SparkSQL on pyspark: how to generate time series?
  3. How to pivot Spark DataFrame?
  4. How to melt Spark DataFrame?
  5. Unpivot in spark-sql/pyspark
  6. Load CSV file with Spark
  7. How do I add a new column to a Spark DataFrame (using PySpark)?
  8. How to perform union on two DataFrames with different amounts of columns in spark?
  9. Does spark predicate pushdown work with JDBC?
  10. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  11. Spark Dataframe validating column names for parquet writes
  12. How to check if spark dataframe is empty?
  13. Filter Spark DataFrame based on another DataFrame that specifies denylist criteria
  14. Passing a data frame column and external list to udf under withColumn
  15. Updating a dataframe column in spark
  16. Pyspark : forward fill with last observation for a DataFrame
  17. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  18. Best way to get the max value in a Spark dataframe column
  19. Apache Spark Python Cosine Similarity over DataFrames
  20. Spark Error:expected zero arguments for construction of ClassDict (for numpy.core.multiarray._reconstruct)
  21. GroupBy column and filter rows with maximum value in Pyspark
  22. How to set timezone to UTC in Apache Spark?
  23. Rename more than one column using withColumnRenamed
  24. How to aggregate over rolling time window with groups in Spark
  25. Concatenate two PySpark dataframes
  26. Filtering DataFrame using the length of a column
  27. How to exclude multiple columns in Spark dataframe in Python
  28. Stratified sampling with pyspark
  29. How to turn off scientific notation in pyspark?
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment