Repository: RADAR-base/radarpipeline (language: python)

Issue:
Publish radar-pipeline as a PIP package
After the next release of radarpipeline, we should aim at publishing it as a pip package. This will help us streamline the pipeline so that it can be used as a module by other researcher. Major tasks needed before publishing are --

- [x] #9 
- [x] #40 
- [x] support installing radar-pipeline using pip
- [x] read and execute local or remote config file
- [ ] show all the available modules in the Mock data analytics module
- [x] have a custom radar data reader module
- [x] #65 
- [x] #58 

Read config.yaml file from the remote repo
When reading a remote repo, it would make sense to read the config.yaml file available in that repo. 


Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index 3ee249c..c276c93 100644
--- a/.gitignore
+++ b/.gitignore
@@ -140,5 +140,6 @@ data/
 features/
 temp/
 output/
+tmp/
 
 .DS_STORE
diff --git a/config.yaml b/config.yaml
index 38aaf45..42faed2 100644
--- a/config.yaml
+++ b/config.yaml
@@ -3,9 +3,24 @@ project:
     description: mock_description
     version: mock_version
 
-input_data:
-    data_location: local
-    local_directory: mockdata/mockdata
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        # In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        # In case of s3, use the following format
+        #       aws_access_key_id:
+        #       aws_secret_access_key:
+        #       region_name:
+        #       s3_access_url:
+        #       bucket:
+        #       prefix:
+        # In case of local or Mock, use the following format
+        source_path: mockdata/mockdata
     data_format: csv
 
 configurations:
@@ -19,8 +34,13 @@ features:
       feature_names:
         - all
 
-output_data:
-    output_location: local
-    local_directory: output/mockdata
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
     data_format: csv
     compress: false
+
+spark_config:
+    spark.executor.instances: 2
+    spark.driver.memory: 13G
\ No newline at end of file
diff --git a/config.yaml.template b/config.yaml.template
new file mode 100644
index 0000000..e25b4e7
--- /dev/null
+++ b/config.yaml.template
@@ -0,0 +1,46 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        ## In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        ## In case of s3, use the following format
+        # aws_access_key_id:
+        # aws_secret_access_key:
+        # region_name:
+        # s3_access_url:
+        # bucket:
+        # prefix:
+        ## In case of local or Mock, use the following format
+        # source_path: mockdata/mockdata
+    data_format: csv
+
+configurations:
+    df_type: 'pandas' # can be pandas or spark
+
+features:
+    - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
+      branch: main
+      feature_groups:
+          - MockFeatureGroup
+      feature_names:
+        - all
+
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
+    data_format: csv
+    compress: false
+
+spark_config:
+    spark.executor.instances: 6
+    spark.driver.memory: 10G
\ No newline at end of file
diff --git a/radarpipeline/common/utils.py b/radarpipeline/common/utils.py
index 42a78c9..1e393de 100644
--- a/radarpipeline/common/utils.py
+++ b/radarpipeline/common/utils.py
@@ -2,12 +2,17 @@
 from functools import reduce
 from typing import Any, Dict, List
 from urllib.parse import urlparse
+import pathlib
 from pathlib import Path
 
 import pyspark.sql as ps
 import requests
 import yaml
-from strictyaml import load, Map, Str, Seq, Bool, Optional, YAMLError, CommaSeparated
+from strictyaml import load, Map, Int, Str, Seq, Bool, Optional
+from strictyaml import YAMLError, CommaSeparated, MapPattern
+
+import ntpath
+import posixpath
 
 from radarpipeline.common import constants
 
@@ -139,9 +144,9 @@ def get_yaml_schema() -> Map:
             Optional("description"): Str(),
             Optional("version"): Str()
         }),
-        "input_data": Map({
-            "data_location": Str(),
-            "local_directory": Seq(Str()) | Str(),
+        "input": Map({
+            "data_type": Str(),
+            "config": MapPattern(Str(), Str()),
             "data_format": Str()
         }),
         "configurations": Map({
@@ -153,11 +158,56 @@ def get_yaml_schema() -> Map:
             "feature_groups": Seq(Str()),
             "feature_names": Seq(CommaSeparated(Str()))
         })),
-        "output_data": Map({
+        "output": Map({
             "output_location": Str(),
-            "local_directory": Str(),
+            "config": MapPattern(Str(), Str()),
             "data_format": Str(),
             "compress": Bool()
-        })
+        }),
+        Optional("spark_config"): Map({
+            Optional("spark.executor.instances", default=4): Int(),
+            Optional("spark.executor.cores", default=4): Int(),
+            Optional("spark.executor.memory", default='10g'): Str(),
+            Optional("spark.driver.memory", default='15g'): Str(),
+            Optional("spark.memory.offHeap.enabled", default=True): Bool(),
+            Optional("spark.memory.offHeap.size", default='20g'): Str(),
+            Optional("spark.driver.maxResultSize", default='0'): Str(),
+        }),
     })
     return schema
+
+
+def get_absolute_path(path: str) -> str:
+    """
+    Returns the absolute path of the path
+    Parameters
+    ----------
+    path: str
+        Path to be converted to absolute path
+    Returns
+    -------
+    str
+        Absolute path of the path
+    """
+    if not os.path.isabs(path):
+        pipeline_dir = pathlib.Path(__file__).parent.parent.parent.resolve()
+        path = os.path.join(pipeline_dir, path)
+    return path
+
+
+def reparent(newparent, oldpath):
+    '''when copying or moving a directory structure, you need to re-parent the
+    oldpath.  When using os.path.join to calculate this new path, the
+    appearance of a / root path at the beginning of oldpath, supplants the
+    newparent and we don't want this to happen, so we need to make the oldpath
+    root appear as a child of the newparent.
+
+    :param: str newparent: the new parent location for oldpath (target)
+    :param str oldpath: the path being adopted by newparent (source)
+
+    :returns: (str) resulting adoptive path
+    '''
+
+    if oldpath[0] in (posixpath.sep, ntpath.sep):
+        oldpath = '.' + oldpath
+    return os.path.join(newparent, oldpath)
diff --git a/radarpipeline/io/__init__.py b/radarpipeline/io/__init__.py
index 028152f..dade1ac 100644
--- a/radarpipeline/io/__init__.py
+++ b/radarpipeline/io/__init__.py
@@ -1,3 +1,4 @@
 from radarpipeline.io.abc import DataReader, SchemaReader
-from radarpipeline.io.reader import AvroSchemaReader, SparkCSVDataReader
+from radarpipeline.io.reader import AvroSchemaReader, SparkCSVDataReader, Reader
+from radarpipeline.io.downloader import SftpDataReader
 from radarpipeline.io.writer import *
