<jupyter_start><jupyter_text>Example: Simple ETL lane===================== *Note: it is recommended to took a look at the [official documentation](https://sparklanes.readthedocs.io) first.*In this example, we'll build a simple ETL pipeline using sparklanes and the popular iris dataset. First, we'll load the dataset from CSV, before applying some transformations on it, to then finally dump it as JSON to disk. Let's start by importing all the libs we need:<jupyter_code>from pyspark.sql.functions import monotonically_increasing_id

from sparklanes import Task, Lane, conn<jupyter_output><empty_output><jupyter_text>And write our data processors, or `Tasks` next:<jupyter_code>@Task('extract_data')
class ExtractIrisCSVData(object):
    """Load the iris data set from a CSV file"""
    def __init__(self, iris_csv_path):
        self.iris_csv_path = iris_csv_path

    def extract_data(self):
        # Read the csv
        iris_df = conn.spark.read.csv(path=self.iris_csv_path,
                                      sep=',',
                                      header=True,
                                      inferSchema=True)

        # Make it available to tasks that follow
        self.cache('iris_df', iris_df)


@Task('add_index')
class AddRowIndex(object):
    """Add a index to each row in the data set"""
    def add_index(self):
        # Add id column
        self.iris_df = self.iris_df.withColumn('id', monotonically_increasing_id())

        # Update cache
        self.cache('iris_df', self.iris_df)


@Task('normalize')
class NormalizeColumns(object):
    """Normalize all numerical columns"""
    def normalize(self):
        # Add normalized columns
        columns = self.iris_df.columns
        columns.remove('species')
        for col in columns:
            col_min = float(self.iris_df.agg({col: "min"}).collect()[0]['min(%s)' % col])
            col_max = float(self.iris_df.agg({col: "max"}).collect()[0]['max(%s)' % col])
            self.iris_df = self.iris_df.withColumn(
                col + '_norm', (self.iris_df[col] - col_min) / (col_max - col_min)
            )

        # Update Cache
        self.cache('iris_df', self.iris_df)


@Task('write_to_json')
class SaveAsJSON(object):
    """Dump the data set as JSON to disk"""
    def __init__(self, output_folder):
        self.output_folder = output_folder

    def write_to_json(self):
        self.iris_df.write.format('json').save(self.output_folder)

        # Clear cache
        self.uncache('iris_df')<jupyter_output><empty_output><jupyter_text>Note how in the extractor class, we cache our `DataFrame` using `self.cache`, and as a result make it an attribute to all tasks that follow.With our Tasks being defined, we can now build the lane:<jupyter_code>lane = (Lane(name='IrisExampleLane', run_parallel=False)
        .add(ExtractIrisCSVData, iris_csv_path='data/iris.csv')
        .add(AddRowIndex)
        .add(NormalizeColumns)
        .add(SaveAsJSON, 'out'))<jupyter_output><empty_output><jupyter_text>And run it:<jupyter_code>lane.run()<jupyter_output>2018-06-07 11:52:01,466 - SPARKLANES - INFO - 
--------------------------------------------------------------------------------
Executing `IrisExampleLane`
--------------------------------------------------------------------------------

2018-06-07 11:52:01,468 - SPARKLANES - INFO - 
================================================================================
	IrisExampleLane
	 >Task_ExtractIrisCSVData
	 >Task_AddRowIndex
	 >Task_NormalizeColumns
	 >Task_SaveAsJSON
================================================================================
2018-06-07 11:52:01,469 - SPARKLANES - INFO - 
--------------------------------------------------------------------------------
Executing task `ExtractIrisCSVData.extract_data`
--------------------------------------------------------------------------------
2018-06-07 11:52:01,668 - SPARKLANES - INFO - 
--------------------------------------------------------------------------------
Finished executing task `ExtractIrisCSVData.extract_data`. [...]<jupyter_text>That works, but what makes `sparklanes` more useful, is the capability of defining processing lanes using *YAML configuration files*, to then submit these lanes to a spark cluster. We can define the same lane as above like: ```yamllane:  name: IrisExampleLane  run_parallel: false  tasks:    - class: tasks.iris.ExtractIrisCSVData      kwargs:        iris_csv_path: data/iris.csv    - class: tasks.iris.AddRowIndex    - class: tasks.iris.NormalizeColumns    - class: tasks.iris.SaveAsJSON      args:        - out``` With the file being saved as `iris.yml`, our directory structure looks like this: ```yamldata/    iris.csv  tasks/    __init__.py   Required to be recognized as a python package    iris.py   Contains our processor classes (Tasks)  iris.yml  ``` So to run our pipeline, we can submit it to spark using the `lane-submit` command line script:<jupyter_code>%%bash
lane-submit -y iris.yml -p tasks -e data -s master=local[2]<jupyter_output>2018-06-07 11:50:09,219 - SPARKLANES - INFO - 
--------------------------------------------------------------------------------
Executing `IrisExampleLane`
--------------------------------------------------------------------------------

2018-06-07 11:50:09,219 - SPARKLANES - INFO - 
================================================================================
	IrisExampleLane
	 >Task_ExtractIrisCSVData
	 >Task_AddRowIndex
	 >Task_NormalizeColumns
	 >Task_SaveAsJSON
================================================================================
2018-06-07 11:50:09,219 - SPARKLANES - INFO - 
--------------------------------------------------------------------------------
Executing task `ExtractIrisCSVData.extract_data`
--------------------------------------------------------------------------------
2018-06-07 11:50:13,784 - SPARKLANES - INFO - 
--------------------------------------------------------------------------------
Finished executing task `ExtractIrisCSVData.extract_data`. [...]
StarcoderdataJupyterStructuredCleanDedup