Updating Schema reading procedures and refactoring

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
+    compress: false

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'

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)

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:

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")

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))

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

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

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()
 

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()