How to process RDDs using a Python class?

by

Problem here is a little bit more subtle than using nested RDDs or performing Spark actions inside of transformations. Spark doesn’t allow access to the SparkContext inside action or transformation.

Even you don’t access it explicitly it is referenced inside the closure and has to be serialized and carried around. It means that your transformation method, which references self, keeps SparkContext as well, hence the error.

One way to handle this is to use static method:

class model(object):
    @staticmethod
    def transformation_function(row):
        row = row.split(',')
        return row[0]+row[1]

    def __init__(self):
        self.data = sc.textFile('some.csv')

    def run_model(self):
        self.data = self.data.map(model.transformation_function)

Edit:

If you want to be able to access instance variables you can try something like this:

class model(object):
    @staticmethod
    def transformation_function(a_model):
        delim = a_model.delim
        def _transformation_function(row):
            return row.split(delim)
        return _transformation_function

    def __init__(self):
        self.delim = ','
        self.data = sc.textFile('some.csv')

    def run_model(self):
        self.data = self.data.map(model.transformation_function(self))

More Related Contents:

  1. How to find median and quantiles using Spark
  2. Spark Dataframe distinguish columns with duplicated name
  3. How to change dataframe column names in pyspark?
  4. I can’t seem to get –py-files on Spark to work
  5. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  6. Spark RDD to DataFrame python
  7. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  8. Why is Apache-Spark – Python so slow locally as compared to pandas?
  9. Spark groupByKey alternative
  10. How to extract an element from a array in pyspark
  11. PySpark create new column with mapping from a dict
  12. Tuning parameters for implicit pyspark.ml ALS matrix factorization model through pyspark.ml CrossValidator
  13. Spark iteration time increasing exponentially when using join
  14. Create Spark DataFrame. Can not infer schema for type
  15. PySpark DataFrames – way to enumerate without converting to Pandas?
  16. Getting Spark, Python, and MongoDB to work together
  17. How to transform data with sliding window over time series data in Pyspark
  18. Using UDF ignores condition in when
  19. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  20. pyspark parse fixed width text file
  21. Concatenating string by rows in pyspark
  22. What is the equivalent to scala.util.Try in pyspark?
  23. environment variables PYSPARK_PYTHON and PYSPARK_DRIVER_PYTHON
  24. Multiple condition filter on dataframe
  25. How to return a “Tuple type” in a UDF in PySpark?
  26. PySpark Throwing error Method __getnewargs__([]) does not exist
  27. How to connect HBase and Spark using Python?
  28. Pyspark 2.4.0, read avro from kafka with read stream – Python
  29. How to pivot on multiple columns in Spark SQL?
  30. Apache Spark — Assign the result of UDF to multiple dataframe columns

Leave a Comment