csit.infra.dash/app/cdash/data/data.py

   1 # Copyright (c) 2023 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Prepare data for Plotly Dash applications.
  15 """
  16
  17 import logging
  18 import resource
  19 import awswrangler as wr
  20 import pandas as pd
  21
  22 from yaml import load, FullLoader, YAMLError
  23 from datetime import datetime, timedelta
  24 from time import time
  25 from pytz import UTC
  26 from awswrangler.exceptions import EmptyDataFrame, NoFilesFound
  27
  28
  29 class Data:
  30     """Gets the data from parquets and stores it for further use by dash
  31     applications.
  32     """
  33
  34     def __init__(self, data_spec_file: str) -> None:
  35         """Initialize the Data object.
  36
  37         :param data_spec_file: Path to file specifying the data to be read from
  38             parquets.
  39         :type data_spec_file: str
  40         :raises RuntimeError: if it is not possible to open data_spec_file or it
  41             is not a valid yaml file.
  42         """
  43
  44         # Inputs:
  45         self._data_spec_file = data_spec_file
  46
  47         # Specification of data to be read from parquets:
  48         self._data_spec = list()
  49
  50         # Data frame to keep the data:
  51         self._data = {
  52             "statistics": pd.DataFrame(),
  53             "trending": pd.DataFrame(),
  54             "iterative": pd.DataFrame(),
  55             "coverage": pd.DataFrame()
  56         }
  57
  58         # Read from files:
  59         try:
  60             with open(self._data_spec_file, "r") as file_read:
  61                 self._data_spec = load(file_read, Loader=FullLoader)
  62         except IOError as err:
  63             raise RuntimeError(
  64                 f"Not possible to open the file {self._data_spec_file,}\n{err}"
  65             )
  66         except YAMLError as err:
  67             raise RuntimeError(
  68                 f"An error occurred while parsing the specification file "
  69                 f"{self._data_spec_file,}\n"
  70                 f"{err}"
  71             )
  72
  73     @property
  74     def data(self):
  75         return self._data
  76
  77     @staticmethod
  78     def _get_list_of_files(
  79             path,
  80             last_modified_begin=None,
  81             last_modified_end=None,
  82             days=None
  83         ) -> list:
  84         """Get list of interested files stored in S3 compatible storage and
  85         returns it.
  86
  87         :param path: S3 prefix (accepts Unix shell-style wildcards)
  88             (e.g. s3://bucket/prefix) or list of S3 objects paths
  89             (e.g. [s3://bucket/key0, s3://bucket/key1]).
  90         :param last_modified_begin: Filter the s3 files by the Last modified
  91             date of the object. The filter is applied only after list all s3
  92             files.
  93         :param last_modified_end: Filter the s3 files by the Last modified date
  94             of the object. The filter is applied only after list all s3 files.
  95         :param days: Number of days to filter.
  96         :type path: Union[str, List[str]]
  97         :type last_modified_begin: datetime, optional
  98         :type last_modified_end: datetime, optional
  99         :type days: integer, optional
 100         :returns: List of file names.
 101         :rtype: list
 102         """
 103         file_list = list()
 104         if days:
 105             last_modified_begin = datetime.now(tz=UTC) - timedelta(days=days)
 106         try:
 107             file_list = wr.s3.list_objects(
 108                 path=path,
 109                 suffix="parquet",
 110                 last_modified_begin=last_modified_begin,
 111                 last_modified_end=last_modified_end
 112             )
 113             logging.debug("\n".join(file_list))
 114         except NoFilesFound as err:
 115             logging.error(f"No parquets found.\n{err}")
 116         except EmptyDataFrame as err:
 117             logging.error(f"No data.\n{err}")
 118
 119         return file_list
 120
 121     @staticmethod
 122     def _create_dataframe_from_parquet(
 123             path, partition_filter=None,
 124             columns=None,
 125             validate_schema=False,
 126             last_modified_begin=None,
 127             last_modified_end=None,
 128             days=None
 129         ) -> pd.DataFrame:
 130         """Read parquet stored in S3 compatible storage and returns Pandas
 131         Dataframe.
 132
 133         :param path: S3 prefix (accepts Unix shell-style wildcards)
 134             (e.g. s3://bucket/prefix) or list of S3 objects paths
 135             (e.g. [s3://bucket/key0, s3://bucket/key1]).
 136         :param partition_filter: Callback Function filters to apply on PARTITION
 137             columns (PUSH-DOWN filter). This function MUST receive a single
 138             argument (Dict[str, str]) where keys are partitions names and values
 139             are partitions values. Partitions values will be always strings
 140             extracted from S3. This function MUST return a bool, True to read
 141             the partition or False to ignore it. Ignored if dataset=False.
 142         :param columns: Names of columns to read from the file(s).
 143         :param validate_schema: Check that individual file schemas are all the
 144             same / compatible. Schemas within a folder prefix should all be the
 145             same. Disable if you have schemas that are different and want to
 146             disable this check.
 147         :param last_modified_begin: Filter the s3 files by the Last modified
 148             date of the object. The filter is applied only after list all s3
 149             files.
 150         :param last_modified_end: Filter the s3 files by the Last modified date
 151             of the object. The filter is applied only after list all s3 files.
 152         :param days: Number of days to filter.
 153         :type path: Union[str, List[str]]
 154         :type partition_filter: Callable[[Dict[str, str]], bool], optional
 155         :type columns: List[str], optional
 156         :type validate_schema: bool, optional
 157         :type last_modified_begin: datetime, optional
 158         :type last_modified_end: datetime, optional
 159         :type days: integer, optional
 160         :returns: Pandas DataFrame or None if DataFrame cannot be fetched.
 161         :rtype: DataFrame
 162         """
 163         df = pd.DataFrame()
 164         start = time()
 165         if days:
 166             last_modified_begin = datetime.now(tz=UTC) - timedelta(days=days)
 167         try:
 168             df = wr.s3.read_parquet(
 169                 path=path,
 170                 path_suffix="parquet",
 171                 ignore_empty=True,
 172                 validate_schema=validate_schema,
 173                 use_threads=True,
 174                 dataset=True,
 175                 columns=columns,
 176                 partition_filter=partition_filter,
 177                 last_modified_begin=last_modified_begin,
 178                 last_modified_end=last_modified_end
 179             )
 180             df.info(verbose=True, memory_usage="deep")
 181             logging.debug(
 182                 f"\nCreation of dataframe {path} took: {time() - start}\n"
 183             )
 184         except NoFilesFound as err:
 185             logging.error(
 186                 f"No parquets found in specified time period.\n"
 187                 f"Nr of days: {days}\n"
 188                 f"last_modified_begin: {last_modified_begin}\n"
 189                 f"{err}"
 190             )
 191         except EmptyDataFrame as err:
 192             logging.error(
 193                 f"No data in parquets in specified time period.\n"
 194                 f"Nr of days: {days}\n"
 195                 f"last_modified_begin: {last_modified_begin}\n"
 196                 f"{err}"
 197             )
 198
 199         return df
 200
 201     def read_all_data(self, days: int=None) -> dict:
 202         """Read all data necessary for all applications.
 203
 204         :param days: Number of days to filter. If None, all data will be
 205             downloaded.
 206         :type days: int
 207         :returns: A dictionary where keys are names of parquets and values are
 208             the pandas dataframes with fetched data.
 209         :rtype: dict(str: pandas.DataFrame)
 210         """
 211
 212         lst_trending = list()
 213         lst_iterative = list()
 214         lst_coverage = list()
 215
 216         for data_set in self._data_spec:
 217             logging.info(
 218                 f"Reading data for {data_set['data_type']} "
 219                 f"{data_set['partition_name']} {data_set.get('release', '')}"
 220             )
 221             partition_filter = lambda part: True \
 222                 if part[data_set["partition"]] == data_set["partition_name"] \
 223                     else False
 224             if data_set["data_type"] in ("trending", "statistics"):
 225                 time_period = days
 226             else:
 227                 time_period = None
 228             data = Data._create_dataframe_from_parquet(
 229                 path=data_set["path"],
 230                 partition_filter=partition_filter,
 231                 columns=data_set.get("columns", None),
 232                 days=time_period
 233             )
 234
 235             if data_set["data_type"] == "statistics":
 236                 self._data["statistics"] = data
 237             elif data_set["data_type"] == "trending":
 238                 lst_trending.append(data)
 239             elif data_set["data_type"] == "iterative":
 240                 data["release"] = data_set["release"]
 241                 data["release"] = data["release"].astype("category")
 242                 lst_iterative.append(data)
 243             elif data_set["data_type"] == "coverage":
 244                 data["release"] = data_set["release"]
 245                 data["release"] = data["release"].astype("category")
 246                 lst_coverage.append(data)
 247             else:
 248                 raise NotImplementedError(
 249                     f"The data type {data_set['data_type']} is not implemented."
 250                 )
 251
 252         self._data["iterative"] = pd.concat(
 253             lst_iterative,
 254             ignore_index=True,
 255             copy=False
 256         )
 257         self._data["trending"] = pd.concat(
 258             lst_trending,
 259             ignore_index=True,
 260             copy=False
 261         )
 262         self._data["coverage"] = pd.concat(
 263             lst_coverage,
 264             ignore_index=True,
 265             copy=False
 266         )
 267
 268         for key in self._data.keys():
 269             logging.info(
 270                 f"\nData frame {key}:"
 271                 f"\n{self._data[key].memory_usage(deep=True)}\n"
 272             )
 273             self._data[key].info(verbose=True, memory_usage="deep")
 274
 275         mem_alloc = \
 276             resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1000
 277         logging.info(f"Memory allocation: {mem_alloc:.0f}MB")
 278
 279         return self._data