How to manage options in PySpark more efficiently

user_2122502 0
Viewed 234

Let us consider following pySpark code

my_df = (spark.read.format("csv")
                     .option("header","true")
                     .option("inferSchema", "true")
                     .load(my_data_path))

This is a relatively small code, but sometimes we have codes with many options, where passing string options causes typos frequently. Also we don't get any suggestions from our code editors. As a workaround I am thinking to create a named tuple (or a custom class) to have all the options I need. For example,

from collections import namedtuple
allOptions = namedtuple("allOptions", "csvFormat header inferSchema")
sparkOptions = allOptions("csv", "header", "inferSchema")
my_df = (spark.read.format(sparkOptions.csvFormat)
                     .option(sparkOptions.header,"true")
                     .option(sparkOptions.inferSchema, "true")
                     .load(my_data_path))

I am wondering if there is downsides of this approach or if there is any better and standard approach used by the other pySpark developers.

pyspark apache-spark python-3.x design-patterns python
3 Answers

If you use .csv function to read the file, options are named arguments, thus it throws the TypeError. Also, on VS Code with Python plugin, the options would autocomplete.

df = spark.read.csv(my_data_path,
                    header=True,
                    inferSchema=True)

If I run with a typo, it throws the error.

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
/var/folders/tv/32xjg80x6pb_9t4909z8_hh00000gn/T/ipykernel_3636/4060466279.py in <module>
----> 1 df = spark.read.csv('test.csv', inferSchemaa=True, header=True)

TypeError: csv() got an unexpected keyword argument 'inferSchemaa'

On VS Code, options are suggested in autocomplete.

enter image description here

user_29561350

For that and many other reasons, in production level projects, we used to write a project to wrap spark.

So developers not allowed to deal with spark directly.

In such project we can :

  • Abstract options using enumerations and inheritance to avoid typos and incompatibles options.
  • Set default options for each data format and developers can overwrite them if needed, to reduce the amount of code written by the developer
  • Set and defines any repetitive code like frequently used data sources, default output data format, etc.
user_20372290

I think the best approach is to make a wrapper(s) with some default values and kwargs like this

def csv(path, inferSchema=True, header=True, options={}):
    return hdfs(path, 'csv', {'inferSchema': inferSchema, 'header': header, **options})

def parquet(path, options={}):
    return hdfs(path, 'parquet', {**options})

def hdfs(path, format, options={}):
    return (spark
        .read
        .format(format)
        .options(**options)
        .load(f'hdfs://.../{path}')
    )
user_34415100
Related