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

Issue:
Sampling architecture
Added user and data sampling mechanisms. User sampling mechanism contains option to choose users by fraction, count and IDs Data sampling mechanisms include choosing data between time ranges, by count and by fraction. 


Discussion / hints:

Fix patch:
diff --git a/config.yaml b/config.yaml
index e7b9097..da635f3 100644
--- a/config.yaml
+++ b/config.yaml
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: local # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         # In case of sftp, use the following format
         # sftp_host:
@@ -25,6 +25,36 @@ input:
 
 configurations:
     df_type: 'pandas'
+    #user_sampling:
+        ## Possible methods: fraction, count, userid
+    #    method: fraction
+    #    config:
+    #        fraction: 0.8
+        #method: count
+        #config:
+        #    count: 2
+        #method: userid
+        #config:
+        #    userids:
+        #        - 2a02e53a-951e-4fd0-b47f-195a87096bd0
+    ## TODO: For future
+    #data_sampling:
+        ## Possible methods:  time, count, fraction
+        ## starttime and endtime format is dd-mm-yyyy hh:mm:ss in UTC timezone
+        ## It is possible to have multiple time ranges. See below Example
+        #method: time
+        #config:
+        #- starttime: 2018-11-22 00:00:00
+        #  endtime:  2018-11-26 00:00:00
+        #  time_column: value.time
+        #- starttime: 2018-12-27 00:00:00
+        #  time_column: value.time
+        #method: count
+        #config:
+        #   count: 100
+        #method: fraction
+        #config:
+        #   fraction: 0.3
 
 features:
     - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
@@ -39,8 +69,4 @@ output:
     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
+    compress: false
\ No newline at end of file
diff --git a/config.yaml.template b/config.yaml.template
index e25b4e7..152faa7 100644
--- a/config.yaml.template
+++ b/config.yaml.template
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: mock # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         ## In case of sftp, use the following format
         # sftp_host:
@@ -24,7 +24,37 @@ input:
     data_format: csv
 
 configurations:
-    df_type: 'pandas' # can be pandas or spark
+    df_type: 'pandas'
+    #user_sampling:
+        ## Possible methods: fraction, count, userid
+    #    method: fraction
+    #    config:
+    #        fraction: 0.8
+        #method: count
+        #config:
+        #    count: 2
+        #method: userid
+        #config:
+        #    userids:
+        #        - 2a02e53a-951e-4fd0-b47f-195a87096bd0
+    ## TODO: For future
+    data_sampling:
+        ## Possible methods:  time, count, fraction
+        ## starttime and endtime format is dd-mm-yyyy hh:mm:ss in UTC timezone
+        ## It is possible to have multiple time ranges. See below Example
+        #method: time
+        #config:
+        #- starttime: 2018-11-22 00:00:00
+        #  endtime:  2018-11-26 00:00:00
+        #  time_column: value.time
+        #- starttime: 2018-12-27 00:00:00
+        #  time_column: value.time
+        #method: count
+        #config:
+        #   count: 100
+        #method: fraction
+        #config:
+        #   fraction: 0.3
 
 features:
     - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
diff --git a/mockdata b/mockdata
index c791236..17e9ff9 160000
--- a/mockdata
+++ b/mockdata
@@ -1,1 +1,1 @@
-Subproject commit c7912366b946edc8a7e51bbde6b41ae4a248c995
+Subproject commit 17e9ff951ea125dea0c9e65a5695d707749fdc84
diff --git a/radarpipeline/common/utils.py b/radarpipeline/common/utils.py
index afd8f75..2b935d9 100644
--- a/radarpipeline/common/utils.py
+++ b/radarpipeline/common/utils.py
@@ -10,11 +10,16 @@
 import yaml
 from strictyaml import load, Map, Int, Str, Seq, Bool, Optional
 from strictyaml import YAMLError, CommaSeparated, MapPattern
+from dateutil import parser
 
 import ntpath
 import posixpath
 
 from radarpipeline.common import constants
+import unittest
+from radarpipeline.project.sparkengine import SparkEngine
+import pyspark.sql.functions as f
+from pyspark.sql.types import TimestampType
 
 
 def read_yaml(yaml_file_path: str) -> Dict[str, Any]:
