Objective

Join multiple input files using PySpark to produce an output DataFrame in the required format.

Requirements

Create a PySpark DataFrame that contains data about each of the councils from the files in the england_councils directory and enrich it by adding some additional columns from the property_avg_price.csv and property_sales_volume.csv files.

To achieve this, please use PySpark and complete the following four methods:

extract_councils() –

this method should use the files in the england_councils directory (available inside input_directory) as input. It should return a Spark DataFrame which combines the rows from all these files and contains following columns:

council – this column contains the data from the council column from the raw files;

county – this column contains the data from the county column from the raw files;

council_type – this would be a new column based on the file from which the row is taken:

Rows from district_councils.csv should have the value District Council;

Rows from london_boroughs.csv should have the value London Borough;

Rows from metropolitan_districts.csv should have the value Metropolitan District;

Rows from unitary_authorities.csv should have the value Unitary Authority.

extract_avg_price() –

this method should use the file property_avg_price.csv (available inside input_directory) as input and return a DataFrame containing the following two columns:

council – this column contains the data from the local_authority column from the raw files;

avg_price_nov_2019 – this column contains the data from the avg_price_nov_2019 column from the raw files.

extract_sales_volume() –

this method should use the file property_sales_volume.csv (available inside input_directory) as input and return a DataFrame containing the following two columns:

council – this column contains the data from the local_authority column from the raw files;

sales_volume_sep_2019 – this column contains the data from the sales_volume_sep_2019 column from the raw files.

transform() –

this method uses the output DataFrames from the three previous methods. It should produce a DataFrame which joins them together and contains the following columns: council, county, council_type, avg_price_nov_2019, sales_volume_sep_2019.

Please join the data in such a way that each council is still available in the output even if it does not have any columns populated.

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit, col

class CouncilsJob:
    def __init__(self):
        self.spark_session = (SparkSession.builder
                              .master("local[*]")
                              .appName("EnglandCouncilsJob")
                              .getOrCreate())
        self.input_directory = "data"

    def extract_councils(self):
       pass 
          
        
        
       

    def extract_avg_price(self):
        pass
       

    def extract_sales_volume(self):
        pass
           
    def transform(self, councils_df, avg_price_df, sales_volume_df):
       pass
        
        
       

    def run(self):
        return self.transform(self.extract_councils(),
                              self.extract_avg_price(),
                              self.extract_sales_volume())