diff --git a/radarpipeline/io/connection.py b/radarpipeline/io/connection.py
new file mode 100644
index 0000000..c35cd15
--- /dev/null
+++ b/radarpipeline/io/connection.py
@@ -0,0 +1,250 @@
+import paramiko
+from paramiko import AgentKey, RSAKey, DSSKey
+from paramiko import SSHException, AuthenticationException
+
+from typing import Any, Dict, List, Optional, Union
+import json
+import logging
+import os
+from radarpipeline.common.utils import reparent
+
+from contextlib import contextmanager
+from stat import S_IMODE, S_ISDIR, S_ISREG
+import warnings
+
+logger = logging.getLogger(__name__)
+
+
+class ConnectionException(Exception):
+    """Exception raised for connection problems
+
+    Attributes:
+        message  -- explanation of the error
+    """
+
+    def __init__(self, host, port):
+        # Call the base class constructor with the parameters it needs
+        Exception.__init__(self, host, port)
+        self.message = 'Could not connect to host:port.  %s:%s'
+
+
+class CredentialException(Exception):
+    """Exception raised for credential problems
+
+    Attributes:
+        message  -- explanation of the error
+    """
+
+    def __init__(self, message):
+        # Call the base class constructor with the parameters it needs
+        Exception.__init__(self, message)
+        self.message = message
+
+
+class HostKeysException(Exception):
+    '''raised when a problem with HostKeys is encountered'''
+    pass
+
+
+class SftpConnector():
+    """
+    Class for reading data from SFTP
+    """
+    def __init__(self, config_dict: Dict, variables: List[str]):
+        """
+        Parameters
+        ----------
+        config_dict : Dict
+            Dictionary containing configuration for SFTP
+        variables : List
+            List of variables to read
+        root_dir : str, optional
+        """
+
+        self.config_dict = config_dict
+        self.variables = variables
+
+    def connect(self):
+        """
+        Connect to SFTP server
+        """
+        username = self.config_dict.get('sftp_username')
+        host = self.config_dict.get('sftp_host')
+        password = self.config_dict.get('sftp_password')
+        port = self.config_dict.get('sftp_port', 22)
+        private_key = self.config_dict.get('sftp_private_key')
+        self._tconnect = {'username': username, 'password': password, 'hostkey': None,
+                          'pkey': None}
+        self._sftp_live = False
+        self._transport = None
+        self._transport = paramiko.Transport((host, port))
+        self._transport.use_compression(False)
+        private_key_pass = None
+        self._set_authentication(password, private_key, private_key_pass)
+        self._transport.connect(**self._tconnect)
+        self._sftp_connect()
+
+    def _set_authentication(self, password, private_key, private_key_pass):
+        '''Authenticate the transport. prefer password if given'''
+        if password is None:
+            # Use Private Key.
+            if not private_key:
+                # Try to use default key.
+                if os.path.exists(os.path.expanduser('~/.ssh/id_rsa')):
+                    private_key = '~/.ssh/id_rsa'
+                elif os.path.exists(os.path.expanduser('~/.ssh/id_dsa')):
+                    private_key = '~/.ssh/id_dsa'
+                else:
+                    raise CredentialException("No password or key specified.")
+
+            if isinstance(private_key, (AgentKey, RSAKey)):
+                # use the paramiko agent or rsa key
+                self._tconnect['pkey'] = private_key
+            else:
+                # isn't a paramiko AgentKey or RSAKey, try to build a
+                # key from what we assume is a path to a key
+                private_key_file = os.path.expanduser(private_key)
+                try:  # try rsa
+                    self._tconnect['pkey'] = RSAKey.from_private_key_file(
+                        private_key_file, private_key_pass)
+                except paramiko.SSHException:   # if it fails, try dss
+                    # pylint:disable=r0204
+                    self._tconnect['pkey'] = DSSKey.from_private_key_file(
+                        private_key_file, private_key_pass)
+
+    def _sftp_connect(self):
+        """Establish the SFTP connection."""
+        if not self._sftp_live:
+            self._sftp = paramiko.SFTPClient.from_transport(self._transport)
+            self._sftp_live = True
+
+    @property
+    def pwd(self):
+        '''return the current working directory
+
+        :returns: (str) current working directory
+
+        '''
+        self._sftp_connect()
+        return self._sftp.normalize('.')
+
+    @contextmanager
+    def cd(self, remotepath=None):  # pylint:disable=c0103
+        """context manager that can change to a optionally specified remote
+        directory and restores the old pwd on exit.
+
+        :param str|None remotepath: *Default: None* -
+            remotepath to temporarily make the current directory
+        :returns: None
+        :raises: IOError, if remote path doesn't exist
+        """
+        original_path = self.pwd
+        try:
+            if remotepath is not None:
+                self.chdir(remotepath)
+            yield
+        finally:
+            self.chdir(original_path)
+
+    def chdir(self, remotepath):
+        """change the current working directory on the remote
+
+        :param str remotepath: the remote path to change to
+
+        :returns: None
+
+        :raises: IOError, if path does not exist
+
+        """
+        self._sftp_connect()
+        self._sftp.chdir(remotepath)
+
+    def listdir(self, remotepath='.'):
+        """return a list of files/directories for the given remote path.
+        Unlike, paramiko, the directory listing is sorted.
+
+        :param str remotepath: path to list on the server
+
+        :returns: (list of str) directory entries, sorted
+
+        """
+        self._sftp_connect()
+        return sorted(self._sftp.listdir(remotepath))
+
+    def isfile(self, remotepath):
+        """return true if remotepath is a file
+
+        :param str remotepath: the path to test
+
+        :returns: (bool)
+
+        """
+        self._sftp_connect()
+        try:
+            result = S_ISREG(self._sftp.stat(remotepath).st_mode)
+        except IOError:     # no such file
+            result = False
+        return result
+
+    def get(self, remotepath, localpath=None, callback=None,
+            preserve_mtime=False):
+        """Copies a file between the remote host and the local host.
+
+        :param str remotepath: the remote path and filename, source
+        :param str localpath:
+            the local path and filename to copy, destination. If not specified,
+            file is copied to local current working directory
+        :param callable callback:
+            optional callback function (form: ``func(int, int)``) that accepts
+            the bytes transferred so far and the total bytes to be transferred.
+        :param bool preserve_mtime:
+            *Default: False* - make the modification time(st_mtime) on the
+            local file match the time on the remote. (st_atime can differ
+            because stat'ing the localfile can/does update it's st_atime)
+
+        :returns: None
+
+        :raises: IOError
+
+        """
+        if not localpath:
+            localpath = os.path.split(remotepath)[1]
+
+        self._sftp_connect()
+        if preserve_mtime:
+            sftpattrs = self._sftp.stat(remotepath)
+
+        self._sftp.get(remotepath, localpath, callback=callback)
+        if preserve_mtime:
+            os.utime(localpath, (sftpattrs.st_atime, sftpattrs.st_mtime))
+
+    def get_d(self, remotedir, localdir, preserve_mtime=False):
+        """get the contents of remotedir and write to locadir. (non-recursive)
+
+        :param str remotedir: the remote directory to copy from (source)
+        :param str localdir: the local directory to copy to (target)
+        :param bool preserve_mtime: *Default: False* -
+            preserve modification time on files
+
+        :returns: None
+
+        :raises:
+        """
+        self._sftp_connect()
+        with self.cd(remotedir):
+            for sattr in self._sftp.listdir_attr('.'):
+                if S_ISREG(sattr.st_mode):
+                    rname = sattr.filename
+                    self.get(rname, reparent(localdir, rname),
+                             preserve_mtime=preserve_mtime)
+
+    def close(self):
+        """Closes the connection and cleans up."""
+        # Close SFTP Connection.
+        if self._sftp_live:
+            self._sftp.close()
+            self._sftp_live = False
+        # Close the SSH Transport.
+        if self._transport:
+            self._transport.close()
+            self._transport = None
diff --git a/radarpipeline/io/downloader.py b/radarpipeline/io/downloader.py
new file mode 100644
index 0000000..9262846
--- /dev/null
+++ b/radarpipeline/io/downloader.py
@@ -0,0 +1,130 @@
+import json
+import logging
+import os
+from typing import Any, Dict, List, Optional, Union
+from multiprocessing import Pool
+from functools import partial
+from datetime import datetime
+from radarpipeline.io.connection import SftpConnector
+
+logger = logging.getLogger(__name__)
+
+
+class SftpDataReader():
+    """
+    Class for reading data from SFTP
+    """
+    def __init__(self, config_dict: Dict, variables: List[str]):
+        """
+        Parameters
+        ----------
+        config_dict : Dict
+            Dictionary containing configuration for SFTP
+        variables : List
+            List of variables to read
+        root_dir : str, optional
+        """
+
+        self.config_dict = config_dict
+        self.variables = variables
+        if "sftp_target_path" not in config_dict:
+            now = datetime.now()
+            current_time = now.strftime("%d-%m-%y::%H:%M:%S")
+            self.root_dir = os.path.join("tmp", current_time)
+        else:
+            self.root_dir = config_dict["sftp_target_path"]
+
+    def read_sftp_data(self) -> None:
+        sftp_connection_args = {}
+        sftp_source_path = self.config_dict.get('sftp_source_path')
+        sftp_connection_args["username"] = self.config_dict.get('sftp_username')
+        sftp_connection_args["host"] = self.config_dict.get('sftp_host')
+        sftp_connection_args["private_key"] = self.config_dict.get('sftp_private_key')
+        all_participants_ids = self._get_all_id_sftp(sftp_source_path)
+        func = partial(self._fetch_data, self.root_dir, sftp_source_path,
+                       self.variables)
+        try:
+            func = partial(self._fetch_data, self.root_dir, sftp_source_path,
+                           self.variables)
+            with Pool(os.cpu_count()) as p:
+                p.map(func, all_participants_ids)
+        except Exception as e:
+            logger.warn(f"Cannot use parallel processing to download the data from \
+                sftp.Error: {e}")
+            logger.warn("Downloading the data from sftp sequentially. \
+                        This may take a while...")
+            for uid in all_participants_ids:
+                self._fetch_data(self.root_dir, sftp_source_path, self.variables, uid)
+        logger.info(f"Data read from sftp and stored in {self.root_dir} folder")
+        logger.info("To avoid redownloading, change config file to read from local")
+
+    def get_root_dir(self) -> str:
+        return self.root_dir
+
+    def _fetch_data(self, root_path, sftp_source_path, included_var_cat, uid):
+        sftp = SftpConnector(self.config_dict, included_var_cat)
+        sftp.connect()
+        try:
+            with sftp.cd(os.path.join(sftp_source_path, uid)):
+                source_folders = sftp.listdir(".")
+                for src in source_folders:
+                    if self._is_src_in_category(src, included_var_cat):
+                        dir_path = os.path.join(uid, src)
+                        os.makedirs(os.path.join(root_path, dir_path),
+                                    exist_ok=True)
+                        try:
+                            with sftp.cd(src):
+                                # To add new files only
+                                src_files = sftp.listdir(".")
+                                for src_file in src_files:
+                                    # check if src file is a file or directory
+                                    if sftp.isfile(src_file):
+                                        sftp.get(src_file,
+                                                 os.path.join(
+                                                     root_path, dir_path, src_file),
+                                                 preserve_mtime=True)
+                                    else:
+                                        if not os.path.exists(
+                                            os.path.join(
+                                                root_path, dir_path, src_file
+                                            )
+                                        ):
+                                            os.makedirs(
+                                                os.path.join(
+                                                    root_path, dir_path,
+                                                    src_file),
+                                                exist_ok=True)
+                                            sftp.get_d(src_file,
+                                                       os.path.join(
+                                                           root_path,
+                                                           dir_path,
+                                                           src_file),
+                                                       preserve_mtime=True)
+                        except FileNotFoundError:
+                            print("Folder not found: " + dir_path + "/" + src_file)
+                            continue
+                        except EOFError:
+                            print("EOFError: " + dir_path + "/" + src_file)
+                            continue
+        except FileNotFoundError:
+            print("Folder not found: " + uid)
+            return
+        sftp.close()
+
+    def _is_src_in_category(self, src, categories):
+        if categories == "all":
+            return True
+        for category in categories:
+            if src[:len(category)] == category:
+                return True
+        return False
+
+    def _get_all_id_sftp(self, sftp_source_path):
+        sftp = SftpConnector(self.config_dict, self.variables)
+        sftp.connect()
+        print(sftp.listdir())
+        with sftp.cd(sftp_source_path + '/'):
+            print(sftp.listdir())
+            ids = [x for x in sftp.listdir() if x[0] != "."]
+        sftp.close()
+        return ids
diff --git a/radarpipeline/io/reader.py b/radarpipeline/io/reader.py
index 5ff0796..714550e 100644
--- a/radarpipeline/io/reader.py
+++ b/radarpipeline/io/reader.py
@@ -11,6 +11,10 @@
 from radarpipeline.datalib import RadarData, RadarUserData, RadarVariableData
 from radarpipeline.io.abc import DataReader, SchemaReader
 
+from multiprocessing import Pool
+from functools import partial
+from datetime import datetime
+
 logger = logging.getLogger(__name__)
 
 