@@ -145,12 +150,20 @@ def get_yaml_schema() -> Map:
             Optional("version"): Str()
         }),
         "input": Map({
-            "data_type": Str(),
+            "source_type": Str(),
             "config": MapPattern(Str(), Str()),
             "data_format": Str()
         }),
         "configurations": Map({
-            "df_type": Str()
+            "df_type": Str(),
+            Optional("user_sampling"): Map({
+                "method": Str(),
+                "config": MapPattern(Str(), Seq(Str()) | Str()),
+            }),
+            Optional("data_sampling"): Map({
+                "method": Str(),
+                "config": MapPattern(Str(), Str()) | Seq(MapPattern(Str(), Str())),
+            }),
         }),
         "features": Seq(Map({
             "location": Str(),
@@ -165,6 +178,7 @@ def get_yaml_schema() -> Map:
             "compress": Bool()
         }),
         Optional("spark_config"): Map({
+            Optional("spark_master", default="local"): Str(),
             Optional("spark.executor.instances", default=4): Int(),
             Optional("spark.executor.cores", default=4): Int(),
             Optional("spark.executor.memory", default='10g'): Str(),
@@ -250,3 +264,54 @@ def get_write_file_attr(feature_name, output_dir, data_format, compression):
         raise ValueError(f"Invalid data format {data_format} specified \
             for spark writer")
     return file_path
+
+
+def get_hash(array : List) -> int:
+    """
+    Returns the hash of the array
+
+    Parameters
+    ----------
+    array : list
+        List of values
+
+    Returns
+    -------
+    str
+        Hash of the array
+    """
+    return hash(tuple(array))
+
+
+def preprocess_time_data(data):
+    time_cols = ["value.time", "value.timeReceived", "value.dateTime",
+                 "value.timeCompleted", "value.timeNotification"]
+    for i, col in enumerate(time_cols):
+        if col in data.columns:
+            data = data.withColumn(col, data[f"`{col}`"].cast(TimestampType()))
+            data.withColumn(col, f.from_unixtime(
+                f.unix_timestamp(f"`{col}`")))
+    return data
+
+
+def convert_str_to_time(time):
+    try:
+        return parser.parse(time)
+    except ValueError:
+        raise ValueError(
+            "Invalid value for the key: time. It should be a valid time format"
+        )
+
+
+class PySparkTestCase(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.spark_engine = SparkEngine()
+        cls.spark = cls.spark_engine.initialize_spark_session()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.spark_engine.close_spark_session()
+
+    def preprocess_data(self, data):
+        return preprocess_time_data(data)
diff --git a/radarpipeline/datalib/radar_data.py b/radarpipeline/datalib/radar_data.py
index b628d3d..d60bf01 100644
--- a/radarpipeline/datalib/radar_data.py
+++ b/radarpipeline/datalib/radar_data.py
@@ -94,10 +94,9 @@ def get_combined_data_by_variable(
         # Combine the all data for each variable
         for var in variable_dict:
             if len(variable_dict[var]) > 0:
-                if self.df_type == "spark":
-                    combined_df = utils.combine_pyspark_dfs(variable_dict[var])
-                else:
-                    combined_df = pd.concat(variable_dict[var], ignore_index=True)
+                combined_df = utils.combine_pyspark_dfs(variable_dict[var])
+                if self.df_type == "pandas":
+                    combined_df = combined_df.toPandas()
                 variable_data_list.append(combined_df)
 
         if is_only_one_var:
diff --git a/radarpipeline/datalib/radar_variable_data.py b/radarpipeline/datalib/radar_variable_data.py
index 0cdb032..d02c8ae 100644
--- a/radarpipeline/datalib/radar_variable_data.py
+++ b/radarpipeline/datalib/radar_variable_data.py
@@ -2,8 +2,10 @@
 import logging
 import pandas as pd
 import pyspark.sql.functions as f
+from pyspark.sql.types import TimestampType
 
 from radarpipeline.datalib.abc import Data
+from radarpipeline.common.utils import preprocess_time_data
 from radarpipeline.datatypes import DataType
 
 logger = logging.getLogger(__name__)
@@ -16,10 +18,13 @@ class RadarVariableData(Data):
 
     _data: DataType
 
-    def __init__(self, data: DataType, df_type: str = "pandas") -> None:
+    def __init__(self, data: DataType, df_type: str = "pandas",
+                 data_sampler=None) -> None:
         self._data = data
         self.df_type = df_type
         self._preprocess_data()
+        if data_sampler is not None:
+            self._data = data_sampler.sample_data(self._data)
 
     def get_data(self) -> DataType:
         return self._data
@@ -31,42 +36,13 @@ def get_data_keys(self) -> List[str]:
         return list(self._data.columns)
 
     def get_data_size(self) -> int:
-        if self.df_type == "pandas":
-            return len(self._data.index)
-        else:
-            return int(self._data.count())
+        return int(self._data.count())
 
     def _preprocess_data(self) -> None:
         """
         Converts all time value columns to datetime format
         """
-
-        if self.df_type == "spark":
-            if "value.time" in self.get_data_keys():
-                self._data = self._data.withColumn(
-                    "value.time", f.to_date(self._data["`value.time`"])
-                )
-            if "value.timeReceived" in self.get_data_keys():
-                self._data = self._data.withColumn(
-                    "value.timeReceived", f.to_date(self._data["`value.timeReceived`"])
-                )
-            if "value.dateTime" in self.get_data_keys():
-                self._data = self._data.withColumn(
-                    "value.dateTime", f.to_date(self._data["`value.dateTime`"])
-                )
-        else:
-            try:
-                if "value.time" in self.get_data_keys():
-                    self._data["value.time"] = pd.to_datetime(
-                        self._data["value.time"].astype(str), unit="s"
-                    )
-                if "value.timeReceived" in self.get_data_keys():
-                    self._data["value.timeReceived"] = pd.to_datetime(
-                        self._data["value.timeReceived"].astype(str), unit="s"
-                    )
-                if "value.dateTime" in self.get_data_keys():
-                    self._data["value.dateTime"] = pd.to_datetime(
-                        self._data["value.dateTime"].astype(str), unit="s"
-                    )
-            except ValueError:
-                logger.warning("Unable to convert time columns to datetime format")
+        try:
+            self._data = preprocess_time_data(self._data)
+        except ValueError:
+            logger.warning("Unable to convert time columns to datetime format")
diff --git a/radarpipeline/features/feature_group.py b/radarpipeline/features/feature_group.py
index 60d2f3e..551a84e 100644
--- a/radarpipeline/features/feature_group.py
+++ b/radarpipeline/features/feature_group.py
@@ -65,7 +65,7 @@ def get_all_features(self, data: RadarData) -> Tuple[List[str], List[DataType]]:
         feature_values = []
         preprocessed_data = self.preprocess(data)
         for feature in self.features:
-            print(feature.name)
+            logger.info(f"Computing feature {feature.name}")
             feature_names.append(feature.name)
             preprocessed_feature = feature.preprocess(preprocessed_data)
             feature_values.append(feature.calculate(preprocessed_feature))
diff --git a/radarpipeline/io/__init__.py b/radarpipeline/io/__init__.py
index dade1ac..cafc0ae 100644
--- a/radarpipeline/io/__init__.py
+++ b/radarpipeline/io/__init__.py
@@ -1,4 +1,6 @@
 from radarpipeline.io.abc import DataReader, SchemaReader
-from radarpipeline.io.reader import AvroSchemaReader, SparkCSVDataReader, Reader
+from radarpipeline.io.reader import AvroSchemaReader, Reader, SparkCSVDataReader
 from radarpipeline.io.downloader import SftpDataReader
 from radarpipeline.io.writer import *
+from radarpipeline.io.ingestion import CustomDataReader
+from radarpipeline.io.sampler import UserSampler, DataSampler
diff --git a/radarpipeline/io/abc.py b/radarpipeline/io/abc.py
index 28d176f..711e201 100644
--- a/radarpipeline/io/abc.py
+++ b/radarpipeline/io/abc.py
@@ -55,3 +55,12 @@ def __init__(self, features: Dict[str, DataType], output_dir: str) -> None:
     @abstractmethod
     def write_data(self) -> None:
         pass
+
+
+class Sampler(ABC):
+    """
+    Abstract class for sampling the RADAR data
+    """
+
+    def __init__(self, config) -> None:
+        self.config = config
diff --git a/radarpipeline/io/downloader.py b/radarpipeline/io/downloader.py
index e128320..6b8b220 100644
--- a/radarpipeline/io/downloader.py
+++ b/radarpipeline/io/downloader.py
@@ -101,13 +101,14 @@ def _fetch_data(self, root_path, sftp_source_path, included_var_cat, uid):
                                                            src_file),
                                                        preserve_mtime=True)
                         except FileNotFoundError:
-                            print("Folder not found: " + dir_path + "/" + src_file)
+                            logger.warning("Folder not found: " + dir_path
+                                           + "/" + src_file)
                             continue
                         except EOFError:
-                            print("EOFError: " + dir_path + "/" + src_file)
+                            logger.warning("EOFError: " + dir_path + "/" + src_file)
                             continue
         except FileNotFoundError:
-            print("Folder not found: " + uid)
+            logger.warning("Folder not found: " + uid)
             return
         sftp.close()
 
diff --git a/radarpipeline/io/ingestion.py b/radarpipeline/io/ingestion.py
new file mode 100644
index 0000000..e56b31c
--- /dev/null
+++ b/radarpipeline/io/ingestion.py
@@ -0,0 +1,39 @@
+import logging
+
+from radarpipeline.io.reader import Reader
+from radarpipeline.project.sparkengine import SparkEngine
+
+from typing import Dict
+
+logger = logging.getLogger(__name__)
+
+
+class CustomDataReader():
+    def __init__(self, input_config, variables, source_type="local", data_format="csv",
+                 df_type="pandas") -> None:
+        self.variables = variables
+        self.data_format = data_format
+        self.source_type = source_type
+        self.config = self.modify_config(input_config, data_format)
+        self.sparkengine = SparkEngine()
+        self.spark = self.sparkengine.initialize_spark_session()
+        self.data_reader = Reader(self.spark, self.config, variables, df_type)
+
+    def modify_config(self, input_config, data_format) -> Dict:
+        """
+        Modify the input configuration to include the variables of interest
+        """
+        config = {'input': {}, "configurations": {}}
+        config['input'] = input_config
+        config['input']['data_format'] = data_format
+        config['input']['source_type'] = self.source_type
+        config['configurations']['df_type'] = "pandas"
+        config['configurations']['user_sampling'] = None
+        config['configurations']['data_sampling'] = None
+        return config
+
+    def read_data(self):
+        return self.data_reader.read_data()
+
+    def close_session(self):
+        self.sparkengine.close_spark_session()
diff --git a/radarpipeline/io/reader.py b/radarpipeline/io/reader.py
index 5defb82..4cffa40 100644
--- a/radarpipeline/io/reader.py
+++ b/radarpipeline/io/reader.py
@@ -2,29 +2,29 @@
 import logging
 import os
 from glob import glob
+import gzip
 import re
 from typing import Any, Dict, List, Optional, Union
 
 import pyspark.sql as ps
-from pyspark.sql import SparkSession
+from pyspark.sql import SparkSession, DataFrame
 from pyspark.sql.types import StructField, StructType
 from pyspark.sql.utils import IllegalArgumentException
 
 from radarpipeline.common import constants
 from radarpipeline.datalib import RadarData, RadarUserData, RadarVariableData
 from radarpipeline.io.abc import DataReader, SchemaReader
+from radarpipeline.io.sampler import UserSampler, DataSampler
+from radarpipeline.common.utils import get_hash
 
 import avro
 from avro.datafile import DataFileReader, DataFileWriter
 from avro.io import DatumReader, DatumWriter
 from avro.schema import RecordSchema, Field, PrimitiveSchema, UnionSchema, Schema
-
-from multiprocessing import Pool
-from functools import partial
 from datetime import datetime
 
 from collections import Counter
-
+from functools import reduce, partial
 
 logger = logging.getLogger(__name__)
 
@@ -33,27 +33,82 @@ class Schemas(object):
     def __init__(self, original_schema, original_schema_keys):
         self.original_schema = original_schema
         self.original_schema_hash = self._get_schema_hash(original_schema_keys)
-        self.counterdict = Counter({self.original_schema_hash: 1})
         self.hashdict = {self.original_schema_hash: original_schema}
 
     def _get_schema_hash(self, schema_keys):
-        return hash(frozenset(schema_keys))
+        return get_hash(schema_keys)
 
     def is_original_schema(self, schema_keys):
         return self._get_schema_hash(schema_keys) == self.original_schema_hash
 
-    def get_schema(self):
-        most_freq_schema_hash = self.counterdict.most_common(1)[0][0]
-        return self.hashdict[most_freq_schema_hash]
+    def is_schema_present(self, schema_keys):
+        return self._get_schema_hash(schema_keys) in self.hashdict
+
+    def is_schema_hash_present(self, schema_hash):
+        return schema_hash in self.hashdict
+
+    def get_schema(self, schema_keys=None):
+        if schema_keys is None:
+            return self.original_schema
+        return self.hashdict[self._get_schema_hash(schema_keys)]
+
+    def get_schema_by_hash(self, schema_hash):
+        return self.hashdict[schema_hash]
 
     def add_schema(self, schema_keys, schema):
         schema_hash = self._get_schema_hash(schema_keys)
-        if schema_hash not in self.hashdict:
-            self.hashdict[schema_hash] = schema
-        self.counterdict[schema_hash] += 1
+        self.hashdict[schema_hash] = schema
+
+
+class Reader():
+    '''
+    Class for reading data from a file
+    Reader(source_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, spark_session: ps.SparkSession,
+                 config: Dict, required_data: List[str], df_type: str = "pandas"):
+        """_summary_
+
+        Args:
+            spark_session (ps.SparkSession): spark session instance
+            config (Dict): Configuration data from the config.yaml file
+            required_data (List[str]): List of required data
+            df_type (str, optional): Type of dataframe format. Defaults to "pandas".
+        """
+        self.config = config
+        self.source_type = self.config["input"]["data_format"]
+        self.required_data = required_data
+        self.df_type = df_type
+        if self.config["configurations"]['user_sampling'] is None:
+            self.user_sampler = None
+        else:
+            self.user_sampler = UserSampler(self.config["configurations"]
+                                            ['user_sampling'])
+        if self.config["configurations"]['data_sampling'] is None:
+            self.data_sampler = None
+        else:
+            self.data_sampler = DataSampler(self.config["configurations"]
+                                            ['data_sampling'])
 
-    def update_schema_counter(self, schema_keys):
-        self.counterdict[self._get_schema_hash(schema_keys)] += 1
+        if self.source_type in ['csv', 'csv.gz']:
+            self.reader_class = SparkCSVDataReader(spark_session, config,
+                                                   required_data, df_type,
+                                                   self.user_sampler, self.data_sampler)
+        else:
+            raise NotImplementedError("Only csv data type is supported for now")
+
+    def read_data(self):
+        self.data = self.reader_class.read_data()
+        return self.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)
 
 
 class SparkCSVDataReader(DataReader):
@@ -61,8 +116,9 @@ class SparkCSVDataReader(DataReader):
     Read CSV data from local directory using pySpark
     """
 
-    def __init__(self, config: Dict, required_data: List[str], df_type: str = "pandas",
-                 spark_config: Dict = {}):
+    def __init__(self, spark_session: ps.SparkSession,
+                 config: Dict, required_data: List[str], df_type: str = "pandas",
+                 user_sampler: UserSampler = None, data_sampler: DataSampler = None):
         super().__init__(config)
         self.source_formats = {
             # RADAR_OLD: uid/variable/yyyymmdd_hh00.csv.gz
@@ -72,91 +128,14 @@ def __init__(self, config: Dict, required_data: List[str], df_type: str = "panda
             "RADAR_NEW": re.compile(r"""^[\w-]+/([\w]+)/
                                     [\d]+/([\d]+.csv.gz$|schema-\1.json$)""", re.X),
         }
-        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',
-                                'spark.log.level': "OFF"}
         self.required_data = required_data
         self.df_type = df_type
-        self.source_path = self.config['config'].get("source_path", "")
-        self.spark_config = default_spark_config
+        self.source_path = self.config['input']['config'].get("source_path", "")
+        self.user_sampler = user_sampler
+        self.data_sampler = data_sampler
         self.schema_reader = AvroSchemaReader()
-        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:
-        """
-        Initializes and returns a SparkSession
-
-        Returns
-        -------
-        SparkSession
-            A SparkSession object
-        """
-
-        """
-        Spark configuration documentation:
-        https://spark.apache.org/docs/latest/configuration.html
-
-        `spark.executor.instances` is the number of executors to
-        launch for an application.
-
-        `spark.executor.cores` is the number of cores to =
-        use on each executor.
-
-        `spark.executor.memory` is the amount of memory to
-        use per executor process.
-
-        `spark.driver.memory` is the amount of memory to use for the driver process,
-        i.e. where SparkContext is initialized, in MiB unless otherwise specified.
-
-        `spark.memory.offHeap.enabled` is to enable off-heap memory allocation
-
-        `spark.memory.offHeap.size` is the absolute amount of memory which can be used
-        for off-heap allocation, in bytes unless otherwise specified.
-
-        `spark.driver.maxResultSize` is the limit of total size of serialized results of
-        all partitions for each Spark action (e.g. collect) in bytes.
-        Should be at least 1M, or 0 for unlimited.
-        """
-        spark = (
-            SparkSession.builder.master("local").appName("radarpipeline")
-            .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'])
-            .config('spark.log.level',
-                    self.spark_config['spark.log.level'])
-            .getOrCreate()
-        )
-        spark._jsc.setLogLevel(self.spark_config['spark.log.level'])
-        spark.sparkContext.setLogLevel("OFF")
-        # Enable Apache Arrow for optimizations in Spark to Pandas conversion
-        spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
-        # Fallback to use non-Arrow conversion in case of errors
-        spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", "true")
-        # For further reading:
-        # https://spark.apache.org/docs/3.0.1/sql-pyspark-pandas-with-arrow.html
-        logger.info("Spark Session created")
-        return spark
-
-    def close_spark_session(self):
-        self.spark.stop()
+        self.spark = spark_session
+        self.unionByName = partial(DataFrame.unionByName, allowMissingColumns=True)
 
     def _get_source_type(self, source_path):
         """
@@ -199,6 +178,19 @@ def read_data(self) -> RadarData:
                     source_path_item, user_data_dict)
         return radar_data
 
+    def _filter_files_by_headers(self, data_files):
+        file_dict = {}
+        for file in data_files:
+            with gzip.open(file, 'rb') as f:
+                columns = f.readline().decode("utf-8").split(",")
+                f.close()
+            column_hash = get_hash(columns)
+            if column_hash in file_dict:
+                file_dict[column_hash].append(file)
+            else:
+                file_dict[column_hash] = [file]
+        return file_dict
+
     def _read_variable_data_files(
         self,
         data_files: List[str],
@@ -220,51 +212,60 @@ def _read_variable_data_files(
         RadarVariableData
             A RadarVariableData object containing all the read data
         """
+        dfs = []
+        file_dict = self._filter_files_by_headers(data_files)
         if schema:
-            df = self.spark.read.load(
-                data_files,
-                format="csv",
-                header=True,
-                schema=schema.get_schema(),
-                enforceSchema="false",
-                encoding=constants.ENCODING,
-            )
+            for column_hash in file_dict.keys():
+                if schema.is_schema_hash_present(column_hash):
+                    df = self.spark.read.load(
+                        file_dict[column_hash],
+                        format="csv",
+                        header=True,
+                        schema=schema.get_schema_by_hash(column_hash),
+                        enforceSchema="false",
+                        encoding=constants.ENCODING,
+                    )
+                    dfs.append(df)
+                else:
+                    df = self.spark.read.load(
+                        file_dict[column_hash],
+                        format="csv",
+                        header=True,
+                        inferSchema="true",
+                        encoding=constants.ENCODING,
+                    )
+                    inferred_schema = df.schema
+                    schema.add_schema(df.columns, inferred_schema)
+                    dfs.append(df)
         else:
-            df = self.spark.read.load(
-                data_files,
-                format="csv",
-                header=True,
-                inferSchema="true",
-                encoding=constants.ENCODING,
-            )
-
-        if self.df_type == "pandas":
-            try:
-                df = df.toPandas()
-                schema.update_schema_counter(df.columns)
-            except Exception:
-                logger.warning("Failed to convert to pandas dataframe. "
-                               "inferring schema")
+            for column_hash in file_dict:
                 df = self.spark.read.load(
-                    data_files,
+                    file_dict[column_hash],
                     format="csv",
                     header=True,
                     inferSchema="true",
                     encoding=constants.ENCODING,
                 )
-                inferred_schema = df.schema
-                schema.add_schema(df.columns, inferred_schema)
-                df = df.toPandas()
-
-        variable_data = RadarVariableData(df, self.df_type)
-
+                dfs.append(df)
+
+        # Spark Join all the dfs
+        # check if dfs are empty
+        if len(dfs) == 0:
+            # creating empty spark df
+            df = self.spark.createDataFrame([], schema=schema.get_schema())
+            variable_data = RadarVariableData(df, self.df_type)
+        else:
+            df = reduce(self.unionByName, dfs)
+            variable_data = RadarVariableData(df, self.df_type,
+                                              data_sampler=self.data_sampler)
         return variable_data
 
     def _read_data_from_old_format(self, source_path: str, user_data_dict: dict):
-        for uid in os.listdir(source_path):
-            # Skip hidden files
-            if uid[0] == ".":
-                continue
+        uids = os.listdir(source_path)
+        uids = self._remove_hidden_dirs(uids)
+        if self.user_sampler is not None:
+            uids = self.user_sampler.sample_uids(uids)
+        for uid in uids:
             logger.info(f"Reading data for user: {uid}")
             variable_data_dict = {}
             for dirname in self.required_data:
@@ -297,7 +298,11 @@ def _read_data_from_old_format(self, source_path: str, user_data_dict: dict):
 
     def _read_data_from_new_format(self, source_path: str, user_data_dict: dict):
         # RADAR_NEW: uid/variable/yyyymm/yyyymmdd.csv.gz
-        for uid in os.listdir(source_path):
+        uids = os.listdir(source_path)
+        uids = self._remove_hidden_dirs(uids)
+        if self.user_sampler is not None:
+            uids = self.user_sampler.sample_uids(uids)
+        for uid in uids:
             # Skip hidden files
             if uid[0] == ".":
                 continue
@@ -335,6 +340,9 @@ def _read_data_from_new_format(self, source_path: str, user_data_dict: dict):
         radar_data = RadarData(user_data_dict, self.df_type)
         return radar_data, user_data_dict
 
+    def _remove_hidden_dirs(self, uids):
+        return [uid for uid in uids if uid[0] != "."]
+
 
 class AvroSchemaReader(SchemaReader):
     """
@@ -387,17 +395,17 @@ def _get_schema(self, schema_dir, schema_dir_base) -> StructType:
         schema_file = os.path.join(
             schema_dir, f"schema-{schema_dir_base}.json"
         )
-        schema_dict = json.load(
+        schema_content = json.load(
             open(
                 schema_file,
                 "r",
                 encoding=constants.ENCODING,
             )
         )
-        avro_schema = avro.schema.parse(json.dumps(schema_dict))
-        schema_dict = self._recursive_schema_loader(avro_schema)
+        avro_schema = avro.schema.parse(json.dumps(schema_content))
+        schema_content_dict = self._recursive_schema_loader(avro_schema)
 
-        schema, schema_keys = self._to_structtype(schema_dict)
+        schema, schema_keys = self._to_structtype(schema_content_dict)
         return schema, schema_keys
 
     def _add_new_schema(self, schema_dir_base, schema):
@@ -484,13 +492,13 @@ def _recursive_schema_loader(self, record_schema, precursor="", schema_dict={}):
         else:
             return {}
 
-    def _get_field(self, data_type: Union[str, Dict, List]) -> Any:
+    def _get_field(self, source_type: Union[str, Dict, List]) -> Any:
         """
         Returns a Spark data type for a given data type
 
         Parameters
         ----------
-        data_type : Union[str, Dict]
+        source_type : Union[str, Dict]
             Data type to convert to a Spark data type
 
         Returns
@@ -499,12 +507,12 @@ def _get_field(self, data_type: Union[str, Dict, List]) -> Any:
             A Spark data type
         """
 
-        if type(data_type) is dict:
-            spark_data_type = self._get_data_type_from_dict(data_type)
-        elif type(data_type) is list:
-            spark_data_type = self._get_superior_type_from_list(data_type)
+        if type(source_type) is dict:
+            spark_data_type = self._get_data_type_from_dict(source_type)
+        elif type(source_type) is list:
+            spark_data_type = self._get_superior_type_from_list(source_type)
         else:
-            spark_data_type = self._get_data_type_from_mapping(data_type)
+            spark_data_type = self._get_data_type_from_mapping(source_type)
 
         return spark_data_type
 
@@ -514,13 +522,13 @@ def _resolve_union_schema(self, union_schemas: List[Schema]):
             list_type.append(schema.type)
         return self._get_superior_type_from_list(list_type)
 
-    def _handle_unknown_data_type(self, data_type: Union[str, Dict, List]) -> Any:
+    def _handle_unknown_data_type(self, source_type: Union[str, Dict, List]) -> Any:
         """
         Handles unknown data types
 
         Parameters
         ----------
-        data_type : Union[str, Dict]
+        source_type : Union[str, Dict]
             Data type to handle
 
         Returns
@@ -529,16 +537,16 @@ def _handle_unknown_data_type(self, data_type: Union[str, Dict, List]) -> Any:
             A Spark data type
         """
 
-        logger.warning(f"Unknown data type: {data_type}. Returning String type.")
+        logger.warning(f"Unknown data type: {source_type}. Returning String type.")
         return constants.STRING_TYPE
 
-    def _get_data_type_from_mapping(self, data_type: Union[str, Dict, List]) -> Any:
+    def _get_data_type_from_mapping(self, source_type: Union[str, Dict, List]) -> Any:
         """
         Returns a Spark data type for a given data type
 
         Parameters
         ----------
-        data_type : str
+        source_type : str
             Data type to convert to a Spark data type
 
         Returns
@@ -547,20 +555,20 @@ def _get_data_type_from_mapping(self, data_type: Union[str, Dict, List]) -> Any:
             A Spark data type
         """
 
-        if data_type in constants.DATA_TYPE_MAPPING:
-            spark_data_type = constants.DATA_TYPE_MAPPING[data_type]
+        if source_type in constants.DATA_TYPE_MAPPING:
+            spark_data_type = constants.DATA_TYPE_MAPPING[source_type]
         else:
-            spark_data_type = self._handle_unknown_data_type(data_type)
+            spark_data_type = self._handle_unknown_data_type(source_type)
 
         return spark_data_type
 
-    def _get_data_type_from_dict(self, data_type: Dict) -> Any:
+    def _get_data_type_from_dict(self, source_type: Dict) -> Any:
         """
         Returns a Spark data type for a given data type
 
         Parameters
         ----------
-        data_type : Dict
+        source_type : Dict
             Data type to convert to a Spark data type
 
         Returns
@@ -569,10 +577,10 @@ def _get_data_type_from_dict(self, data_type: Dict) -> Any:
             A Spark data type
         """
 
-        if "type" in data_type:
-            return self._get_field(data_type["type"])
+        if "type" in source_type:
+            return self._get_field(source_type["type"])
         else:
-            return self._handle_unknown_data_type(data_type)
+            return self._handle_unknown_data_type(source_type)
 
     def _get_superior_type_from_list(self, data_type_list: List[Any]) -> Any:
         """
@@ -594,13 +602,14 @@ def _get_superior_type_from_list(self, data_type_list: List[Any]) -> Any:
         if "null" in spark_data_type_list:
             spark_data_type_list.remove("null")
 
-        for index, data_type in enumerate(spark_data_type_list):
-            if type(data_type) is dict:
-                spark_data_type_list[index] = self._get_data_type_from_dict(data_type)
-            elif data_type in constants.DATA_TYPE_MAPPING:
-                spark_data_type_list[index] = constants.DATA_TYPE_MAPPING[data_type]
+        for index, source_type in enumerate(spark_data_type_list):
+            if type(source_type) is dict:
+                spark_data_type_list[index] = self._get_data_type_from_dict(source_type)
+            elif source_type in constants.DATA_TYPE_MAPPING:
+                spark_data_type_list[index] = constants.DATA_TYPE_MAPPING[source_type]
             else:
-                spark_data_type_list[index] = self._handle_unknown_data_type(data_type)
+                spark_data_type_list[index] = self._handle_unknown_data_type(
+                    source_type)
 
         if len(data_type_list) == 0:
             return constants.STRING_TYPE
@@ -679,46 +688,3 @@ 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/io/sampler.py b/radarpipeline/io/sampler.py
new file mode 100644
index 0000000..92a5838
--- /dev/null
+++ b/radarpipeline/io/sampler.py
@@ -0,0 +1,89 @@
+from radarpipeline.io.abc import Sampler
+from typing import Any, Dict, List, Optional, Union
+from random import sample
+import pyspark.sql as ps
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class UserSampler(Sampler):
+
+    def __init__(self, config: Dict) -> None:
+        super().__init__(config)
+
+    def sample_uids(self, uid_list) -> None:
+        if self.config['method'] == "fraction":
+            fraction = self.config['config']['fraction']
+            # sample fraction of the uids
+            return self._sample_list(uid_list, round(len(uid_list) * fraction))
+        elif self.config['method'] == "count":
+            count = self.config['config']['count']
+            # sample count of the uids
+            return self._sample_list(uid_list, count)
+        elif self.config['method'] == "userid":
+            sampled_uids = self.config['config']['userids']
+            for sampled_uid in sampled_uids:
+                if sampled_uid not in uid_list:
+                    logger.warning(f"User id {sampled_uid} not found in the data")
+                    sampled_uids.remove(sampled_uid)
+            return sampled_uids
+        else:
+            raise ValueError("Invalid method")
+
+    def _sample_list(self, uid_list, number):
+        return sample(uid_list, number)
+
+
+class DataSampler(Sampler):
+    def __init__(self, config: Dict) -> None:
+        super().__init__(config)
+
+    def sample_data(self, df: ps.DataFrame) -> Optional[ps.DataFrame]:
+        if self.config['method'] == "fraction":
+            fraction = self.config['config']['fraction']
+            # sample fraction of the data
+            return self._sample_data(df, fraction)
+        elif self.config['method'] == "count":
+            count = self.config['config']['count']
+            # sample count of the data
+            return self._sample_data(df, count / df.count())
+        elif (self.config['method'] == "time"
+              and not (type(self.config['config']) is list
+                       and len(self.config['config']) != 1)):
+            if len(self.config['config']) == 1:
+                self.config['config'] = self.config['config'][0]
+            starttime = self.config['config'].get('starttime', None)
+            endtime = self.config['config'].get('endtime', None)
+            time_col = self.config['config'].get('time_col', "value.time")
+            # check if time_col is present in df
+            if time_col not in df.columns:
+                raise ValueError(f"Column {time_col} not found in the dataframe")
+            return self._sample_data_by_time(df, starttime, endtime, time_col)
+        elif self.config['method'] == "time" and type(self.config['config']) is list:
+            sampled_dfs = []
+            for time_config in self.config['config']:
+                starttime = time_config.get('starttime', None)
+                endtime = time_config.get('endtime', None)
+                time_col = time_config.get('time_col', "value.time")
+                # check if time_col is present in df
+                if time_col not in df.columns:
+                    raise ValueError(f"Column {time_col} not found in the dataframe")
+                sampled_dfs.append(
+                    self._sample_data_by_time(df, starttime, endtime, time_col))
+            # combined sampled_dfs into one dataframe and deduplicate it
+            return sampled_dfs[0].union(*sampled_dfs[1:]).distinct()
+        else:
+            raise ValueError("Invalid method")
+
+    def _sample_data(self, df, fraction):
+        return df.sample(fraction=fraction, withReplacement=True)
+
+    def _sample_data_by_time(self, df, starttime, endtime, time_col):
+        if endtime is None:
+            return df.filter(df[f"`{time_col}`"] >= starttime)
+        elif starttime is None:
+            return df.filter(df[f"`{time_col}`"] < endtime)
+        else:
+            return df.filter(df[f"`{time_col}`"].between(starttime, endtime))
diff --git a/radarpipeline/io/writer.py b/radarpipeline/io/writer.py
index 237da8f..7d1a103 100644
--- a/radarpipeline/io/writer.py
+++ b/radarpipeline/io/writer.py
@@ -102,7 +102,6 @@ def write_data(self) -> None:
                         index=False,
                         sep=constants.CSV_DELIMITER,
                         encoding=constants.ENCODING,
-                        line_terminator=constants.LINESEP,
                         compression=self.compression,
                     )
                 elif self.data_format == "pickle":
diff --git a/radarpipeline/project/__init__.py b/radarpipeline/project/__init__.py
index e6bb37c..9afa79d 100644
--- a/radarpipeline/project/__init__.py
+++ b/radarpipeline/project/__init__.py
@@ -1,1 +1,2 @@
 from radarpipeline.project.project import Project
+from radarpipeline.project.sparkengine import SparkEngine
diff --git a/radarpipeline/project/project.py b/radarpipeline/project/project.py
index d9dd666..86bfc3b 100644
--- a/radarpipeline/project/project.py
+++ b/radarpipeline/project/project.py
@@ -11,9 +11,10 @@
 
 from radarpipeline.common import utils
 from radarpipeline.features import Feature, FeatureGroup
-from radarpipeline.io import PandasDataWriter, SparkCSVDataReader, SparkDataWriter
+from radarpipeline.io import PandasDataWriter, SparkDataWriter, Reader
 from radarpipeline.io import SftpDataReader
 from radarpipeline.project.validations import ConfigValidator
+from radarpipeline.project.sparkengine import SparkEngine
 from strictyaml import load, YAMLError
 
 logger = logging.getLogger(__name__)
@@ -43,6 +44,13 @@ def __init__(self, input_data: Union[str, dict]) -> None:
         self.validator.validate()
         self.feature_groups = self._get_feature_groups()
         self.total_required_data = self._get_total_required_data()
+        if "spark_config" not in self.config:
+            self.config["spark_config"] = {}
+        self.spark_engine = SparkEngine(self.config['spark_config'])
+        self.spark_session = self.spark_engine.initialize_spark_session()
+
+    def close_spark_session(self):
+        self.spark_engine.close_spark_session()
 
     def _resolve_input_data(self, input_data) -> str:
         """
@@ -218,42 +226,38 @@ 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_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"],
-                    self.config['spark_config']
-                )
-                self.data = sparkcsvdatareader.read_data()
-                sparkcsvdatareader.close_spark_session()
-            else:
-                raise ValueError("Wrong data format")
 
-        elif self.config["input"]["data_type"] == "mock":
+        if self.config["input"]["source_type"] == "local":
+            datareader = Reader(
+                self.spark_session,
+                self.config,
+                self.total_required_data,
+                self.config["configurations"]["df_type"],
+            )
+            self.data = datareader.read_data()
+
+        elif self.config["input"]["source_type"] == "mock":
             MOCK_URL = "https://github.com/RADAR-base-Analytics/mockdata"
             cache_dir = os.path.join(
                 os.path.expanduser("~"), ".cache", "radarpipeline", "mockdata")
             if not os.path.exists(cache_dir):
                 Repo.clone_from(MOCK_URL, cache_dir)
             mock_data_directory = os.path.join(cache_dir, "mockdata")
-            mock_config_input = {
-                "config": {
-                    "source_path": mock_data_directory
-                }
-            }
-            sparkcsvdatareader = SparkCSVDataReader(
-                mock_config_input, self.total_required_data,
-                spark_config=self.config['spark_config']
+            mock_config = {
+                "input": {
+                    "config": {
+                        "source_path": mock_data_directory},
+                    "data_format": "csv"}}
+            mock_config["configurations"] = self.config["configurations"]
+            datareader = Reader(
+                self.spark_session,
+                mock_config,
+                self.total_required_data,
+                self.config["configurations"]["df_type"],
             )
-            self.data = sparkcsvdatareader.read_data()
-            sparkcsvdatareader.close_spark_session()
+            self.data = datareader.read_data()
 
-        elif self.config["input"]["data_type"] == "sftp":
+        elif self.config["input"]["source_type"] == "sftp":
             sftp_data_reader = SftpDataReader(self.config["input"]["config"],
                                               self.total_required_data)
             root_dir = sftp_data_reader.get_root_dir()
@@ -264,14 +268,13 @@ def fetch_data(self) -> None:
                     "source_path": root_dir
                 }
             }
-            sparkcsvdatareader = SparkCSVDataReader(
+            datareader = Reader(
+                self.spark_session,
                 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()
+            )
+            self.data = datareader.read_data()
         else:
             raise ValueError("Wrong data location")
 
@@ -299,7 +302,6 @@ def export_data(self) -> None:
         """
         df_type = self.config["configurations"]["df_type"]
         output_config = self.config["output"]
-        print(output_config)
         if output_config['output_location'] == "local":
             if df_type == "pandas":
                 writer = PandasDataWriter(
diff --git a/radarpipeline/project/sparkengine.py b/radarpipeline/project/sparkengine.py
new file mode 100644
index 0000000..616a5fc
--- /dev/null
+++ b/radarpipeline/project/sparkengine.py
@@ -0,0 +1,106 @@
+import pyspark.sql as ps
+from pyspark.sql import SparkSession, DataFrame
+from pyspark.sql.types import StructField, StructType
+from pyspark.sql.utils import IllegalArgumentException
+from typing import Any, Dict, List, Optional, Union
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class SparkEngine():
+    """
+    Read CSV data from local directory using pySpark
+    """
+
+    def __init__(self, spark_config: Dict = None):
+        default_spark_config = {'spark_master': 'local',
+                                'spark.executor.instances': 2,
+                                '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',
+                                'spark.log.level': "OFF"}
+        self.spark_config = default_spark_config
+        if spark_config is not None:
+            self.spark_config.update(spark_config)
+
+    def initialize_spark_session(self) -> ps.SparkSession:
+        """
+        Initializes and returns a SparkSession
+
+        Returns
+        -------
+        SparkSession
+            A SparkSession object
+        """
+
+        """
+        Spark configuration documentation:
+        https://spark.apache.org/docs/latest/configuration.html
+
+        `spark.executor.instances` is the number of executors to
+        launch for an application.
+
+        `spark.executor.cores` is the number of cores to =
+        use on each executor.
+
+        `spark.executor.memory` is the amount of memory to
+        use per executor process.
+
+        `spark.driver.memory` is the amount of memory to use for the driver process,
+        i.e. where SparkContext is initialized, in MiB unless otherwise specified.
+
+        `spark.memory.offHeap.enabled` is to enable off-heap memory allocation
+
+        `spark.memory.offHeap.size` is the absolute amount of memory which can be used
+        for off-heap allocation, in bytes unless otherwise specified.
+
+        `spark.driver.maxResultSize` is the limit of total size of serialized results of
+        all partitions for each Spark action (e.g. collect) in bytes.
+        Should be at least 1M, or 0 for unlimited.
+        """
+        if self.spark_config['spark_master'] == "local":
+            self.spark = (
+                SparkSession.builder.master("local[*]").appName("radarpipeline")
+                .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'])
+                .config('spark.log.level',
+                        self.spark_config['spark.log.level'])
+                .getOrCreate()
+            )
+        else:
+            self.spark = (
+                SparkSession.builder.master(
+                    self.spark_config['spark_master']).appName("radarpipeline")
+                .getOrCreate()
+            )
+        self.spark._jsc.setLogLevel(self.spark_config['spark.log.level'])
+        self.spark.sparkContext.setLogLevel("OFF")
+        # Enable Apache Arrow for optimizations in Spark to Pandas conversion
+        self.spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
+        # Fallback to use non-Arrow conversion in case of errors
+        self.spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled",
+                            "true")
+        self.spark.conf.set("spark.sql.session.timeZone", "UTC")
+        # For further reading:
+        # https://spark.apache.org/docs/3.0.1/sql-pyspark-pandas-with-arrow.html
+        logger.info("Spark Session created")
+        return self.spark
+
+    def close_spark_session(self):
+        self.spark.stop()
diff --git a/radarpipeline/project/validations.py b/radarpipeline/project/validations.py
index 1fc25de..e740d9a 100644
--- a/radarpipeline/project/validations.py
+++ b/radarpipeline/project/validations.py
@@ -59,7 +59,7 @@ def _validate_input(self) -> None:
         Validates the input data config
         """
 
-        if self.config["input"]["data_type"] == "sftp":
+        if self.config["input"]["source_type"] == "sftp":
             sftp_config_keys = [
                 "sftp_host",
                 "sftp_username",
@@ -70,7 +70,7 @@ def _validate_input(self) -> None:
                 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":
+        elif self.config["input"]["source_type"] == "local":
             if "source_path" not in self.config["input"]["config"]:
                 raise ValueError("Key not present in the config: source_path")
             else:
@@ -93,7 +93,7 @@ def _validate_input(self) -> None:
             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":
+        elif self.config["input"]["source_type"] == "mock":
             self._update_mock_data()
 
         else:
@@ -108,11 +108,71 @@ def _validate_configurations(self) -> None:
             self.config["configurations"] = {}
 
         valid_df_types = ["pandas", "spark"]
+        valid_user_sampling_methods = ["fraction", "count", "userid"]
 
         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")
+        if "user_sampling" in self.config["configurations"]:
+            if "method" not in self.config["configurations"]["user_sampling"]:
+                raise ValueError("Key not present in the user_sampling config: method")
+            elif (
+                self.config["configurations"]["user_sampling"]["method"]
+                not in valid_user_sampling_methods
+            ):
+                raise ValueError("Invalid value for the key: method")
+            else:
+                self.config["configurations"][
+                    "user_sampling"] = self._validate_user_sampling_config(
+                        self.config["configurations"]["user_sampling"])
+        else:
+            self.config["configurations"]["user_sampling"] = None
+
+        # Validating data sampling
+        valid_data_sampling_methods = ["fraction", "count", "time"]
+        if "data_sampling" in self.config["configurations"]:
+            if "method" not in self.config["configurations"]["data_sampling"]:
+                raise ValueError("Key not present in the data_sampling config: method")
+            elif (
+                self.config["configurations"]["data_sampling"]["method"]
+                not in valid_data_sampling_methods
+            ):
+                raise ValueError("Invalid value for the key: method")
+            else:
+                self.config["configurations"][
+                    "data_sampling"] = self._validate_data_sampling_config(
+                        self.config["configurations"]["data_sampling"])
+        else:
+            self.config["configurations"]["data_sampling"] = None
+
+    def _validate_user_sampling_config(self, user_sampling_config):
+        """
+        Validates the user_sampling config
+        """
+
+        if user_sampling_config["method"] == "fraction":
+            user_sampling_config = self._validate_sampling_fraction(
+                user_sampling_config)
+        elif user_sampling_config["method"] == "count":
+            user_sampling_config = self._validate_sampling_count(
+                user_sampling_config)
+        elif user_sampling_config["method"] == "userid":
+            user_sampling_config = self._validate_sampling_userids(
+                user_sampling_config)
+        return user_sampling_config
+
+    def _validate_data_sampling_config(self, data_sampling_config):
+        if data_sampling_config["method"] == "fraction":
+            data_sampling_config = self._validate_sampling_fraction(
+                data_sampling_config)
+        elif data_sampling_config["method"] == "count":
+            data_sampling_config = self._validate_sampling_count(
+                data_sampling_config)
+        elif data_sampling_config["method"] == "time":
+            data_sampling_config = self._validate_sampling_time(
+                data_sampling_config)
+        return data_sampling_config
 
     def _validate_features(self) -> None:
         """
@@ -226,3 +286,95 @@ def _validate_output(self) -> None:
                 self.config["output"]["compress"] = True
         else:
             raise ValueError("Key not present in the config: output_location")
+
+    def _validate_sampling_fraction(self, sampling_config):
+        if "config" not in sampling_config:
+            raise ValueError("Key not present in the user_sampling config: config")
+        if "fraction" not in sampling_config["config"]:
+            raise ValueError(
+                "Key not present in the user_sampling config: fraction"
+            )
+        # converting fraction to float
+        try:
+            sampling_config["config"]["fraction"] = float(
+                sampling_config["config"]["fraction"]
+            )
+        except ValueError:
+            raise ValueError(
+                "Invalid value for the key: fraction. It should be a number"
+            )
+        if not 0 < sampling_config["config"]["fraction"] <= 1:
+            raise ValueError(
+                "Invalid value for the key: fraction. It should be between 0 and 1"
+            )
+        return sampling_config
+
+    def _validate_sampling_count(self, sampling_config):
+        if "config" not in sampling_config:
+            raise ValueError("Key not present in the user_sampling config: config")
+        if "count" not in sampling_config["config"]:
+            raise ValueError("Key not present in the user_sampling config: count")
+        try:
+            sampling_config["config"]["count"] = int(
+                sampling_config["config"]["count"]
+            )
+        except ValueError:
+            raise ValueError(
+                "Invalid value for the key: count. It should be a number"
+            )
+        if sampling_config["config"]["count"] <= 0:
+            raise ValueError(
+                "Invalid value for the key: count. It should be greater than 0"
+            )
+        return sampling_config
+
+    def _validate_sampling_userids(self, sampling_config):
+        if "config" not in sampling_config:
+            raise ValueError("Key not present in the user_sampling config: config")
+        if "userids" not in sampling_config["config"]:
+            raise ValueError("Key not present in the user_sampling config: userids")
+        if len(sampling_config["config"]["userids"]) == 0:
+            raise ValueError(
+                "user_ids array cannot be empty in the user_sampling config"
+            )
+        # check if userids are in array. If not convert to array
+        if not isinstance(sampling_config["config"]["userids"], list):
+            sampling_config["config"]["userids"] = [
+                sampling_config["config"]["userids"]
+            ]
+        return sampling_config
+
+    def _validate_sampling_time_instance(self, time_dict: dict):
+        if ("starttime" not in time_dict
+            ) and (
+                "endtime" not in time_dict):
+            raise ValueError("Neither startime nor endtime present in the config")
+        # check if starttime and endtime are can be converted into time format
+        # if so, convert them
+        if "starttime" in time_dict:
+            time_dict['starttime'] = utils.convert_str_to_time(
+                time_dict['starttime'])
+        if "endtime" in time_dict:
+            time_dict['endtime'] = utils.convert_str_to_time(
+                time_dict['endtime'])
+        if "time_column" not in time_dict:
+            logger.warning("time_column not present in the config. \
+                Using default time column: value.time")
+            time_dict["time_column"] = "value.time"
+        return time_dict
+
+    def _validate_sampling_time(self, sampling_config):
+        if "config" not in sampling_config:
+            raise ValueError("Key not present in the user_sampling config: config")
+        if type(sampling_config["config"]) is list:
+            if len(sampling_config["config"]) == 0:
+                raise ValueError(
+                    "No starttime and endtime present in the config"
+                )
+            for i, time_dict in enumerate(sampling_config["config"]):
+                sampling_config["config"][i] = self._validate_sampling_time_instance(
+                    time_dict)
+        else:
+            sampling_config["config"] = self._validate_sampling_time_instance(
+                sampling_config["config"])
+        return sampling_config
diff --git a/radarpipeline/radarpipeline.py b/radarpipeline/radarpipeline.py
index 8f1d580..70c810e 100644
--- a/radarpipeline/radarpipeline.py
+++ b/radarpipeline/radarpipeline.py
@@ -1,7 +1,7 @@
 import logging
 import sys
 import traceback
-
+from typing import Dict, Union
 from radarpipeline import Project
 from radarpipeline.common.logger import logger_init
 
@@ -10,7 +10,7 @@
 logger = logging.getLogger(__name__)
 
 
-def run(config_path: str = "config.yaml"):
+def run(config_path: Union[str, Dict] = "config.yaml"):
     """
     Pipeline entry point.
     config_path could be a local path to a configuration file
@@ -26,6 +26,8 @@ def run(config_path: str = "config.yaml"):
         project.compute_features()
         logger.info("Exporting the features data...")
         project.export_data()
+        logger.info("Data exported successfully. Closing Spark Engine")
+        project.close_spark_session()
         logger.info("Pipeline run completed successfully")
     except KeyboardInterrupt:
         logger.info("Pipeline run interrupted by user")
diff --git a/requirements.txt b/requirements.txt
index 2d5b859..0a13102 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,15 +1,15 @@
 twine==3.8.0
 pyYaml==6.0
-numpy==1.22.3
-pandas==1.4.1
+numpy==1.24.4
+pandas==2.0.3
 matplotlib==3.5.1
 seaborn==0.11.2
 scipy==1.10.0
-pyspark[sql]==3.3.0
-GitPython>=3.1.32
+pyspark[sql]==3.5.0
+GitPython>=3.1.41
 strictyaml==1.7.3
 pyspark-test==0.2.0
-paramiko==3.1.0
+paramiko==3.4.0
 pre-commit
 pytest
 pytest-cov
@@ -17,3 +17,4 @@ pytest-mock
 pytest-sftpserver
 dhp
 avro==1.11.2
+pyarrow==16.0.0
diff --git a/setup.py b/setup.py
index b1de97b..aa72458 100644
--- a/setup.py
+++ b/setup.py
@@ -24,13 +24,13 @@ def read_file(filename):
     packages=find_packages(),
     install_requires=[
         "pyYaml==6.0",
-        "pandas==1.4.1",
-        "numpy==1.22.3",
+        "pandas==2.0.3",
+        "numpy==1.24.4",
         "scipy==1.10.0",
-        "pyspark[sql]==3.3.0",
-        "GitPython>=3.1.32",
+        "pyspark[sql]==3.5.0",
+        "GitPython>=3.1.41",
         "strictyaml==1.7.3",
-        "paramiko==3.1.0",
+        "paramiko==3.4.0",
         "avro==1.11.2"],
     test_suite="tests",
     include_package_data=True,


Test patch:
diff --git a/tests/resources/expected_output/phone_battery_charging_duration.csv b/tests/resources/expected_output/phone_battery_charging_duration.csv
index a13b994..9d59082 100644
--- a/tests/resources/expected_output/phone_battery_charging_duration.csv
+++ b/tests/resources/expected_output/phone_battery_charging_duration.csv
@@ -1,37 +1,37 @@
 key.userId,date,value.status,value.statusTime,value.statusTimeInSeconds
-072ddb22-82ef-4b81-8460-41ab096b54bb,2019-10-26,CHARGING,0 days 02:21:00.209999800,141.00349999666668
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-01,CHARGING,2 days 02:45:02.355999947,3045.0392666657835
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-03,CHARGING,0 days 20:42:59.004000187,1242.9834000031167
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-04,CHARGING,3 days 00:26:06.661000014,4346.1110166669005
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-07,CHARGING,0 days 01:11:28.029999971,71.46716666618335
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-08,CHARGING,0 days 04:17:37.694000007,257.62823333345
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-09,CHARGING,0 days 06:29:13.859999896,389.23099999826667
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-10,CHARGING,0 days 01:10:49.706000089,70.82843333481668
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-23,CHARGING,0 days 00:02:31.164000034,2.519400000566667
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-07,CHARGING,0 days 01:05:19.040999889,65.31734999815001
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-08,CHARGING,0 days 01:36:19.865000010,96.3310833335
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-12,CHARGING,0 days 02:03:21.999000072,123.36665000120001
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-13,CHARGING,0 days 00:40:04.375999927,40.07293333211667
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-14,CHARGING,0 days 01:34:26.461999893,94.44103333155
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-19,CHARGING,0 days 00:14:06.765000104,14.112750001733334
-07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-29,CHARGING,0 days 03:15:18.118000031,195.30196666718336
-07a69f47-1923-4cfc-b89b-0eefad483f43,2020-01-13,CHARGING,0 days 00:05:05.431999920,5.090533332000001
-07a69f47-1923-4cfc-b89b-0eefad483f43,2020-01-27,CHARGING,0 days 01:10:28.252999784,70.47088332973334
-07a69f47-1923-4cfc-b89b-0eefad483f43,2020-01-30,CHARGING,0 days 00:02:42.644000053,2.710733334216667
-07a69f47-1923-4cfc-b89b-0eefad483f43,2020-02-04,CHARGING,0 days 01:42:59.499999999,102.99166666665
-07a69f47-1923-4cfc-b89b-0eefad483f43,2020-03-04,CHARGING,0 days 02:12:53.666999817,132.89444999695002
-07a69f47-1923-4cfc-b89b-0eefad483f43,2020-03-05,CHARGING,146 days 21:23:05.388999938,211523.08981666566
-07a69f47-1923-4cfc-b89b-0eefad483f43,2020-10-20,CHARGING,0 days 00:42:19.325000048,42.32208333413334
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-23,CHARGING,0 days 02:12:54.634000063,132.91056666771667
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-24,CHARGING,0 days 01:11:10.625000237,71.17708333728335
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-25,CHARGING,0 days 02:10:00.172000170,130.0028666695
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-26,CHARGING,0 days 03:38:00.754999876,218.01258333126668
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-27,CHARGING,0 days 03:11:40.501999856,191.67503333093333
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-28,CHARGING,0 days 01:45:50.284999848,105.8380833308
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-29,CHARGING,0 days 02:02:29.530000209,122.49216667015001
-2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-30,CHARGING,0 days 01:16:54.997999668,76.9166333278
-5c0e2ec7-6f85-4041-9669-7145075d1754,2018-10-31,CHARGING,0 days 02:16:37.223000050,136.62038333416666
-5c0e2ec7-6f85-4041-9669-7145075d1754,2018-11-08,CHARGING,0 days 20:06:12.864000082,1206.2144000013666
-5c0e2ec7-6f85-4041-9669-7145075d1754,2018-11-09,CHARGING,0 days 00:09:59.984000206,9.999733336766667
-5c0e2ec7-6f85-4041-9669-7145075d1754,2018-12-03,CHARGING,41 days 18:10:47.425999880,60130.79043333134
+072ddb22-82ef-4b81-8460-41ab096b54bb,2019-10-26,CHARGING,0 days 02:21:00.210000,141.00349999999997
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-01,CHARGING,2 days 02:45:02.356000,3045.0392666666667
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-03,CHARGING,0 days 20:42:59.004000,1242.9834
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-04,CHARGING,3 days 00:26:06.661000,4346.111016666667
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-07,CHARGING,0 days 01:11:28.030000,71.46716666666666
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-08,CHARGING,0 days 04:17:37.694000,257.6282333333333
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-09,CHARGING,0 days 06:29:13.860000,389.231
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-10,CHARGING,0 days 01:10:49.706000,70.82843333333334
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-10-23,CHARGING,0 days 00:02:31.164000,2.5193999999999996
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-07,CHARGING,0 days 01:05:19.041000,65.31735
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-08,CHARGING,0 days 01:36:19.865000,96.33108333333332
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-12,CHARGING,0 days 02:03:21.999000,123.36664999999999
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-13,CHARGING,0 days 00:40:04.376000,40.07293333333334
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-14,CHARGING,0 days 01:34:26.462000,94.44103333333334
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-19,CHARGING,0 days 00:14:06.765000,14.11275
+07a69f47-1923-4cfc-b89b-0eefad483f43,2019-11-29,CHARGING,0 days 03:15:18.118000,195.3019666666667
+07a69f47-1923-4cfc-b89b-0eefad483f43,2020-01-13,CHARGING,0 days 00:05:05.432000,5.090533333333334
+07a69f47-1923-4cfc-b89b-0eefad483f43,2020-01-27,CHARGING,0 days 01:10:28.253000,70.47088333333333
+07a69f47-1923-4cfc-b89b-0eefad483f43,2020-01-30,CHARGING,0 days 00:02:42.644000,2.710733333333333
+07a69f47-1923-4cfc-b89b-0eefad483f43,2020-02-04,CHARGING,0 days 01:42:59.500000,102.99166666666666
+07a69f47-1923-4cfc-b89b-0eefad483f43,2020-03-04,CHARGING,0 days 02:12:53.667000,132.89445
+07a69f47-1923-4cfc-b89b-0eefad483f43,2020-03-05,CHARGING,146 days 21:23:05.389000,211523.08981666667
+07a69f47-1923-4cfc-b89b-0eefad483f43,2020-10-20,CHARGING,0 days 00:42:19.325000,42.32208333333333
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-23,CHARGING,0 days 02:12:54.634000,132.91056666666665
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-24,CHARGING,0 days 01:11:10.625000,71.17708333333333
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-25,CHARGING,0 days 02:10:00.172000,130.00286666666665
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-26,CHARGING,0 days 03:38:00.755000,218.0125833333333
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-27,CHARGING,0 days 03:11:40.502000,191.67503333333335
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-28,CHARGING,0 days 01:45:50.285000,105.83808333333333
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-29,CHARGING,0 days 02:02:29.530000,122.49216666666666
+2a02e53a-951e-4fd0-b47f-195a87096bd0,2018-11-30,CHARGING,0 days 01:16:54.998000,76.91663333333332
+5c0e2ec7-6f85-4041-9669-7145075d1754,2018-10-31,CHARGING,0 days 02:16:37.223000,136.62038333333334
+5c0e2ec7-6f85-4041-9669-7145075d1754,2018-11-08,CHARGING,0 days 20:06:12.864000,1206.2144
+5c0e2ec7-6f85-4041-9669-7145075d1754,2018-11-09,CHARGING,0 days 00:09:59.984000,9.999733333333333
+5c0e2ec7-6f85-4041-9669-7145075d1754,2018-12-03,CHARGING,41 days 18:10:47.426000,60130.790433333335
 5c0e2ec7-6f85-4041-9669-7145075d1754,2019-01-14,CHARGING,0 days 00:00:00,0.0
diff --git a/tests/resources/test_yamls/config_with_incorrect_spark.yaml b/tests/resources/test_yamls/config_with_incorrect_spark.yaml
index 8c784de..e44ca4b 100644
--- a/tests/resources/test_yamls/config_with_incorrect_spark.yaml
+++ b/tests/resources/test_yamls/config_with_incorrect_spark.yaml
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: mock # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         # In case of sftp, use the following format
         # sftp_host:
diff --git a/tests/resources/test_yamls/config_with_spark.yaml b/tests/resources/test_yamls/config_with_spark.yaml
index 728f68f..4965022 100644
--- a/tests/resources/test_yamls/config_with_spark.yaml
+++ b/tests/resources/test_yamls/config_with_spark.yaml
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: mock # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         # In case of sftp, use the following format
         # sftp_host:
diff --git a/tests/resources/test_yamls/test_config.yaml b/tests/resources/test_yamls/test_config.yaml
index 6368c18..f2e8885 100644
--- a/tests/resources/test_yamls/test_config.yaml
+++ b/tests/resources/test_yamls/test_config.yaml
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: mock # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         # In case of sftp, use the following format
         # sftp_host:
diff --git a/tests/resources/test_yamls/test_config_all_sampling.yaml b/tests/resources/test_yamls/test_config_all_sampling.yaml
new file mode 100644
index 0000000..d48363c
--- /dev/null
+++ b/tests/resources/test_yamls/test_config_all_sampling.yaml
@@ -0,0 +1,69 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input:
+    source_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'
+    user_sampling:
+        ## Possible methods: fraction, count, userid
+    #    method: fraction
+    #    config:
+    #        fraction: 0.8
+        method: count
+        config:
+          count: 2
+        #method: userid
+        #config:
+        #    userids:
+        #        - 2a02e53a-951e-4fd0-b47f-195a87096bd0
+    ## TODO: For future
+    data_sampling:
+        ## Possible methods:  time, count, fraction
+        ## starttime and endtime format is dd-mm-yyyy hh:mm:ss in UTC timezone
+        #method: time
+        #config:
+        #   starttime: 2018-11-25 00:00:00
+        #   endtime:  2018-11-29 00:00:00
+        #   time_column: value.time
+        method: count
+        config:
+           count: 100
+        #method: fraction
+        #config:
+        #   fraction: 0.3
+
+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_date_sampling.yaml b/tests/resources/test_yamls/test_config_date_sampling.yaml
new file mode 100644
index 0000000..7b2ae83
--- /dev/null
+++ b/tests/resources/test_yamls/test_config_date_sampling.yaml
@@ -0,0 +1,69 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input:
+    source_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'
+    #user_sampling:
+        ## Possible methods: fraction, count, userid
+    #    method: fraction
+    #    config:
+    #        fraction: 0.8
+        #method: count
+        #config:
+        #    count: 2
+        #method: userid
+        #config:
+        #    userids:
+        #        - 2a02e53a-951e-4fd0-b47f-195a87096bd0
+    ## TODO: For future
+    data_sampling:
+        ## Possible methods:  time, count, fraction
+        ## starttime and endtime format is dd-mm-yyyy hh:mm:ss in UTC timezone
+        method: time
+        config:
+           starttime: 2018-11-25 00:00:00
+           endtime:  2018-11-29 00:00:00
+           time_column: value.time
+        #method: count
+        #config:
+        #   count: 100
+        #method: fraction
+        #config:
+        #   fraction: 0.3
+
+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 22fa2b4..2e5e7b7 100644
--- a/tests/resources/test_yamls/test_config_incomplete.yaml
+++ b/tests/resources/test_yamls/test_config_incomplete.yaml
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: mock # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         # In case of sftp, use the following format
         # sftp_host:
diff --git a/tests/resources/test_yamls/test_config_input_invalid.yaml b/tests/resources/test_yamls/test_config_input_invalid.yaml
index ad2e3da..689e7d7 100644
--- a/tests/resources/test_yamls/test_config_input_invalid.yaml
+++ b/tests/resources/test_yamls/test_config_input_invalid.yaml
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: mock # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         # In case of sftp, use the following format
         # sftp_host:
diff --git a/tests/resources/test_yamls/test_config_project.yaml b/tests/resources/test_yamls/test_config_project.yaml
index e520122..929a61b 100644
--- a/tests/resources/test_yamls/test_config_project.yaml
+++ b/tests/resources/test_yamls/test_config_project.yaml
@@ -4,7 +4,7 @@ project:
     version: mock_version
 
 input:
-    data_type: mock # couldbe mock, local, sftp, s3
+    source_type: mock # couldbe mock, local, sftp, s3
     config:
         # In case of sftp, use the following format
         # sftp_host:
diff --git a/tests/resources/test_yamls/test_config_user_sampling.yaml b/tests/resources/test_yamls/test_config_user_sampling.yaml
new file mode 100644
index 0000000..2689954
--- /dev/null
+++ b/tests/resources/test_yamls/test_config_user_sampling.yaml
@@ -0,0 +1,69 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input:
+    source_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'
+    user_sampling:
+        ## Possible methods: fraction, count, userid
+    #    method: fraction
+    #    config:
+    #        fraction: 0.8
+        #method: count
+        #config:
+        #    count: 2
+        method: userid
+        config:
+            userids:
+                - 2a02e53a-951e-4fd0-b47f-195a87096bd0
+    ## TODO: For future
+    #data_sampling:
+        ## Possible methods:  time, count, fraction
+        ## starttime and endtime format is dd-mm-yyyy hh:mm:ss in UTC timezone
+        #method: time
+        #config:
+        #   starttime: 2018-11-25 00:00:00
+        #   endtime:  2018-11-29 00:00:00
+        #   time_column: value.time
+        #method: count
+        #config:
+        #   count: 100
+        #method: fraction
+        #config:
+        #   fraction: 0.3
+
+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/test_integration/test_sampling.py b/tests/test_integration/test_sampling.py
new file mode 100644
index 0000000..c0286f5
--- /dev/null
+++ b/tests/test_integration/test_sampling.py
@@ -0,0 +1,256 @@
+import unittest
+from radarpipeline import radarpipeline, Project
+from strictyaml.exceptions import YAMLValidationError
+import pathlib as pl
+import os
+import pandas as pd
+from pandas.testing import assert_frame_equal
+
+
+class TestSampling(unittest.TestCase):
+
+    def setUp(self):
+        self.default_config = {
+            'project': {
+                'project_name': 'mock_project',
+                'description': 'mock_description',
+                'version': 'mock_version'},
+            'input': {
+                'source_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}}
+
+    def get_config_output(self, config):
+        project = Project(input_data=config)
+        project.fetch_data()
+        project.compute_features()
+        output_data = project.features
+        project.close_spark_session()
+        return output_data
+
+    def test_user_sampling_userid(self):
+        user_sampling_config = self.default_config
+        user_sampling_config['configurations']['user_sampling'] = {}
+        user_sampling_config['configurations']['user_sampling']['method'] = 'userid'
+        user_sampling_config['configurations']['user_sampling']['config'] = {
+            'userids': ["2a02e53a-951e-4fd0-b47f-195a87096bd0"]}
+        output_data = self.get_config_output(user_sampling_config)
+        self.assertEqual(output_data['PhoneBatteryChargingDuration'][
+            'key.userId'].unique(), ['2a02e53a-951e-4fd0-b47f-195a87096bd0'])
+        self.assertEqual(output_data['StepCountPerDay']['key.userId'].unique(),
+                         ['2a02e53a-951e-4fd0-b47f-195a87096bd0'])
+
+    def test_user_sampling_count(self):
+        user_sampling_config = self.default_config
+        user_sampling_config['configurations']['user_sampling'] = {}
+        user_sampling_config['configurations']['user_sampling']['method'] = 'count'
+        user_sampling_config['configurations']['user_sampling']['config'] = {'count': 2}
+        output_data = self.get_config_output(user_sampling_config)
+        self.assertEqual(output_data['PhoneBatteryChargingDuration'][
+            'key.userId'].unique().shape[0], 2)
+        self.assertEqual(output_data['StepCountPerDay'][
+            'key.userId'].unique().shape[0], 2)
+
+    def test_user_sampling_fraction(self):
+        user_sampling_config = self.default_config
+        user_sampling_config['configurations']['user_sampling'] = {}
+        user_sampling_config['configurations']['user_sampling']['method'] = 'fraction'
+        user_sampling_config['configurations']['user_sampling'][
+            'config'] = {'fraction': 0.75}
+        output_data = self.get_config_output(user_sampling_config)
+        self.assertEqual(output_data['PhoneBatteryChargingDuration'][
+            'key.userId'].unique().shape[0], 3)
+        self.assertEqual(output_data['StepCountPerDay'][
+            'key.userId'].unique().shape[0], 3)
+
+    def test_data_sampling_time(self):
+        starttime = "2018-11-25 00:00:00"
+        endtime = "2018-11-29 00:00:00"
+        data_sampling_config = self.default_config
+        data_sampling_config['configurations']['data_sampling'] = {}
+        data_sampling_config['configurations']['data_sampling']['method'] = 'time'
+        data_sampling_config['configurations'][
+            'data_sampling']['config'] = {
+                'starttime': starttime,
+                'endtime': endtime,
+                'time_column': 'value.time'}
+        output_data = self.get_config_output(data_sampling_config)
+        self.assertGreaterEqual(output_data['PhoneBatteryChargingDuration'][
+            'date'].min(), pd.Timestamp(starttime).date())
+        self.assertLessEqual(output_data['PhoneBatteryChargingDuration'][
+            'date'].max(), pd.Timestamp(endtime).date())
+        self.assertGreaterEqual(output_data['StepCountPerDay'][
+            'date'].min(), pd.Timestamp(starttime).date())
+        self.assertLessEqual(output_data['StepCountPerDay'][
+            'date'].max(), pd.Timestamp(endtime).date())
+
+    def test_data_sampling_time_list(self):
+        starttime = "2018-11-25 00:00:00"
+        endtime = "2018-11-29 00:00:00"
+        time_list = [{"starttime": starttime, "endtime": endtime,
+                      "time_column": 'value.time'}]
+        data_sampling_config = self.default_config
+        data_sampling_config['configurations']['data_sampling'] = {}
+        data_sampling_config['configurations']['data_sampling']['method'] = 'time'
+        data_sampling_config['configurations'][
+            'data_sampling']['config'] = time_list
+        output_data = self.get_config_output(data_sampling_config)
+        self.assertGreaterEqual(output_data['PhoneBatteryChargingDuration'][
+            'date'].min(), pd.Timestamp(starttime).date())
+        self.assertLessEqual(output_data['PhoneBatteryChargingDuration'][
+            'date'].max(), pd.Timestamp(endtime).date())
+        self.assertGreaterEqual(output_data['StepCountPerDay'][
+            'date'].min(), pd.Timestamp(starttime).date())
+        self.assertLessEqual(output_data['StepCountPerDay'][
+            'date'].max(), pd.Timestamp(endtime).date())
+
+    def test_data_sampling_multiple_time(self):
+        starttime_1 = "2018-11-25 00:00:00"
+        endtime_1 = "2018-11-29 00:00:00"
+        starttime_2 = "2019-01-01 00:00:00"
+        endtime_2 = "2019-04-30 00:00:00"
+        time_list = [{"starttime": starttime_1, "endtime": endtime_1,
+                      "time_column": 'value.time'},
+                     {"starttime": starttime_2, "endtime": endtime_2}]
+        data_sampling_config = self.default_config
+        data_sampling_config['configurations']['data_sampling'] = {}
+        data_sampling_config['configurations']['data_sampling']['method'] = 'time'
+        data_sampling_config['configurations'][
+            'data_sampling']['config'] = time_list
+        output_data = self.get_config_output(data_sampling_config)
+        self.assertGreaterEqual(output_data['PhoneBatteryChargingDuration'][
+            'date'].min(), pd.Timestamp(starttime_1).date())
+        self.assertLessEqual(output_data['PhoneBatteryChargingDuration'][
+            'date'].max(), pd.Timestamp(endtime_2).date())
+        self.assertGreaterEqual(output_data['StepCountPerDay'][
+            'date'].min(), pd.Timestamp(starttime_1).date())
+        self.assertLessEqual(output_data['StepCountPerDay'][
+            'date'].max(), pd.Timestamp(endtime_2).date())
+        # check output_data['PhoneBatteryChargingDuration']['date'] is
+        # between the time range
+        # starttime_1 and endtime_1 and starttime_2 and endtime_2
+        self.assertTrue(all((output_data['PhoneBatteryChargingDuration']['date']
+                             >= pd.Timestamp(starttime_1).date())
+                            & (output_data['PhoneBatteryChargingDuration']['date']
+                               <= pd.Timestamp(endtime_1).date())
+                            | (output_data['PhoneBatteryChargingDuration']['date']
+                               >= pd.Timestamp(starttime_2).date())
+                            & (output_data['PhoneBatteryChargingDuration']['date']
+                               <= pd.Timestamp(endtime_2).date())))
+
+        self.assertTrue(all((output_data['StepCountPerDay']['date']
+                             >= pd.Timestamp(starttime_1).date())
+                            & (output_data['StepCountPerDay']['date']
+                               <= pd.Timestamp(endtime_1).date())
+                            | (output_data['StepCountPerDay']['date']
+                               >= pd.Timestamp(starttime_2).date())
+                            & (output_data['StepCountPerDay']['date']
+                               <= pd.Timestamp(endtime_2).date())))
+
+    def test_data_sampling_multiple_time_single_starttime(self):
+        starttime_1 = "2018-11-25 00:00:00"
+        endtime_1 = "2018-11-29 00:00:00"
+        starttime_2 = "2019-01-01 00:00:00"
+        time_list = [{"starttime": starttime_1, "endtime": endtime_1,
+                      "time_column": 'value.time'},
+                     {"starttime": starttime_2}]
+        data_sampling_config = self.default_config
+        data_sampling_config['configurations']['data_sampling'] = {}
+        data_sampling_config['configurations']['data_sampling']['method'] = 'time'
+        data_sampling_config['configurations'][
+            'data_sampling']['config'] = time_list
+        output_data = self.get_config_output(data_sampling_config)
+        # check output_data['PhoneBatteryChargingDuration']['date'] is between
+        # the time range
+        # starttime_1 and endtime_1 and starttime_2
+        self.assertTrue(all((output_data['PhoneBatteryChargingDuration']['date']
+                             >= pd.Timestamp(starttime_1).date())
+                            & (output_data['PhoneBatteryChargingDuration']['date']
+                               <= pd.Timestamp(endtime_1).date())
+                            | (output_data['PhoneBatteryChargingDuration']['date']
+                               >= pd.Timestamp(starttime_2).date())))
+
+        self.assertTrue(all((output_data['StepCountPerDay']['date']
+                             >= pd.Timestamp(starttime_1).date())
+                            & (output_data['StepCountPerDay']['date']
+                               <= pd.Timestamp(endtime_1).date())
+                            | (output_data['StepCountPerDay']['date']
+                               >= pd.Timestamp(starttime_2).date())))
+
+    def test_data_sampling_multiple_time_single_endtime(self):
+        endtime_1 = "2018-11-29 00:00:00"
+        starttime_2 = "2019-01-01 00:00:00"
+        endtime_2 = "2019-04-30 00:00:00"
+        time_list = [{"endtime": endtime_1,
+                      "time_column": 'value.time'},
+                     {"starttime": starttime_2, "endtime": endtime_2}]
+        data_sampling_config = self.default_config
+        data_sampling_config['configurations']['data_sampling'] = {}
+        data_sampling_config['configurations']['data_sampling']['method'] = 'time'
+        data_sampling_config['configurations'][
+            'data_sampling']['config'] = time_list
+        output_data = self.get_config_output(data_sampling_config)
+        # check output_data['PhoneBatteryChargingDuration']['date'] is between
+        # the time range
+        # endtime_1 and starttime_2 and endtime_2
+        self.assertTrue(all((output_data['PhoneBatteryChargingDuration']['date']
+                             <= pd.Timestamp(endtime_1).date())
+                            | (output_data['PhoneBatteryChargingDuration']['date']
+                               >= pd.Timestamp(starttime_2).date())
+                            & (output_data['PhoneBatteryChargingDuration']['date']
+                               <= pd.Timestamp(endtime_2).date())))
+
+        self.assertTrue(all((output_data['StepCountPerDay']['date']
+                             <= pd.Timestamp(endtime_1).date())
+                            | (output_data['StepCountPerDay']['date']
+                               >= pd.Timestamp(starttime_2).date())
+                            & (output_data['StepCountPerDay']['date']
+                               <= pd.Timestamp(endtime_2).date())))
+
+    def test_data_sampling_count(self):
+        data_sampling_config = self.default_config
+        data_sampling_config['configurations']['data_sampling'] = {}
+        data_sampling_config['configurations']['data_sampling']['method'] = 'count'
+        data_sampling_config['configurations']['data_sampling']['config'] = {
+            'count': 100
+        }
+        output_data = self.get_config_output(data_sampling_config)
+        # check if count of per key.userId is less than or equal to 100
+        self.assertTrue(all(output_data['PhoneBatteryChargingDuration'].groupby(
+            'key.userId').size() <= 100))
+        self.assertTrue(all(output_data['StepCountPerDay'].groupby(
+            'key.userId').size() <= 100))
+
+    def test_user_data_sampling(self):
+        user_data_sampling_config = self.default_config
+        user_data_sampling_config['configurations']['user_sampling'] = {}
+        user_data_sampling_config['configurations']['user_sampling'][
+            'method'] = 'userid'
+        user_data_sampling_config['configurations']['user_sampling'][
+            'config'] = {'userids': ["2a02e53a-951e-4fd0-b47f-195a87096bd0"]}
+        user_data_sampling_config['configurations']['data_sampling'] = {}
+        user_data_sampling_config['configurations']['data_sampling'][
+            'method'] = 'count'
+        user_data_sampling_config['configurations']['data_sampling'][
+            'config'] = {'count': 100}
+        output_data = self.get_config_output(user_data_sampling_config)
+        self.assertEqual(output_data['PhoneBatteryChargingDuration'][
+            'key.userId'].unique(), ['2a02e53a-951e-4fd0-b47f-195a87096bd0'])
+        self.assertEqual(output_data['StepCountPerDay'][
+            'key.userId'].unique(), ['2a02e53a-951e-4fd0-b47f-195a87096bd0'])
+        self.assertTrue(all(output_data[
+            'PhoneBatteryChargingDuration'].groupby('key.userId').size() <= 100))
+        self.assertTrue(all(output_data[
+            'StepCountPerDay'].groupby('key.userId').size() <= 100))
diff --git a/tests/tests_common/test_utils.py b/tests/tests_common/test_utils.py
index d173e19..0e3fe4b 100644
--- a/tests/tests_common/test_utils.py
+++ b/tests/tests_common/test_utils.py
@@ -80,17 +80,18 @@ def setUp(self):
             "config_with_incorrect_spark.yaml"
         self.TESTDATA_FILENAME_WRONG = "tests/resources/config.yaml"
         self.TESTDATA_FILENAME_EMPTY = "tests/resources/test_config.yaml"
-
+        self.TESTDATA_FILENAME_USER_SAMPLING = "tests/resources/test_yamls/test_config_user_sampling.yaml"
+        self.TESTDATA_FILENAME_DATA_SAMPLING = "tests/resources/test_yamls/test_config_date_sampling.yaml"
+        self.TESTDATA_FILENAME_ALL_SAMPLING = "tests/resources/test_yamls/test_config_all_sampling.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_type': 'mock',
+                'source_type': 'mock',
                 'config': {'source_path': 'mockdata/mockdata'},
                 'data_format': 'csv'
             },
@@ -114,14 +115,13 @@ def test_read_unavailable_yaml(self):
 
     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',
+                'source_type': 'mock',
                 'config': {'source_path': 'mockdata/mockdata'},
                 'data_format': 'csv'
             },
@@ -137,6 +137,7 @@ def test_read_yaml_with_spark_config(self):
                 'data_format': 'csv',
                 'compress': False},
             'spark_config': {
+                "spark_master": "local",
                 "spark.executor.instances": 2,
                 "spark.memory.offHeap.enabled": False,
                 "spark.executor.cores": 4,
@@ -149,3 +150,105 @@ def test_read_yaml_with_spark_config(self):
     def test_read_yaml_with_incorrect_spark_config(self):
         with self.assertRaises(YAMLValidationError):
             read_yaml(self.TESTDATA_FILENAME_INCORRECT_SPARK)
+
+    def test_read_yaml_with_user_sampling(self):
+        config = read_yaml(self.TESTDATA_FILENAME_USER_SAMPLING)
+        expected_config = {
+            'project': {
+                'project_name': 'mock_project',
+                'description': 'mock_description',
+                'version': 'mock_version'},
+            'input': {
+                'source_type': 'mock',
+                'config': {'source_path': 'mockdata/mockdata'},
+                'data_format': 'csv'
+            },
+            'configurations': {
+                'df_type': 'pandas',
+                'user_sampling': {
+                    'method': 'userid',
+                    'config': {'userids': ["2a02e53a-951e-4fd0-b47f-195a87096bd0"]}
+                }
+            },
+            '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}}
+        self.assertDictEqual(config, expected_config)
+
+    def test_read_yaml_with_data_sampling(self):
+        config = read_yaml(self.TESTDATA_FILENAME_DATA_SAMPLING)
+        expected_config = {
+            'project': {
+                'project_name': 'mock_project',
+                'description': 'mock_description',
+                'version': 'mock_version'},
+            'input': {
+                'source_type': 'mock',
+                'config': {'source_path': 'mockdata/mockdata'},
+                'data_format': 'csv'
+            },
+            'configurations': {
+                'df_type': 'pandas',
+                'data_sampling': {
+                    'method': 'time',
+                    'config': {
+                        'starttime': '2018-11-25 00:00:00',
+                        'endtime': '2018-11-29 00:00:00',
+                        'time_column': 'value.time'
+                    }
+                }
+            },
+            '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}
+        }
+        self.assertDictEqual(config, expected_config)
+
+    def test_read_yaml_with_all_sampling(self):
+        config = read_yaml(self.TESTDATA_FILENAME_ALL_SAMPLING)
+        expected_config = {
+            'project': {
+                'project_name': 'mock_project',
+                'description': 'mock_description',
+                'version': 'mock_version'},
+            'input': {
+                'source_type': 'mock',
+                'config': {'source_path': 'mockdata/mockdata'},
+                'data_format': 'csv'
+            },
+            'configurations': {
+                'df_type': 'pandas',
+                'user_sampling': {
+                    'method': 'count',
+                    'config': {'count': "2"}
+                },
+                'data_sampling': {
+                    'method': 'count',
+                    'config': {'count': "100"}
+                }
+            },
+            '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}}
+        self.assertDictEqual(config, expected_config)
diff --git a/tests/tests_datalib/test_radar_user_data.py b/tests/tests_datalib/test_radar_user_data.py
index 666065b..7978cdf 100644
--- a/tests/tests_datalib/test_radar_user_data.py
+++ b/tests/tests_datalib/test_radar_user_data.py
@@ -1,16 +1,19 @@
 from radarpipeline.datalib import RadarVariableData, RadarUserData
+from radarpipeline.common.utils import PySparkTestCase
 import unittest
 import os
 import pandas as pd
-from pandas.testing import assert_frame_equal
+from pyspark.testing import assertDataFrameEqual
 
 
-class TestRadarUserData(unittest.TestCase):
+class TestRadarUserData(PySparkTestCase):
     def setUp(self):
-        PANDAS_MOCK_PATH = ("tests/resources/test_data/test_participant/"
-                            "android_phone_step_count/0000_11.csv.gz")
-        self.mock_pandas = pd.read_csv(PANDAS_MOCK_PATH)
-        self.radar_variable_data = RadarVariableData(self.mock_pandas)
+        MOCK_PATH = ("tests/resources/test_data/test_participant/"
+                     "android_phone_step_count/0000_11.csv.gz")
+        self.mock_df = self.spark.read.csv(MOCK_PATH,
+                                           header=True,
+                                           inferSchema=True)
+        self.radar_variable_data = RadarVariableData(self.mock_df)
         self.radar_user_data = RadarUserData({"test_variable_data":
                                               self.radar_variable_data})
 
diff --git a/tests/tests_datalib/test_radar_variable_data.py b/tests/tests_datalib/test_radar_variable_data.py
index c4c3b04..bf1de78 100644
--- a/tests/tests_datalib/test_radar_variable_data.py
+++ b/tests/tests_datalib/test_radar_variable_data.py
@@ -1,24 +1,33 @@
 from radarpipeline.datalib import RadarVariableData
+from radarpipeline.common.utils import PySparkTestCase
 import unittest
 import os
 import pandas as pd
-from pandas.testing import assert_frame_equal
+from pyspark.sql.types import TimestampType
+import pyspark.sql.functions as f
+from pyspark.testing import assertDataFrameEqual
 
 
-class TestRadarVariableData(unittest.TestCase):
+class TestRadarVariableData(PySparkTestCase):
+
     def setUp(self):
-        PANDAS_MOCK_PATH = ("tests/resources/test_data/test_participant/"
-                            "android_phone_step_count/0000_11.csv.gz")
-        self.mock_pandas = pd.read_csv(PANDAS_MOCK_PATH)
-        self.radar_variable_data = RadarVariableData(self.mock_pandas)
+        MOCK_PATH = ("tests/resources/test_data/test_participant/"
+                     "android_phone_step_count/0000_11.csv.gz")
+        self.mock_df = self.spark.read.csv(MOCK_PATH,
+                                           header=True,
+                                           inferSchema=True)
+        self.mock_df = self.preprocess_data(self.mock_df)
+        self.radar_variable_data = RadarVariableData(self.mock_df)
 
     def test_get_data(self):
-        assert_frame_equal(self.radar_variable_data.get_data(), self.mock_pandas)
+        assertDataFrameEqual(self.radar_variable_data.get_data(), self.mock_df,
+                             checkRowOrder=True)
 
     def test_get_data_keys(self):
         self.assertEqual(self.radar_variable_data.get_data_keys(),
-                         list(self.mock_pandas.columns))
+                         list(self.mock_df.columns))
 
     def test_get_data_sizes(self):
         self.assertEqual(self.radar_variable_data.get_data_size(),
-                         len(self.mock_pandas.index))
+                         int(self.mock_df.count()))
+
diff --git a/tests/tests_datalib/test_radardata.py b/tests/tests_datalib/test_radardata.py
index 3888595..ba18781 100644
--- a/tests/tests_datalib/test_radardata.py
+++ b/tests/tests_datalib/test_radardata.py
@@ -2,15 +2,21 @@
 import unittest
 import os
 import pandas as pd
+from radarpipeline.common.utils import PySparkTestCase
+import pyspark.sql.functions as f
+from pyspark.sql.types import TimestampType
 from pandas.testing import assert_frame_equal
 
 
-class TestRadarData(unittest.TestCase):
+class TestRadarData(PySparkTestCase):
+
     def setUp(self):
-        PANDAS_MOCK_PATH = ("tests/resources/test_data/test_participant/"
-                            "android_phone_step_count/0000_11.csv.gz")
-        self.mock_pandas = pd.read_csv(PANDAS_MOCK_PATH)
-        self.radar_variable_data = RadarVariableData(self.mock_pandas)
+        MOCK_PATH = ("tests/resources/test_data/test_participant/"
+                     "android_phone_step_count/0000_11.csv.gz")
+        self.mock_df = self.spark.read.csv(MOCK_PATH,
+                                           header=True,
+                                           inferSchema=True)
+        self.radar_variable_data = RadarVariableData(self.mock_df)
         self.radar_user_data = RadarUserData({"test_variable_data":
                                               self.radar_variable_data})
         self.radar_data = RadarData({"test_user_data": self.radar_user_data})
@@ -33,6 +39,8 @@ def test_get_all_user_ids(self):
         self.assertEqual(self.radar_data._get_all_user_ids(), ["test_user_data"])
 
     def test_get_combined_data_by_variable(self):
+        self.mock_df = self.preprocess_data(self.mock_df)
+        self.mock_pandas = self.mock_df.toPandas()
         assert_frame_equal(self.radar_data.get_combined_data_by_variable(
             "test_variable_data"),
             self.mock_pandas)
diff --git a/tests/tests_io/test_reader.py b/tests/tests_io/test_reader.py
index 5084c12..266a7cb 100644
--- a/tests/tests_io/test_reader.py
+++ b/tests/tests_io/test_reader.py
@@ -9,68 +9,40 @@
 from radarpipeline.common import constants
 from pandas.testing import assert_frame_equal
 from numpy.testing import assert_array_equal
-
+from radarpipeline.common.utils import PySparkTestCase
 from pyspark.sql.types import StructField, StructType
 
 
-class TestSparkCSVDataReader(unittest.TestCase):
+class TestSparkCSVDataReader(PySparkTestCase):
     def setUp(self):
-        mock_config = {"config": {"source_path": "tests/resources/test_data/"}}
+        mock_config = {
+            "input": {
+                "config": {
+                    "source_path": "tests/resources/test_data/"},
+                "data_format": "csv"}}
         data_list = ['android_phone_step_count']
-        self.sparkcsvdatareader = SparkCSVDataReader(mock_config,
+        self.sparkcsvdatareader = SparkCSVDataReader(self.spark, mock_config,
                                                      required_data=data_list)
-        PANDAS_MOCK_PATH = ("tests/resources/test_data/test_participant/"
-                            "android_phone_step_count/0000_11.csv.gz")
-        self.mock_pandas = pd.read_csv(PANDAS_MOCK_PATH)
-        self.radar_variable_data = RadarVariableData(self.mock_pandas)
+        MOCK_PATH = ("tests/resources/test_data/test_participant/"
+                     "android_phone_step_count/0000_11.csv.gz")
+        self.mock_df = self.spark.read.csv(MOCK_PATH,
+                                           header=True,
+                                           inferSchema=True)
+        self.radar_variable_data = RadarVariableData(self.mock_df)
         self.radar_user_data = RadarUserData({"android_phone_step_count":
                                               self.radar_variable_data})
         self.radar_data = RadarData({"test_participant": self.radar_user_data})
 
     def test_read_data(self):
+        mock_dataframe = self.preprocess_data(self.mock_df)
         spark_data = self.sparkcsvdatareader.read_data()
-        assert_array_equal(spark_data.get_data()["test_participant"].get_data()
-                           ["android_phone_step_count"].get_data().values,
-                           self.mock_pandas.values)
-        self.assertTrue(isinstance(spark_data, RadarData))
-        self.assertTrue(isinstance(spark_data.get_data()["test_participant"],
-                                   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",
-            "spark.log.level": "ERROR"}
-        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()
+        self.assertEqual(set(spark_data.get_data()["test_participant"].get_data()
+                             ["android_phone_step_count"].get_data().collect()),
+                         set(mock_dataframe.collect()))
+        self.assertIsInstance(spark_data, RadarData)
+        self.assertIsInstance(spark_data.get_data()["test_participant"], RadarUserData)
+        self.assertIsInstance(spark_data.get_data()["test_participant"].get_data()
+                              ["android_phone_step_count"], RadarVariableData)
 
 
 class TestAvroSchemaReader(unittest.TestCase):
diff --git a/tests/tests_project/test_project.py b/tests/tests_project/test_project.py
index 41edbc8..6e2ebf8 100644
--- a/tests/tests_project/test_project.py
+++ b/tests/tests_project/test_project.py
@@ -71,6 +71,7 @@ def test_get_total_required_data(self):
         self.assertListEqual(sorted(required_data_output), sorted(expected_data))
 
     def tearDown(self) -> None:
+        self.project.close_spark_session()
         del self.project
 
 
@@ -94,7 +95,5 @@ 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_sparkengine.py b/tests/tests_project/test_sparkengine.py
new file mode 100644
index 0000000..1d9f0b2
--- /dev/null
+++ b/tests/tests_project/test_sparkengine.py
@@ -0,0 +1,61 @@
+import unittest
+from radarpipeline.project import SparkEngine
+
+
+class TestSparkDefaultConfig(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.default_spark_config = {'spark.executor.instances': 2,
+                                    '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',
+                                    'spark.log.level': "OFF"}
+        cls.spark_engine = SparkEngine()
+        cls.spark = cls.spark_engine.initialize_spark_session()
+
+    def test_spark_config(self):
+        spark_config_output = dict(self.spark.sparkContext.
+                                   getConf().getAll())
+        for key, value in self.default_spark_config.items():
+            self.assertEqual(spark_config_output[key], str(value))
+
+    def test_spark_config_dict(self):
+        spark_config_output = self.default_spark_config
+        self.assertDictEqual(self.default_spark_config, spark_config_output)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.spark_engine.close_spark_session()
+
+
+class TestSparkCustomConfig(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.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",
+            "spark.log.level": "ERROR"}
+        cls.spark_engine = SparkEngine(spark_config=cls.spark_config)
+        cls.spark = cls.spark_engine.initialize_spark_session()
+
+    def test_spark_config(self):
+        spark_config_output = dict(self.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.spark_config
+        self.assertDictEqual(self.spark_config, spark_config_output)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.spark_engine.close_spark_session()
diff --git a/tests/tests_project/test_validations.py b/tests/tests_project/test_validations.py
index 33b7a94..9d292e9 100644
--- a/tests/tests_project/test_validations.py
+++ b/tests/tests_project/test_validations.py
@@ -9,7 +9,7 @@
         'description': 'mock_description',
         'version': 'mock_version'},
     'input': {
-        'data_type': 'mock',
+        'source_type': 'mock',
         'config': {
             'source_path': 'mockdata/mockdata'},
         'data_format': 'csv'},
@@ -40,7 +40,7 @@ def test_validate_mock(self):
 
     def test_validate_sftp(self):
         config = self.mock_config
-        config['input']['data_type'] = 'sftp'
+        config['input']['source_type'] = 'sftp'
         config['input']['config'] = {
             "sftp_host": "mock_host",
             "sftp_source_path": "mock_source_path",
@@ -54,7 +54,7 @@ def test_validate_sftp(self):
 
     def test_validate_invalid_sftp(self):
         config = self.mock_config
-        config['input']['data_type'] = 'sftp'
+        config['input']['source_type'] = 'sftp'
         config['input']['config'] = {
             "sftp_host": "mock_host",
             "sftp_source_path": "mock_source_path",
@@ -68,7 +68,7 @@ def test_validate_invalid_sftp(self):
 
     def test_validate_local(self):
         config = self.mock_config
-        config['input']['data_type'] = 'local'
+        config['input']['source_type'] = 'local'
         config['input']['config'] = {
             "source_path": "mockdata/mockdata"
         }
@@ -78,7 +78,7 @@ def test_validate_local(self):
 
     def test_validate_local_wrong_source_path(self):
         config = self.mock_config
-        config['input']['data_type'] = 'local'
+        config['input']['source_type'] = 'local'
         config['input']['config'] = {
             "source_path": "xyz"
         }