@@ -19,11 +23,23 @@ class SparkCSVDataReader(DataReader):
     Read CSV data from local directory using pySpark
     """
 
-    def __init__(self, config: Dict, required_data: List[str], df_type: str = "pandas"):
+    def __init__(self, config: Dict, required_data: List[str], df_type: str = "pandas",
+                 spark_config: Dict = {}):
         super().__init__(config)
+        default_spark_config = {'spark.executor.instances': 6,
+                                'spark.driver.memory': '10G',
+                                'spark.executor.cores': 4,
+                                'spark.executor.memory': '10g',
+                                'spark.memory.offHeap.enabled': True,
+                                'spark.memory.offHeap.size': '20g',
+                                'spark.driver.maxResultSize': '0'}
+
         self.required_data = required_data
         self.df_type = df_type
-        self.source_path = self.config.get("local_directory", "")
+        self.source_path = self.config['config'].get("source_path", "")
+        self.spark_config = default_spark_config
+        if spark_config is not None:
+            self.spark_config.update(spark_config)
         self.spark = self._initialize_spark_session()
 
     def _initialize_spark_session(self) -> ps.SparkSession:
@@ -63,13 +79,20 @@ def _initialize_spark_session(self) -> ps.SparkSession:
         """
         spark = (
             SparkSession.builder.master("local").appName("radarpipeline")
-            .config('spark.executor.instances', 4)
-            .config('spark.executor.cores', 4)
-            .config('spark.executor.memory', '10g')
-            .config('spark.driver.memory', '15g')
-            .config('spark.memory.offHeap.enabled', True)
-            .config('spark.memory.offHeap.size', '20g')
-            .config('spark.driver.maxResultSize', '0')
+            .config('spark.executor.instances',
+                    self.spark_config['spark.executor.instances'])
+            .config('spark.executor.cores',
+                    self.spark_config['spark.executor.cores'])
+            .config('spark.executor.memory',
+                    self.spark_config['spark.executor.memory'])
+            .config('spark.driver.memory',
+                    self.spark_config['spark.driver.memory'])
+            .config('spark.memory.offHeap.enabled',
+                    self.spark_config['spark.memory.offHeap.enabled'])
+            .config('spark.memory.offHeap.size',
+                    self.spark_config['spark.memory.offHeap.size'])
+            .config('spark.driver.maxResultSize',
+                    self.spark_config['spark.driver.maxResultSize'])
             .getOrCreate()
         )
 
@@ -82,9 +105,11 @@ def _initialize_spark_session(self) -> ps.SparkSession:
 
         spark.sparkContext.setLogLevel("ERROR")
         logger.info("Spark Session created")
-
         return spark
 
+    def close_spark_session(self):
+        self.spark.stop()
+
     def read_data(self) -> RadarData:
         """
         Reads RADAR data from local CSV files
@@ -433,3 +458,46 @@ def _get_superior_spark_type(self, spark_data_type_list: List[Any]) -> Any:
                 f"Conflicting types: {spark_data_type_list}. Returning String type."
             )
             return constants.STRING_TYPE
+
+
+class Reader():
+    '''
+    Class for reading data from a file
+    Reader(data_type : str, data_path: str, variables: Union[str, List])
+    reader = Reader(...)
+    reader.get_data(variables=Union[List, str])
+    reader.get_user_data(user_id=..)
+    '''
+    def __init__(self, data_type: str, data_path: str, variables: Union[str, List]):
+        '''
+        Parameters : data_type : str, data_path: str, variables: Union[str, List]
+        data_type : str
+            Type of data to be read
+            Only supports csv for now
+        data_path : str
+            Path to the data directory
+        variables : Union[str, List]
+            List of variables to be read
+        '''
+        self.data_type = data_type
+        self.data_path = data_path
+        # check if variables is a str
+        # If so, convert it to a list
+        if isinstance(variables, str):
+            variables = [variables]
+        self.variables = variables
+        config_dict = {"local_directory": self.data_path}
+        # check if data_type is csv
+        if self.data_type == 'csv':
+            self.reader_class = SparkCSVDataReader(config_dict, self.variables)
+        else:
+            raise NotImplementedError("Only csv data type is supported for now")
+
+    def read_data(self):
+        self.data = self.reader_class.read_data()
+
+    def get_data(self, variables: Union[List, str]) -> RadarData:
+        return self.data.get_combined_data_by_variable(variables)
+
+    def get_user_data(self, user_id: str) -> RadarData:
+        return self.data.get_data_by_user_id(user_id)
diff --git a/radarpipeline/project/project.py b/radarpipeline/project/project.py
index 492ec27..edd3126 100644
--- a/radarpipeline/project/project.py
+++ b/radarpipeline/project/project.py
@@ -12,6 +12,8 @@
 from radarpipeline.common import utils
 from radarpipeline.features import Feature, FeatureGroup
 from radarpipeline.io import PandasDataWriter, SparkCSVDataReader, SparkDataWriter
+from radarpipeline.io import SftpDataReader
+from radarpipeline.project.validations import ConfigValidator
 from strictyaml import load, YAMLError
 
 logger = logging.getLogger(__name__)
@@ -30,16 +32,47 @@ def __init__(self, input_data: Union[str, dict]) -> None:
 
         self.valid_input_formats = ["csv", "csv.gz"]
         self.valid_output_formats = ["csv"]
-        self.input_data = input_data
+        self.input_data = self._resolve_input_data(input_data)
         self.feature_path = os.path.abspath(
             os.path.join("radarpipeline", "features", "features")
         )
         self.features = {}
         self.config = self._get_config()
-        self._validate_config()
+        self.validator = ConfigValidator(self.config, self.valid_input_formats,
+                                         self.valid_output_formats)
+        self.validator.validate()
         self.feature_groups = self._get_feature_groups()
         self.total_required_data = self._get_total_required_data()
 
+    def _resolve_input_data(self, input_data) -> str:
+        """
+        Resolves the input_data variable and check
+        if it is a local path or a remote github url.
+        Returns:
+            str: Path to the config file
+        """
+        # checking if input_data is a local path
+        if isinstance(input_data, str):
+            if os.path.exists(input_data):
+                return input_data
+            elif utils.is_valid_github_path(input_data):
+                # get config.yaml from github
+                try:
+                    repo_name = utils.get_repo_name_from_url(input_data)
+                    cache_dir = os.path.join(
+                        os.path.expanduser("~"), ".cache", "radarpipeline", repo_name
+                    )
+                    if not os.path.exists(cache_dir):
+                        Repo.clone_from(input_data, cache_dir)
+                    input_data = os.path.join(cache_dir, "config.yaml")
+                    return input_data
+                except GitCommandError as err:
+                    logger.error(f"Error while cloning the repo: {err}")
+                    sys.exit(1)
+            else:
+                raise ValueError(f"File or URL does not exist: {input_data}")
+        return input_data
+
     def _get_config(self) -> Dict[str, Any]:
         """
         Reads the config and returns it as a dictionary
@@ -64,240 +97,6 @@ def _get_config(self) -> Dict[str, Any]:
         logger.info("Config file read successfully")
         return config
 
-    def _validate_config(self) -> None:
-        """
-        Validates the config to check if all the keys are present
-        """
-
-        required_keys = ["project", "input_data", "features", "output_data"]
-        for key in required_keys:
-            if key not in self.config:
-                raise ValueError(f"Key not present in the config: {key}")
-
-        self._validate_input()
-        self._validate_configurations()
-        self._validate_features()
-        self._validate_output()
-
-        logger.info("Config file validated successfully")
-
-    def _validate_input(self) -> None:
-        """
-        Validates the input data config
-        """
-
-        if self.config["input_data"]["data_location"] == "sftp":
-            sftp_config_keys = [
-                "sftp_host",
-                "sftp_username",
-                "sftp_directory",
-                "sftp_private_key",
-            ]
-            for key in sftp_config_keys:
-                if key not in self.config["input_data"]:
-                    raise ValueError(f"Key not present in the config: {key}")
-
-        elif self.config["input_data"]["data_location"] == "local":
-            if "local_directory" not in self.config["input_data"]:
-                raise ValueError("Key not present in the config: local_directory")
-            else:
-                # Check if local_directory is absolute path. If not, then set it.
-                local_directory = self.config["input_data"]["local_directory"]
-                if isinstance(local_directory, list):
-                    local_directory = [
-                        self._get_absolute_path(local_path)
-                        for local_path in local_directory]
-                    for local_path in local_directory:
-                        if not os.path.exists(local_path):
-                            raise ValueError(f"Path does not exist: {local_path}")
-                else:
-                    local_directory = self._get_absolute_path(local_directory)
-                    if not os.path.exists(local_directory):
-                        raise ValueError(f"Path does not exist: {local_directory}")
-                self.config["input_data"]["local_directory"] = local_directory
-
-            # Raise error if data_format it not valid input formats
-            if self.config["input_data"]["data_format"] not in self.valid_input_formats:
-                raise ValueError("Invalid value for key in input_data: data_format")
-
-        elif self.config["input_data"]["data_location"] == "mock":
-            self._update_mock_data()
-
-        else:
-            raise ValueError("Invalid value for the key: data_location")
-
-    def _validate_configurations(self) -> None:
-        """
-        Validates the configurations config
-        """
-
-        if "configurations" not in self.config:
-            self.config["configurations"] = {}
-
-        valid_df_types = ["pandas", "spark"]
-
-        if "df_type" not in self.config["configurations"]:
-            self.config["configurations"]["df_type"] = "pandas"
-        elif self.config["configurations"]["df_type"] not in valid_df_types:
-            raise ValueError("Invalid value for the key: df_type")
-
-    def _validate_features(self) -> None:
-        """
-        Validates the features config
-        """
-
-        if len(self.config["features"]) == 0:
-            raise ValueError("features array cannot be empty")
-
-        for index, feature in enumerate(self.config["features"]):
-            if "location" not in feature:
-                raise ValueError(
-                    f"Key not present in the config: location at index {index}"
-                )
-
-            if "feature_groups" not in feature:
-                raise ValueError(
-                    f"Key not present in the config: feature_groups at index {index}"
-                )
-
-            if len(feature["feature_groups"]) == 0:
-                raise ValueError(
-                    f"feature_groups array cannot be empty at index {index}"
-                )
-            if "feature_names" not in feature:
-                raise ValueError(
-                    f"Key not present in the config: feature_groups at index {index}"
-                )
-
-            if len(feature["feature_names"]) == 0:
-                raise ValueError(
-                    f"feature_groups array cannot be empty at index {index}"
-                )
-            feature_location = feature["location"]
-
-            if utils.is_valid_github_path(feature_location):
-                repo_name = utils.get_repo_name_from_url(feature_location)
-                cache_dir = os.path.join(
-                    os.path.expanduser("~"), ".cache", "radarpipeline", repo_name
-                )
-
-                if not os.path.exists(cache_dir):
-                    Repo.clone_from(feature_location, cache_dir)
-
-                repo = Repo(cache_dir)
-                repo.git.reset("--hard")
-                repo.git.clean("-xdf")
-
-                active_branch_name = repo.active_branch.name
-                feature_branch = feature.get("branch", active_branch_name)
-
-                try:
-                    repo.git.checkout(feature_branch)
-                except GitCommandError:
-                    logger.warning(
-                        "Branch %s does not exist. Using the %s branch instead.",
-                        feature_branch,
-                        repo.active_branch.name,
-                    )
-                    feature_branch = repo.active_branch.name
-                    repo.git.checkout(feature_branch)
-                repo.remotes.origin.pull(feature_branch)
-
-                feature_location = cache_dir
-                logger.info(f"Using feature from cache location: {feature_location}")
-            else:
-                feature_location = os.path.expanduser(feature_location)
-                if not os.path.isdir(feature_location):
-                    raise ValueError(f"Invalid feature location: {feature_location}")
-                logger.info(f"Using feature from local path: {feature_location}")
-
-            self.config["features"][index]["location"] = feature_location
-
-    def _validate_output(self) -> None:
-        """
-        Validates the output data config
-        """
-
-        if self.config["output_data"]["output_location"] == "postgres":
-            postgres_config_keys = [
-                "postgres_host",
-                "postgres_username",
-                "postgres_database",
-                "postgres_password",
-                "postgres_table",
-            ]
-            for key in postgres_config_keys:
-                if key not in self.config["output_data"]:
-                    raise ValueError(f"Key not present in the config: {key}")
-
-        elif self.config["output_data"]["output_location"] == "local":
-            if "local_directory" not in self.config["output_data"]:
-                raise ValueError("Key not present in the config: local_directory")
-            else:
-                # Check if local_directory is absolute path. If not, then set it.
-                local_directory = self.config["output_data"]["local_directory"]
-                local_directory = self._get_absolute_path(local_directory)
-                if not os.path.exists(local_directory):
-                    os.makedirs(local_directory, exist_ok=True)
-                self.config["output_data"]["output_directory"] = local_directory
-
-            # Raise error if data_format it not valid output formats
-            if (
-                self.config["output_data"]["data_format"]
-                not in self.valid_output_formats
-            ):
-                raise ValueError("Invalid value for key in output_data: data_format")
-
-            if "compress" not in self.config["output_data"]:
-                self.config["output_data"]["compress"] = False
-            if self.config["output_data"]["compress"] == "true":
-                self.config["output_data"]["compress"] = True
-
-        elif self.config["output_data"]["output_location"] == "mock":
-            pass
-
-        else:
-            raise ValueError("Key not present in the config: output_location")
-
-    def _get_absolute_path(self, path: str) -> str:
-        """
-        Returns the absolute path of the path
-
-        Parameters
-        ----------
-        path: str
-            Path to be converted to absolute path
-
-        Returns
-        -------
-        str
-            Absolute path of the path
-        """
-
-        if not os.path.isabs(path):
-            pipeline_dir = pathlib.Path(__file__).parent.parent.parent.resolve()
-            path = os.path.join(pipeline_dir, path)
-        return path
-
-    def _update_mock_data(self) -> None:
-        """
-        Updates the mock data submodule of the project
-        """
-
-        repo = Repo(
-            os.path.dirname(os.path.abspath(__file__)),
-            search_parent_directories=True,
-        )
-        sms = repo.submodules
-        try:
-            mock_data_submodule = sms["mockdata"]
-        except IndexError:
-            raise ValueError("mockdata submodule not found in the repository")
-        if not (mock_data_submodule.exists() and mock_data_submodule.module_exists()):
-            logger.info("Mock data submodule not found. Cloning it...")
-            repo.git.submodule("update", "--init", "--recursive")
-            logger.info("Mock data input cloned")
-
     def _get_feature_groups(self) -> List[FeatureGroup]:
         """
         Get all the feature groups from the config
@@ -419,29 +218,54 @@ def fetch_data(self) -> None:
         """
         Fetches the data from the data source
         """
+        if 'spark_config' not in self.config:
+            self.config['spark_config'] = {}
 
-        if self.config["input_data"]["data_location"] == "local":
-            if self.config["input_data"]["data_format"] in self.valid_input_formats:
-                self.data = SparkCSVDataReader(
-                    self.config["input_data"],
+        if self.config["input"]["data_type"] == "local":
+            if self.config["input"]["data_format"] in self.valid_input_formats:
+                sparkcsvdatareader = SparkCSVDataReader(
+                    self.config["input"],
                     self.total_required_data,
                     self.config["configurations"]["df_type"],
-                ).read_data()
+                    self.config['spark_config']
+                )
+                self.data = sparkcsvdatareader.read_data()
+                sparkcsvdatareader.close_spark_session()
             else:
                 raise ValueError("Wrong data format")
 
-        elif self.config["input_data"]["data_location"] == "mock":
+        elif self.config["input"]["data_type"] == "mock":
             mock_config_input = {
-                "local_directory": os.path.join("mockdata", "mockdata")
+                "config": {
+                    "source_path": os.path.join("mockdata", "mockdata")
+                }
             }
-            mock_required_data = [
-                "android_phone_battery_level",
-                "android_phone_step_count",
-            ]
-            self.data = SparkCSVDataReader(
-                mock_config_input, mock_required_data
+            sparkcsvdatareader = SparkCSVDataReader(
+                mock_config_input, self.total_required_data,
+                spark_config=self.config['spark_config']
+            )
+            self.data = sparkcsvdatareader.read_data()
+            sparkcsvdatareader.close_spark_session()
+
+        elif self.config["input"]["data_type"] == "sftp":
+            sftp_data_reader = SftpDataReader(self.config["input"]["config"],
+                                              self.total_required_data)
+            root_dir = sftp_data_reader.get_root_dir()
+            logger.info("Reading data from sftp")
+            sftp_data_reader.read_sftp_data()
+            sftp_local_config = {
+                "config": {
+                    "source_path": root_dir
+                }
+            }
+            sparkcsvdatareader = SparkCSVDataReader(
+                sftp_local_config,
+                self.total_required_data,
+                self.config["configurations"]["df_type"],
+                self.config['spark_config']
             ).read_data()
-
+            self.data = sparkcsvdatareader.read_data()
+            sparkcsvdatareader.close_spark_session()
         else:
             raise ValueError("Wrong data location")
 
@@ -471,14 +295,14 @@ def export_data(self) -> None:
         if self.config["configurations"]["df_type"] == "pandas":
             writer = PandasDataWriter(
                 self.features,
-                self.config["output_data"]["local_directory"],
-                self.config["output_data"]["compress"],
+                self.config["output"]['config']["target_path"],
+                self.config["output"]["compress"],
             )
         elif self.config["configurations"]["df_type"] == "spark":
             writer = SparkDataWriter(
                 self.features,
-                self.config["output_data"]["local_directory"],
-                self.config["output_data"]["compress"],
+                self.config["output"]['config']["target_path"],
+                self.config["output"]["compress"],
             )
         else:
             raise ValueError("Wrong df_type")
diff --git a/radarpipeline/project/validations.py b/radarpipeline/project/validations.py
new file mode 100644
index 0000000..06f6a29
--- /dev/null
+++ b/radarpipeline/project/validations.py
@@ -0,0 +1,225 @@
+import os
+import logging
+from git.exc import GitCommandError
+from git.repo import Repo
+
+from radarpipeline.common import utils
+
+logger = logging.getLogger(__name__)
+
+
+class ConfigValidator():
+    def __init__(self, config, valid_input_formats, valid_output_formats) -> None:
+        self.valid_input_formats = valid_input_formats
+        self.valid_output_formats = valid_output_formats
+        self.config = config
+
+    def validate(self):
+        """
+        Validates the config to check if all the keys are present
+        """
+
+        required_keys = ["project", "input", "features", "output"]
+        for key in required_keys:
+            if key not in self.config:
+                raise ValueError(f"Key not present in the config: {key}")
+
+        self._validate_input()
+        self._validate_configurations()
+        self._validate_features()
+        self._validate_output()
+
+        logger.info("Config file validated successfully")
+
+    def _update_mock_data(self) -> None:
+        """
+        Updates the mock data submodule of the åproject
+        """
+
+        repo = Repo(
+            os.path.dirname(os.path.abspath(__file__)),
+            search_parent_directories=True,
+        )
+        sms = repo.submodules
+        try:
+            mock_data_submodule = sms["mockdata"]
+        except IndexError:
+            raise ValueError("mockdata submodule not found in the repository")
+        if not (mock_data_submodule.exists() and mock_data_submodule.module_exists()):
+            logger.info("Mock data submodule not found. Cloning it...")
+            repo.git.submodule("update", "--init", "--recursive")
+            logger.info("Mock data input cloned")
+
+    def _validate_input(self) -> None:
+        """
+        Validates the input data config
+        """
+
+        if self.config["input"]["data_type"] == "sftp":
+            sftp_config_keys = [
+                "sftp_host",
+                "sftp_username",
+                "sftp_source_path",
+                "sftp_private_key",
+            ]
+            for key in sftp_config_keys:
+                if key not in self.config["input"]["config"]:
+                    raise ValueError(f"Key not present in the config: {key}")
+
+        elif self.config["input"]["data_type"] == "local":
+            if "source_path" not in self.config["input"]["config"]:
+                raise ValueError("Key not present in the config: source_path")
+            else:
+                # Check if local_directory is absolute path. If not, then set it.
+                local_directory = self.config["input"]["config"]["source_path"]
+                if isinstance(local_directory, list):
+                    local_directory = [
+                        self._get_absolute_path(local_path)
+                        for local_path in local_directory]
+                    for local_path in local_directory:
+                        if not os.path.exists(local_path):
+                            raise ValueError(f"Path does not exist: {local_path}")
+                else:
+                    local_directory = utils.get_absolute_path(local_directory)
+                    if not os.path.exists(local_directory):
+                        raise ValueError(f"Path does not exist: {local_directory}")
+                self.config["input"]["local_directory"] = local_directory
+
+            # Raise error if data_format it not valid input formats
+            if self.config["input"]["data_format"] not in self.valid_input_formats:
+                raise ValueError("Invalid value for key in input: data_format")
+
+        elif self.config["input"]["data_type"] == "mock":
+            self._update_mock_data()
+
+        else:
+            raise ValueError("Invalid value for the key: data_location")
+
+    def _validate_configurations(self) -> None:
+        """
+        Validates the configurations config
+        """
+
+        if "configurations" not in self.config:
+            self.config["configurations"] = {}
+
+        valid_df_types = ["pandas", "spark"]
+
+        if "df_type" not in self.config["configurations"]:
+            self.config["configurations"]["df_type"] = "pandas"
+        elif self.config["configurations"]["df_type"] not in valid_df_types:
+            raise ValueError("Invalid value for the key: df_type")
+
+    def _validate_features(self) -> None:
+        """
+        Validates the features config
+        """
+
+        if len(self.config["features"]) == 0:
+            raise ValueError("features array cannot be empty")
+
+        for index, feature in enumerate(self.config["features"]):
+            if "location" not in feature:
+                raise ValueError(
+                    f"Key not present in the config: location at index {index}"
+                )
+
+            if "feature_groups" not in feature:
+                raise ValueError(
+                    f"Key not present in the config: feature_groups at index {index}"
+                )
+
+            if len(feature["feature_groups"]) == 0:
+                raise ValueError(
+                    f"feature_groups array cannot be empty at index {index}"
+                )
+            if "feature_names" not in feature:
+                raise ValueError(
+                    f"Key not present in the config: feature_groups at index {index}"
+                )
+
+            if len(feature["feature_names"]) == 0:
+                raise ValueError(
+                    f"feature_groups array cannot be empty at index {index}"
+                )
+            feature_location = feature["location"]
+
+            if utils.is_valid_github_path(feature_location):
+                repo_name = utils.get_repo_name_from_url(feature_location)
+                cache_dir = os.path.join(
+                    os.path.expanduser("~"), ".cache", "radarpipeline", repo_name
+                )
+
+                if not os.path.exists(cache_dir):
+                    Repo.clone_from(feature_location, cache_dir)
+
+                repo = Repo(cache_dir)
+                repo.git.reset("--hard")
+                repo.git.clean("-xdf")
+
+                active_branch_name = repo.active_branch.name
+                feature_branch = feature.get("branch", active_branch_name)
+
+                try:
+                    repo.git.checkout(feature_branch)
+                except GitCommandError:
+                    logger.warning(
+                        "Branch %s does not exist. Using the %s branch instead.",
+                        feature_branch,
+                        repo.active_branch.name,
+                    )
+                    feature_branch = repo.active_branch.name
+                    repo.git.checkout(feature_branch)
+                repo.remotes.origin.pull(feature_branch)
+
+                feature_location = cache_dir
+                logger.info(f"Using feature from cache location: {feature_location}")
+            else:
+                feature_location = os.path.expanduser(feature_location)
+                if not os.path.isdir(feature_location):
+                    raise ValueError(f"Invalid feature location: {feature_location}")
+                logger.info(f"Using feature from local path: {feature_location}")
+
+            self.config["features"][index]["location"] = feature_location
+
+    def _validate_output(self) -> None:
+        """
+        Validates the output data config
+        """
+
+        if self.config["output"]["output_location"] == "postgres":
+            postgres_config_keys = [
+                "postgres_host",
+                "postgres_username",
+                "postgres_database",
+                "postgres_password",
+                "postgres_table",
+            ]
+            for key in postgres_config_keys:
+                if key not in self.config["output"]['config']:
+                    raise ValueError(f"Key not present in the config: {key}")
+
+        elif self.config["output"]["output_location"] == "local":
+            if "target_path" not in self.config["output"]['config']:
+                raise ValueError("Key not present in the config: target_path")
+            else:
+                # Check if local_directory is absolute path. If not, then set it.
+                local_directory = self.config["output"]['config']["target_path"]
+                local_directory = utils.get_absolute_path(local_directory)
+                if not os.path.exists(local_directory):
+                    os.makedirs(local_directory, exist_ok=True)
+                self.config["output"]['config']["target_path"] = local_directory
+
+            # Raise error if data_format it not valid output formats
+            if (
+                self.config["output"]["data_format"]
+                not in self.valid_output_formats
+            ):
+                raise ValueError("Invalid value for key in output: data_format")
+
+            if "compress" not in self.config["output"]:
+                self.config["output"]["compress"] = False
+            if self.config["output"]["compress"] == "true":
+                self.config["output"]["compress"] = True
+        else:
+            raise ValueError("Key not present in the config: output_location")
diff --git a/radarpipeline/radarpipeline.py b/radarpipeline/radarpipeline.py
index c4b4073..8f1d580 100644
--- a/radarpipeline/radarpipeline.py
+++ b/radarpipeline/radarpipeline.py
@@ -10,14 +10,16 @@
 logger = logging.getLogger(__name__)
 
 
-def run():
+def run(config_path: str = "config.yaml"):
     """
     Pipeline entry point.
+    config_path could be a local path to a configuration file
+    or a remote url to a configuration file.
     """
     try:
         logger.info("Starting the pipeline run...")
         logger.info("Reading and Validating the configuration file...")
-        project = Project(input_data="config.yaml")
+        project = Project(input_data=config_path)
         logger.info("Fetching the data...")
         project.fetch_data()
         logger.info("Computing the features...")
@@ -33,5 +35,34 @@ def run():
         sys.exit(1)
 
 
+def _gather_clone_urls(organization, no_forks=True):
+    """
+    gh = pygithub3.Github()
+    all_repos = gh.repos.list(user=organization).all()
+    for repo in all_repos:
+
+        # Don't print the urls for repos that are forks.
+        if no_forks and repo.fork:
+            continue
+
+        yield repo.clone_url
+    TODO: implement this function
+    """
+
+
+def show_available_pipelines():
+    """
+    Uses git to show the available pipelines from
+    https://github.com/RADAR-base-Analytics/
+    """
+    """
+    # Still in progress. not working yet
+    clone_urls = _gather_clone_urls("RADAR-base-Analytics")
+    print(clone_urls)
+    return clone_urls
+    # read all the repos from https://github.com/RADAR-base-Analytics/
+    """
+
+
 if __name__ == "__main__":
     run()
diff --git a/requirements.txt b/requirements.txt
index 2ce00da..bfb7b83 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -15,3 +15,5 @@ strictyaml
 pyspark-test==0.2.0
 pytest-cov
 pytest-mock
+pytest-sftpserver
+dhp


Test patch:
diff --git a/__init__.py b/conftest.py
similarity index 100%
rename from __init__.py
rename to conftest.py
diff --git a/tests/__init__.py b/tests/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/tests/resources/test_yamls/config_with_incorrect_spark.yaml b/tests/resources/test_yamls/config_with_incorrect_spark.yaml
new file mode 100644
index 0000000..8c784de
--- /dev/null
+++ b/tests/resources/test_yamls/config_with_incorrect_spark.yaml
@@ -0,0 +1,45 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        # In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        # In case of s3, use the following format
+        #       aws_access_key_id:
+        #       aws_secret_access_key:
+        #       region_name:
+        #       s3_access_url:
+        #       bucket:
+        #       prefix:
+        # In case of local or Mock, use the following format
+        source_path: mockdata/mockdata
+    data_format: csv
+
+configurations:
+    df_type: 'pandas'
+
+features:
+    - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
+      branch: main
+      feature_groups:
+          - MockFeatureGroup
+      feature_names:
+        - all
+
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
+    data_format: csv
+    compress: false
+
+spark_config:
+    spark.memory.offHeapSize: false
diff --git a/tests/resources/test_yamls/config_with_spark.yaml b/tests/resources/test_yamls/config_with_spark.yaml
new file mode 100644
index 0000000..728f68f
--- /dev/null
+++ b/tests/resources/test_yamls/config_with_spark.yaml
@@ -0,0 +1,46 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        # In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        # In case of s3, use the following format
+        #       aws_access_key_id:
+        #       aws_secret_access_key:
+        #       region_name:
+        #       s3_access_url:
+        #       bucket:
+        #       prefix:
+        # In case of local or Mock, use the following format
+        source_path: mockdata/mockdata
+    data_format: csv
+
+configurations:
+    df_type: 'pandas'
+
+features:
+    - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
+      branch: main
+      feature_groups:
+          - MockFeatureGroup
+      feature_names:
+        - all
+
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
+    data_format: csv
+    compress: false
+
+spark_config:
+    spark.executor.instances: 2
+    spark.memory.offHeap.enabled: false
diff --git a/tests/resources/test_yamls/test_config.yaml b/tests/resources/test_yamls/test_config.yaml
new file mode 100644
index 0000000..6368c18
--- /dev/null
+++ b/tests/resources/test_yamls/test_config.yaml
@@ -0,0 +1,42 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        # In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        # In case of s3, use the following format
+        #       aws_access_key_id:
+        #       aws_secret_access_key:
+        #       region_name:
+        #       s3_access_url:
+        #       bucket:
+        #       prefix:
+        # In case of local or Mock, use the following format
+        source_path: mockdata/mockdata
+    data_format: csv
+
+configurations:
+    df_type: 'pandas'
+
+features:
+    - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
+      branch: main
+      feature_groups:
+          - MockFeatureGroup
+      feature_names:
+        - all
+
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
+    data_format: csv
+    compress: false
\ No newline at end of file
diff --git a/tests/resources/test_yamls/test_config_incomplete.yaml b/tests/resources/test_yamls/test_config_incomplete.yaml
index 54907cd..22fa2b4 100644
--- a/tests/resources/test_yamls/test_config_incomplete.yaml
+++ b/tests/resources/test_yamls/test_config_incomplete.yaml
@@ -3,8 +3,24 @@ project:
     description: mock_description
     version: mock_version
 
-input_data:
-    local_directory: mockdata/mockdata
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        # In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        # In case of s3, use the following format
+        #       aws_access_key_id:
+        #       aws_secret_access_key:
+        #       region_name:
+        #       s3_access_url:
+        #       bucket:
+        #       prefix:
+        # In case of local or Mock, use the following format
+        source_path: mockdata/mockdata
     data_format: csv
 
 configurations:
@@ -14,6 +30,9 @@ features:
     - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
       branch: main
 
-output_data:
-    output_location: local
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
     data_format: csv
+    compress: false
diff --git a/tests/resources/test_yamls/test_config_input_invalid.yaml b/tests/resources/test_yamls/test_config_input_invalid.yaml
index 80228d7..ad2e3da 100644
--- a/tests/resources/test_yamls/test_config_input_invalid.yaml
+++ b/tests/resources/test_yamls/test_config_input_invalid.yaml
@@ -3,9 +3,24 @@ project:
     description: mock_description
     version: mock_version
 
-input_data:
-    data_location: local
-    local_directory: xyz
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        # In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        # In case of s3, use the following format
+        #       aws_access_key_id:
+        #       aws_secret_access_key:
+        #       region_name:
+        #       s3_access_url:
+        #       bucket:
+        #       prefix:
+        # In case of local or Mock, use the following format
+        source_path: xyz
     data_format: csv
 
 configurations:
@@ -19,8 +34,9 @@ features:
       feature_names:
         - all
 
-output_data:
-    output_location: local
-    local_directory: output/mockdata
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
     data_format: csv
     compress: false
diff --git a/tests/resources/test_yamls/test_config_project.yaml b/tests/resources/test_yamls/test_config_project.yaml
index 0d59de9..e520122 100644
--- a/tests/resources/test_yamls/test_config_project.yaml
+++ b/tests/resources/test_yamls/test_config_project.yaml
@@ -3,9 +3,24 @@ project:
     description: mock_description
     version: mock_version
 
-input_data:
-    data_location: local
-    local_directory: mockdata/mockdata
+input:
+    data_type: mock # couldbe mock, local, sftp, s3
+    config:
+        # In case of sftp, use the following format
+        # sftp_host:
+        # sftp_source_path:
+        # sftp_username:
+        # sftp_private_key:
+        # sftp_target_path:
+        # In case of s3, use the following format
+        #       aws_access_key_id:
+        #       aws_secret_access_key:
+        #       region_name:
+        #       s3_access_url:
+        #       bucket:
+        #       prefix:
+        # In case of local or Mock, use the following format
+        source_path: mockdata/mockdata
     data_format: csv
 
 configurations:
@@ -19,8 +34,9 @@ features:
       feature_names:
         - all
 
-output_data:
-    output_location: local
-    local_directory: tests/resources/test_output
+output:
+    output_location: local # can be local, postgres, sftp
+    config:
+        target_path: output/mockdata
     data_format: csv
     compress: false
diff --git a/tests/test_integration/test_integration.py b/tests/test_integration/test_integration.py
index 44bf52b..1317476 100644
--- a/tests/test_integration/test_integration.py
+++ b/tests/test_integration/test_integration.py
@@ -25,7 +25,7 @@ def test_run(self):
         self.assertFalse(raised, 'Exception raised')
         project = Project(input_data="config.yaml")
         # Assert if output files are created
-        self.output_dir = project.config['output_data']['local_directory']
+        self.output_dir = project.config['output']["config"]['target_path']
         path = pl.Path(os.path.join(self.output_dir,
                                     "phone_battery_charging_duration.csv"))
         self.assertIsFile(path)
diff --git a/tests/tests_common/test_utils.py b/tests/tests_common/test_utils.py
index ca11fdf..d173e19 100644
--- a/tests/tests_common/test_utils.py
+++ b/tests/tests_common/test_utils.py
@@ -4,6 +4,7 @@
 import os
 from radarpipeline.common.utils import is_valid_github_path, read_yaml
 from radarpipeline.common import utils
+from strictyaml.exceptions import YAMLValidationError
 
 RADAR_PIPELINE_URL = "https://github.com/RADAR-base/radarpipeline"
 WRONG_GITHUB_URL = "https://githuib.com/RADAR-base/radarpipeline"
@@ -72,30 +73,36 @@ def test_combine_pyspark_dfs():
 class TestReadYaml(unittest.TestCase):
 
     def setUp(self):
-        self.TESTDATA_FILENAME = "config.yaml"
+        self.TESTDATA_FILENAME = "tests/resources/test_yamls/test_config.yaml"
+        self.TESTDATA_FILENAME_SPARK = "tests/resources/test_yamls/"\
+            "config_with_spark.yaml"
+        self.TESTDATA_FILENAME_INCORRECT_SPARK = "tests/resources/test_yamls/"\
+            "config_with_incorrect_spark.yaml"
         self.TESTDATA_FILENAME_WRONG = "tests/resources/config.yaml"
         self.TESTDATA_FILENAME_EMPTY = "tests/resources/test_config.yaml"
 
     def test_read_correct_yaml(self):
         config = read_yaml(self.TESTDATA_FILENAME)
+        print(config)
         expected_config = {
             'project': {
                 'project_name': 'mock_project',
                 'description': 'mock_description',
                 'version': 'mock_version'},
-            'input_data': {
-                'data_location': 'local',
-                'local_directory': 'mockdata/mockdata',
-                'data_format': 'csv'},
+            'input': {
+                'data_type': 'mock',
+                'config': {'source_path': 'mockdata/mockdata'},
+                'data_format': 'csv'
+            },
             'configurations': {'df_type': 'pandas'},
             'features': [{
                 'location': 'https://github.com/RADAR-base-Analytics/mockfeatures',
                 'branch': 'main',
                 'feature_groups': ['MockFeatureGroup'],
                 'feature_names': [['all']]}],
-            'output_data': {
+            'output': {
                 'output_location': 'local',
-                'local_directory': 'output/mockdata',
+                'config': {'target_path': 'output/mockdata'},
                 'data_format': 'csv',
                 'compress': False}}
         self.assertDictEqual(config, expected_config)
@@ -104,3 +111,41 @@ def test_read_unavailable_yaml(self):
         with self.assertRaises(ValueError):
             read_yaml(self.TESTDATA_FILENAME_WRONG)
             read_yaml(self.TESTDATA_FILENAME_EMPTY)
+
+    def test_read_yaml_with_spark_config(self):
+        config = read_yaml(self.TESTDATA_FILENAME_SPARK)
+        print(config)
+        expected_config = {
+            'project': {
+                'project_name': 'mock_project',
+                'description': 'mock_description',
+                'version': 'mock_version'},
+            'input': {
+                'data_type': 'mock',
+                'config': {'source_path': 'mockdata/mockdata'},
+                'data_format': 'csv'
+            },
+            'configurations': {'df_type': 'pandas'},
+            'features': [{
+                'location': 'https://github.com/RADAR-base-Analytics/mockfeatures',
+                'branch': 'main',
+                'feature_groups': ['MockFeatureGroup'],
+                'feature_names': [['all']]}],
+            'output': {
+                'output_location': 'local',
+                'config': {'target_path': 'output/mockdata'},
+                'data_format': 'csv',
+                'compress': False},
+            'spark_config': {
+                "spark.executor.instances": 2,
+                "spark.memory.offHeap.enabled": False,
+                "spark.executor.cores": 4,
+                "spark.executor.memory": "10g",
+                "spark.driver.memory": "15g",
+                "spark.memory.offHeap.size": "20g",
+                "spark.driver.maxResultSize": "0"}}
+        self.assertDictEqual(config, expected_config)
+
+    def test_read_yaml_with_incorrect_spark_config(self):
+        with self.assertRaises(YAMLValidationError):
+            read_yaml(self.TESTDATA_FILENAME_INCORRECT_SPARK)
diff --git a/tests/tests_io/test_connection.py b/tests/tests_io/test_connection.py
new file mode 100644
index 0000000..15df190
--- /dev/null
+++ b/tests/tests_io/test_connection.py
@@ -0,0 +1,230 @@
+import unittest
+from pytest_sftpserver.sftp.server import SFTPServer
+import pytest
+from paramiko import Transport
+from paramiko.channel import Channel
+from paramiko.sftp_client import SFTPClient
+import sys
+import os
+
+from copy import deepcopy
+from radarpipeline.io.connection import SftpConnector
+from dhp.test import tempfile_containing
+import shutil
+from tempfile import mkdtemp
+
+STARS8192 = '*' * 8192
+TEST_PATH_DIR = "tests/resources/sftp_test"
+
+CONTENT_OBJ = {
+    'home': {
+        'test': {
+            'pub': {
+                'make.txt': "content of make.txt",
+                'foo1': {
+                    'foo1.txt': 'content of foo1.txt',
+                    'image01.jpg': 'data for image01.jpg'
+                },
+                'foo2': {
+                    'foo2.txt': 'content of foo2.txt',
+                    'bar1': {
+                        'bar1.txt': 'contents bar1.txt'
+                    }
+                }
+            },
+            'read.me': 'contents of read.me',
+        }
+    }
+}
+
+
+def conn(sftpsrv):
+    """return a dictionary holding argument info for the pysftp client"""
+    return {'sftp_host': sftpsrv.host, 'sftp_port': sftpsrv.port,
+            'sftp_username': 'user',
+            'sftp_password': 'pw', 'sftp_source_path': '/home/test'}
+
+
+@pytest.yield_fixture(scope="session")
+def sftpclient(sftpserver):
+    transport = Transport((sftpserver.host, sftpserver.port))
+    transport.connect(username="a", password="b")
+    sftpclient = SFTPClient.from_transport(transport)
+    yield sftpclient
+    sftpclient.close()
+    transport.close()
+
+
+class TestConnection(unittest.TestCase):
+
+    DIR_LIST = [('pub', ),
+                ('pub', 'foo1'),
+                ('pub', 'foo2'),
+                ('pub', 'foo2', 'bar1')]
+    FILE_LIST = [('pub', 'read.me'),
+                 ('pub', 'make.txt'),
+                 ('pub', 'foo1', 'foo1.txt'),
+                 ('pub', 'foo2', 'foo2.txt'),
+                 ('pub', 'foo2', 'bar1', 'bar1.txt')]
+    TEST_PATH_DIR = "./tests/resources/sftp_test"
+    STARS8192 = '*' * 8192
+
+    @pytest.fixture(autouse=True)
+    def prepare_fixture(self, sftpserver):
+        self.sftpserver = sftpserver
+
+    @staticmethod
+    def build_dir_struct(cls, local_path):
+        '''build directory structure'''
+        for dparts in cls.DIR_LIST:
+            os.mkdir(os.path.join(local_path, *dparts))
+        for fparts in cls.FILE_LIST:
+            with open(os.path.join(local_path, *fparts), 'wb') as fhndl:
+                try:
+                    fhndl.write(STARS8192)
+                except TypeError:
+                    fhndl.write(bytes(STARS8192, 'UTF-8'))
+
+    @staticmethod
+    def remove_dir_struct(cls, local_path):
+        '''clean up directory struct'''
+        for fparts in cls.FILE_LIST:
+            os.remove(os.path.join(local_path, *fparts))
+        for dparts in reversed(cls.DIR_LIST):
+            os.rmdir(os.path.join(local_path, *dparts))
+
+    @classmethod
+    def setUpClass(cls):
+        # create TEST_PATH_DIR if it does not exist
+        if not os.path.exists(TEST_PATH_DIR):
+            os.mkdir(TEST_PATH_DIR)
+        cls.build_dir_struct(cls, TEST_PATH_DIR)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.remove_dir_struct(cls, TEST_PATH_DIR)
+        # Delete TEST_PATH_DIR if it is empty
+        if not os.listdir(TEST_PATH_DIR):
+            os.rmdir(TEST_PATH_DIR)
+
+    def setUp(self):
+        self.sftp = SftpConnector(conn(self.sftpserver), ['x'])
+        self.home = "/home/test"
+
+    def tearDown(self):
+        self.sftp.close()
+
+    def test_connection_with(self):
+        '''connect to a public self.sftp server'''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+
+    def test_cd_path(self):
+        '''test .cd with a path'''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            home = self.sftp.pwd
+            with self.sftp.cd(f'{self.home}/pub'):
+                assert self.sftp.pwd == f'{self.home}/pub'
+            assert home == self.sftp.pwd
+
+    def test_cd_nested(self):
+        '''test nested cd's'''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            home = self.sftp.pwd
+            with self.sftp.cd('/home'):
+                assert self.sftp.pwd == '/home'
+                with self.sftp.cd('test/pub'):
+                    assert self.sftp.pwd == f'{self.home}/pub'
+                assert self.sftp.pwd == '/home'
+            assert home == self.sftp.pwd
+
+    def test_listdir(self):
+        '''test listdir'''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            with self.sftp.cd(f'{self.home}/pub'):
+                assert self.sftp.listdir() == ['foo1', 'foo2', 'make.txt']
+            self.sftp.close()
+
+    def test_listdir_attr(self):
+        '''test listdir'''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            with self.sftp.cd(f'{self.home}/pub'):
+                attrs = self.sftp._sftp.listdir_attr()
+                assert len(attrs) == 3
+                # test they are in filename order
+                assert attrs[0].filename == 'make.txt'
+                assert attrs[1].filename == 'foo1'
+                assert attrs[2].filename == 'foo2'
+                # test that longname is there
+                for attr in attrs:
+                    assert attr.longname is not None
+            self.sftp.close()
+
+    def test_isfile(self):
+
+        '''test .isfile() functionality'''
+
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            rfile = f'{self.home}/pub/make.txt'
+            rdir = 'pub'
+            assert self.sftp.isfile(rfile)
+            assert self.sftp.isfile(rdir) is False
+            self.sftp.close()
+
+    def test_get(self):
+        '''download a file'''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            with self.sftp.cd(f'{self.home}/pub/foo1'):
+                with tempfile_containing('') as fname:
+                    self.sftp.get('foo1.txt', fname)
+                    assert open(fname, 'rb').read() == b'content of foo1.txt'
+            self.sftp.close()
+
+    def test_get_preserve_mtime(self):
+        '''test that m_time is preserved from local to remote, when get'''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            with self.sftp.cd(f'{self.home}/pub/foo1'):
+                rfile = 'foo1.txt'
+                with tempfile_containing('') as localfile:
+                    r_stat = self.sftp._sftp.stat(rfile)
+                    self.sftp.get(rfile, localfile, preserve_mtime=True)
+                    assert r_stat.st_mtime == os.stat(localfile).st_mtime
+            self.sftp.close()
+
+    def test_get_d(self):
+        '''test the get_d for remotepath is pwd '.' '''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            with self.sftp.cd(f'{self.home}/pub/'):
+                localpath = mkdtemp()
+                self.sftp.get_d('.', localpath)
+                checks = [(['', ], ['make.txt', ]), ]
+                for pth, fls in checks:
+                    assert sorted(os.listdir(os.path.join(localpath, *pth))) == fls
+                # cleanup local
+                shutil.rmtree(localpath)
+            self.sftp.close()
+
+    def test_get_d_pathed(self):
+        '''test the get_d for localpath, starting deeper then pwd '''
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            self.sftp.connect()
+            with self.sftp.cd(f'{self.home}/pub/'):
+                localpath = mkdtemp()
+                self.sftp.get_d('foo1', localpath)
+
+                chex = [(['', ],
+                         ['foo1.txt', 'image01.jpg']), ]
+                for pth, fls in chex:
+                    assert sorted(os.listdir(os.path.join(localpath, *pth))) == fls
+
+                # cleanup local
+                shutil.rmtree(localpath)
+            self.sftp.close()
diff --git a/tests/tests_io/test_downloader.py b/tests/tests_io/test_downloader.py
new file mode 100644
index 0000000..3a33445
--- /dev/null
+++ b/tests/tests_io/test_downloader.py
@@ -0,0 +1,152 @@
+import unittest
+from pytest_sftpserver.sftp.server import SFTPServer
+import pytest
+from paramiko import Transport
+from paramiko.channel import Channel
+from paramiko.sftp_client import SFTPClient
+import sys
+import os
+
+from copy import deepcopy
+from radarpipeline.io.downloader import SftpDataReader
+from dhp.test import tempfile_containing
+import shutil
+from tempfile import mkdtemp
+
+STARS8192 = '*' * 8192
+TEST_PATH_DIR = "tests/resources/sftp_test"
+
+CONTENT_OBJ = {
+    'mock_radar_sftp_data': {
+        '5c0e2ec7-6f85-4041-9669-7145075d1754': {
+            'android_phone_battery_level': {
+                '20181025_1200.csv.gz': "Mock CSV",
+                '20181025_1300.csv.gz': "Mock CSV",
+                'schema-android_phone_battery_level.json': "Mock JSON"
+            },
+            'android_phone_step_count': {
+                '20181025_1200.csv.gz': "Mock CSV",
+                '20181025_1300.csv.gz': "Mock CSV",
+                'schema-android_phone_battery_level.json': "Mock JSON"
+            }
+        },
+        '2a02e53a-951e-4fd0-b47f-195a87096bd0': {
+            'android_phone_acceleration': {
+                '20181127_0900.csv.gz': "Mock CSV",
+                '20181127_1000.csv.gz': "Mock CSV",
+                'schema-android_phone_acceleration.json': "Mock JSON"
+            },
+            'android_phone_battery_level': {
+                '20181123_1500.csv.gz': "Mock CSV",
+                '20181123_1600.csv.gz': "Mock CSV",
+                'schema-android_phone_battery_level.json': "Mock JSON"
+            },
+            'android_phone_step_count': {
+                '20181123_1500.csv.gz': "Mock CSV",
+                '20181123_1700.csv.gz': "Mock CSV",
+                'schema-android_phone_step_count.json': "Mock JSON"
+            }
+        }
+    }
+}
+
+
+def conn(sftpsrv):
+    """return a dictionary holding argument info for the pysftp client"""
+    return {'sftp_host': sftpsrv.host, 'sftp_port': sftpsrv.port,
+            'sftp_username': 'user',
+            'sftp_password': 'pw', 'sftp_source_path': 'mock_radar_sftp_data',
+            "sftp_target_path": TEST_PATH_DIR}
+
+
+@pytest.yield_fixture(scope="session")
+def sftpclient(sftpserver):
+    transport = Transport((sftpserver.host, sftpserver.port))
+    transport.connect(username="a", password="b")
+    sftpclient = SFTPClient.from_transport(transport)
+    yield sftpclient
+    sftpclient.close()
+    transport.close()
+
+
+class TestDownloader(unittest.TestCase):
+
+    @pytest.fixture(autouse=True)
+    def prepare_fixture(self, sftpserver):
+        self.sftpserver = sftpserver
+
+    def setUp(self):
+        self.variables = ['android_phone_battery_level', 'android_phone_step_count']
+        self.config_dict = conn(self.sftpserver)
+        self.target_path = TEST_PATH_DIR
+        self.source_path = "mock_radar_sftp_data"
+
+    def test_get_root_dir(self):
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            sftp_data_reader = SftpDataReader(self.config_dict, self.variables)
+            root_dir = sftp_data_reader.get_root_dir()
+            self.assertEqual(root_dir, self.target_path)
+
+    def test_get_all_ids(self):
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            sftp_data_reader = SftpDataReader(self.config_dict, self.variables)
+            ids = sftp_data_reader._get_all_id_sftp(self.source_path)
+            expected_ids = ['5c0e2ec7-6f85-4041-9669-7145075d1754',
+                            '2a02e53a-951e-4fd0-b47f-195a87096bd0']
+            # assert list equal
+            self.assertListEqual(sorted(ids), sorted(expected_ids))
+
+    def test_read_sftp_data(self):
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            sftp_data_reader = SftpDataReader(self.config_dict, self.variables)
+            sftp_data_reader.read_sftp_data()
+        # check if the files are downloaded in the target path
+        uids = ['5c0e2ec7-6f85-4041-9669-7145075d1754',
+                '2a02e53a-951e-4fd0-b47f-195a87096bd0']
+        for uid in uids:
+            self.assertTrue(os.path.exists(os.path.join(self.target_path, uid)))
+            for variable in self.variables:
+                self.assertTrue(os.path.exists(os.path.join(self.target_path,
+                                                            uid, variable)))
+
+    def test_read_sftp_data_with_one_variable(self):
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            variables = ['android_phone_acceleration']
+            sftp_data_reader = SftpDataReader(self.config_dict, variables)
+            sftp_data_reader.read_sftp_data()
+            uid_correct = '2a02e53a-951e-4fd0-b47f-195a87096bd0'
+            self.assertTrue(os.path.exists(os.path.join(self.target_path, uid_correct)))
+            for variable in variables:
+                self.assertTrue(os.path.exists(os.path.join(self.target_path,
+                                                            uid_correct, variable)))
+            uid_incorrect = '5c0e2ec7-6f85-4041-9669-7145075d1754'
+            self.assertFalse(os.path.exists(os.path.join(self.target_path,
+                                                         uid_incorrect)))
+
+    def test_read_sftp_data_with_invalid_config(self):
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            variables = ['invalid']
+            sftp_data_reader = SftpDataReader(self.config_dict, variables)
+            sftp_data_reader.read_sftp_data()
+            uids = ['5c0e2ec7-6f85-4041-9669-7145075d1754',
+                    '2a02e53a-951e-4fd0-b47f-195a87096bd0']
+            for uid in uids:
+                self.assertFalse(os.path.exists(os.path.join(self.target_path, uid)))
+
+    def test_fetch_data(self):
+        with self.sftpserver.serve_content(CONTENT_OBJ):
+            sftp_data_reader = SftpDataReader(self.config_dict, self.variables)
+            root_dir = sftp_data_reader.get_root_dir()
+            uid = '5c0e2ec7-6f85-4041-9669-7145075d1754'
+            sftp_data_reader._fetch_data(root_dir, self.source_path,
+                                         self.variables, uid)
+        # check if the files are downloaded in the target path
+        self.assertTrue(os.path.exists(os.path.join(self.target_path, uid)))
+        for variable in self.variables:
+            self.assertTrue(os.path.exists(
+                os.path.join(self.target_path, uid, variable)))
+
+    def tearDown(self):
+        # if self.target_path exists, remove it
+        if os.path.exists(self.target_path):
+            shutil.rmtree(self.target_path)
diff --git a/tests/tests_io/test_reader.py b/tests/tests_io/test_reader.py
index dbeecfe..7590ad0 100644
--- a/tests/tests_io/test_reader.py
+++ b/tests/tests_io/test_reader.py
@@ -1,6 +1,7 @@
 import unittest
 from radarpipeline.io import SparkCSVDataReader, AvroSchemaReader
 import pandas as pd
+import radarpipeline
 from radarpipeline.datalib import RadarVariableData, RadarUserData, RadarData
 from pandas.testing import assert_frame_equal
 from numpy.testing import assert_array_equal
@@ -8,7 +9,7 @@
 
 class TestSparkCSVDataReader(unittest.TestCase):
     def setUp(self):
-        mock_config = {"local_directory": "tests/resources/test_data/"}
+        mock_config = {"config": {"source_path": "tests/resources/test_data/"}}
         data_list = ['android_phone_step_count']
         self.sparkcsvdatareader = SparkCSVDataReader(mock_config,
                                                      required_data=data_list)
@@ -30,3 +31,36 @@ def test_read_data(self):
                                    RadarUserData))
         self.assertTrue(isinstance(spark_data.get_data()["test_participant"].get_data()
                                    ["android_phone_step_count"], RadarVariableData))
+
+    def tearDown(self):
+        self.sparkcsvdatareader.close_spark_session()
+
+
+class TestSparkCustomConfig(unittest.TestCase):
+    def setUp(self):
+        mock_config = {"config": {"source_path": "tests/resources/test_data/"}}
+        self.spark_config = {
+            "spark.executor.instances": "3",
+            "spark.memory.offHeap.enabled": False,
+            "spark.executor.cores": 2,
+            "spark.executor.memory": "5g",
+            "spark.driver.memory": "10g",
+            "spark.memory.offHeap.size": "10g",
+            "spark.driver.maxResultSize": "0"}
+        data_list = ['android_phone_step_count']
+        self.sparkcsvdatareader = SparkCSVDataReader(mock_config,
+                                                     required_data=data_list,
+                                                     spark_config=self.spark_config)
+
+    def test_spark_config(self):
+        spark_config_output = dict(self.sparkcsvdatareader.spark.sparkContext.
+                                   getConf().getAll())
+        for key, value in self.spark_config.items():
+            self.assertEqual(spark_config_output[key], str(value))
+
+    def test_spark_config_dict(self):
+        spark_config_output = self.sparkcsvdatareader.spark_config
+        self.assertDictEqual(self.spark_config, spark_config_output)
+
+    def tearDown(self):
+        self.sparkcsvdatareader.close_spark_session()
diff --git a/tests/tests_project/test_project.py b/tests/tests_project/test_project.py
index 9b466b1..41edbc8 100644
--- a/tests/tests_project/test_project.py
+++ b/tests/tests_project/test_project.py
@@ -3,6 +3,11 @@
 from strictyaml.exceptions import YAMLValidationError
 import pathlib as pl
 import os
+from strictyaml import as_document
+from radarpipeline.common import utils
+
+import requests
+import yaml
 
 
 class TestProjectInit(unittest.TestCase):
@@ -31,7 +36,7 @@ class TestProject(unittest.TestCase):
     Test running mock project
     '''
     def setUp(self):
-        config_path = 'config.yaml'
+        config_path = "tests/resources/test_yamls/test_config.yaml"
         self.project = Project(config_path)
 
     def assertIsFile(self, path):
@@ -51,7 +56,7 @@ def test_export_data(self):
         self.project.fetch_data()
         self.project.compute_features()
         self.project.export_data()
-        self.output_dir = self.project.config['output_data']['local_directory']
+        self.output_dir = self.project.config['output']['config']['target_path']
         path = pl.Path(os.path.join(self.output_dir,
                                     "phone_battery_charging_duration.csv"))
         self.assertIsFile(path)
@@ -60,5 +65,36 @@ def test_export_data(self):
         self.assertIsFile(path)
         path.unlink()
 
+    def test_get_total_required_data(self):
+        required_data_output = self.project._get_total_required_data()
+        expected_data = ['android_phone_battery_level', 'android_phone_step_count']
+        self.assertListEqual(sorted(required_data_output), sorted(expected_data))
+
     def tearDown(self) -> None:
         del self.project
+
+
+class TestProjectRemoteLink(unittest.TestCase):
+    def setUp(self):
+        self.remotelink = "https://github.com/RADAR-base-Analytics/mockfeatures"
+        self.remotelink_raw = "https://raw.githubusercontent.com/"\
+            "RADAR-base-Analytics/mockfeatures/main/config.yaml"
+        response = requests.get(self.remotelink_raw , allow_redirects=True)
+        content = response.content.decode("utf-8")
+        self.expected_config = yaml.safe_load(content)
+
+    def test_resolve_remote_link(self):
+        project = Project(self.remotelink)
+        project_config_file = project._resolve_input_data(self.remotelink)
+        project_config = yaml.safe_load(open(project_config_file, 'r'))
+        # Assert that config file is as same as it is in the remote link
+        self.assertDictEqual(project_config, self.expected_config)
+
+    def test_get_config(self):
+        project = Project(self.remotelink)
+        project_config = project._get_config()
+        schema = utils.get_yaml_schema()
+        print(project_config)
+        print(self.expected_config)
+        expected_config_updated = as_document(self.expected_config, schema).data
+        self.assertDictEqual(project_config, expected_config_updated)
diff --git a/tests/tests_project/test_validations.py b/tests/tests_project/test_validations.py
new file mode 100644
index 0000000..33b7a94
--- /dev/null
+++ b/tests/tests_project/test_validations.py
@@ -0,0 +1,124 @@
+import unittest
+from radarpipeline.project.validations import ConfigValidator
+from copy import deepcopy
+
+
+MOCK_CONFIG = {
+    'project': {
+        'project_name' : 'mock_project',
+        'description': 'mock_description',
+        'version': 'mock_version'},
+    'input': {
+        'data_type': 'mock',
+        'config': {
+            'source_path': 'mockdata/mockdata'},
+        'data_format': 'csv'},
+    'configurations': {'df_type': 'pandas'},
+    'features': [
+        {'location':
+            'https://github.com/RADAR-base-Analytics/mockfeatures',
+            'branch': 'main',
+            'feature_groups': ['MockFeatureGroup'],
+            'feature_names': [['all']]}],
+    'output': {
+        'output_location': 'local',
+        'config': {'target_path': 'output/mockdata'},
+        'data_format': 'csv',
+        'compress': False}}
+
+
+class TestConfigValidatorInput(unittest.TestCase):
+    def setUp(self):
+        self.valid_input_formats = ["csv", "csv.gz"]
+        self.valid_output_formats = ["csv"]
+        self.mock_config = deepcopy(MOCK_CONFIG)
+
+    def test_validate_mock(self):
+        validator = ConfigValidator(self.mock_config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        validator.validate()
+
+    def test_validate_sftp(self):
+        config = self.mock_config
+        config['input']['data_type'] = 'sftp'
+        config['input']['config'] = {
+            "sftp_host": "mock_host",
+            "sftp_source_path": "mock_source_path",
+            "sftp_username": "mock_username",
+            "sftp_private_key": "mock_private_key",
+            "sftp_target_path": "mock_target_path"
+        }
+        validator = ConfigValidator(config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        validator.validate()
+
+    def test_validate_invalid_sftp(self):
+        config = self.mock_config
+        config['input']['data_type'] = 'sftp'
+        config['input']['config'] = {
+            "sftp_host": "mock_host",
+            "sftp_source_path": "mock_source_path",
+            "sftp_username": "mock_username",
+            "sftp_target_path": "mock_target_path"
+        }
+        validator = ConfigValidator(config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        with self.assertRaises(ValueError):
+            validator.validate()
+
+    def test_validate_local(self):
+        config = self.mock_config
+        config['input']['data_type'] = 'local'
+        config['input']['config'] = {
+            "source_path": "mockdata/mockdata"
+        }
+        validator = ConfigValidator(config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        validator.validate()
+
+    def test_validate_local_wrong_source_path(self):
+        config = self.mock_config
+        config['input']['data_type'] = 'local'
+        config['input']['config'] = {
+            "source_path": "xyz"
+        }
+        validator = ConfigValidator(config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        with self.assertRaises(ValueError):
+            validator.validate()
+
+    def test_validate_invalid_input_format(self):
+        config = self.mock_config
+        config.pop('input')
+        validator = ConfigValidator(config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        with self.assertRaises(ValueError):
+            validator.validate()
+
+
+class TestConfigValidatorConfigurations(unittest.TestCase):
+    def setUp(self):
+        self.valid_input_formats = ["csv", "csv.gz"]
+        self.valid_output_formats = ["csv"]
+        self.mock_config = deepcopy(MOCK_CONFIG)
+
+    def test_validate_configurations(self):
+        self.mock_config['configurations']['df_type'] = 'invalid'
+        validator = ConfigValidator(self.mock_config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        with self.assertRaises(ValueError):
+            validator.validate()
+
+
+class TestConfigValidatorFeatures(unittest.TestCase):
+    def setUp(self):
+        self.valid_input_formats = ["csv", "csv.gz"]
+        self.valid_output_formats = ["csv"]
+        self.mock_config = deepcopy(MOCK_CONFIG)
+
+    def test_is_feature_empty(self):
+        self.mock_config['features'] = []
+        validator = ConfigValidator(self.mock_config, self.valid_input_formats,
+                                    self.valid_output_formats)
+        with self.assertRaises(ValueError):
+            validator.validate()