--- /dev/null
+# Copyright (c) 2022 Cisco and/or its affiliates.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Prepare data for Plotly Dash applications.
+"""
+
+import logging
+import awswrangler as wr
+
+from yaml import load, FullLoader, YAMLError
+from datetime import datetime, timedelta
+from time import time
+from pytz import UTC
+from pandas import DataFrame
+from awswrangler.exceptions import EmptyDataFrame, NoFilesFound
+
+
+class Data:
+ """Gets the data from parquets and stores it for further use by dash
+ applications.
+ """
+
+ def __init__(self, data_spec_file: str, debug: bool=False) -> None:
+ """Initialize the Data object.
+
+ :param data_spec_file: Path to file specifying the data to be read from
+ parquets.
+ :param debug: If True, the debuf information is printed to stdout.
+ :type data_spec_file: str
+ :type debug: bool
+ :raises RuntimeError: if it is not possible to open data_spec_file or it
+ is not a valid yaml file.
+ """
+
+ # Inputs:
+ self._data_spec_file = data_spec_file
+ self._debug = debug
+
+ # Specification of data to be read from parquets:
+ self._data_spec = None
+
+ # Data frame to keep the data:
+ self._data = None
+
+ # Read from files:
+ try:
+ with open(self._data_spec_file, "r") as file_read:
+ self._data_spec = load(file_read, Loader=FullLoader)
+ except IOError as err:
+ raise RuntimeError(
+ f"Not possible to open the file {self._data_spec_file,}\n{err}"
+ )
+ except YAMLError as err:
+ raise RuntimeError(
+ f"An error occurred while parsing the specification file "
+ f"{self._data_spec_file,}\n"
+ f"{err}"
+ )
+
+ @property
+ def data(self):
+ return self._data
+
+ def _get_columns(self, parquet: str) -> list:
+ """Get the list of columns from the data specification file to be read
+ from parquets.
+
+ :param parquet: The parquet's name.
+ :type parquet: str
+ :raises RuntimeError: if the parquet is not defined in the data
+ specification file or it does not have any columns specified.
+ :returns: List of columns.
+ :rtype: list
+ """
+
+ try:
+ return self._data_spec[parquet]["columns"]
+ except KeyError as err:
+ raise RuntimeError(
+ f"The parquet {parquet} is not defined in the specification "
+ f"file {self._data_spec_file} or it does not have any columns "
+ f"specified.\n{err}"
+ )
+
+ def _get_path(self, parquet: str) -> str:
+ """Get the path from the data specification file to be read from
+ parquets.
+
+ :param parquet: The parquet's name.
+ :type parquet: str
+ :raises RuntimeError: if the parquet is not defined in the data
+ specification file or it does not have the path specified.
+ :returns: Path.
+ :rtype: str
+ """
+
+ try:
+ return self._data_spec[parquet]["path"]
+ except KeyError as err:
+ raise RuntimeError(
+ f"The parquet {parquet} is not defined in the specification "
+ f"file {self._data_spec_file} or it does not have the path "
+ f"specified.\n{err}"
+ )
+
+ def _get_list_of_files(self,
+ path,
+ last_modified_begin=None,
+ last_modified_end=None,
+ days=None) -> list:
+ """Get list of interested files stored in S3 compatible storage and
+ returns it.
+
+ :param path: S3 prefix (accepts Unix shell-style wildcards)
+ (e.g. s3://bucket/prefix) or list of S3 objects paths
+ (e.g. [s3://bucket/key0, s3://bucket/key1]).
+ :param last_modified_begin: Filter the s3 files by the Last modified
+ date of the object. The filter is applied only after list all s3
+ files.
+ :param last_modified_end: Filter the s3 files by the Last modified date
+ of the object. The filter is applied only after list all s3 files.
+ :param days: Number of days to filter.
+ :type path: Union[str, List[str]]
+ :type last_modified_begin: datetime, optional
+ :type last_modified_end: datetime, optional
+ :type days: integer, optional
+ :returns: List of file names.
+ :rtype: List
+ """
+ if days:
+ last_modified_begin = datetime.now(tz=UTC) - timedelta(days=days)
+ try:
+ file_list = wr.s3.list_objects(
+ path=path,
+ suffix="parquet",
+ last_modified_begin=last_modified_begin,
+ last_modified_end=last_modified_end
+ )
+ if self._debug:
+ logging.info("\n".join(file_list))
+ except NoFilesFound as err:
+ logging.error(f"No parquets found.\n{err}")
+ except EmptyDataFrame as err:
+ logging.error(f"No data.\n{err}")
+
+ return file_list
+
+ def _create_dataframe_from_parquet(self,
+ path, partition_filter=None,
+ columns=None,
+ validate_schema=False,
+ last_modified_begin=None,
+ last_modified_end=None,
+ days=None) -> DataFrame:
+ """Read parquet stored in S3 compatible storage and returns Pandas
+ Dataframe.
+
+ :param path: S3 prefix (accepts Unix shell-style wildcards)
+ (e.g. s3://bucket/prefix) or list of S3 objects paths
+ (e.g. [s3://bucket/key0, s3://bucket/key1]).
+ :param partition_filter: Callback Function filters to apply on PARTITION
+ columns (PUSH-DOWN filter). This function MUST receive a single
+ argument (Dict[str, str]) where keys are partitions names and values
+ are partitions values. Partitions values will be always strings
+ extracted from S3. This function MUST return a bool, True to read
+ the partition or False to ignore it. Ignored if dataset=False.
+ :param columns: Names of columns to read from the file(s).
+ :param validate_schema: Check that individual file schemas are all the
+ same / compatible. Schemas within a folder prefix should all be the
+ same. Disable if you have schemas that are different and want to
+ disable this check.
+ :param last_modified_begin: Filter the s3 files by the Last modified
+ date of the object. The filter is applied only after list all s3
+ files.
+ :param last_modified_end: Filter the s3 files by the Last modified date
+ of the object. The filter is applied only after list all s3 files.
+ :param days: Number of days to filter.
+ :type path: Union[str, List[str]]
+ :type partition_filter: Callable[[Dict[str, str]], bool], optional
+ :type columns: List[str], optional
+ :type validate_schema: bool, optional
+ :type last_modified_begin: datetime, optional
+ :type last_modified_end: datetime, optional
+ :type days: integer, optional
+ :returns: Pandas DataFrame or None if DataFrame cannot be fetched.
+ :rtype: DataFrame
+ """
+ df = None
+ start = time()
+ if days:
+ last_modified_begin = datetime.now(tz=UTC) - timedelta(days=days)
+ try:
+ df = wr.s3.read_parquet(
+ path=path,
+ path_suffix="parquet",
+ ignore_empty=True,
+ validate_schema=validate_schema,
+ use_threads=True,
+ dataset=True,
+ columns=columns,
+ partition_filter=partition_filter,
+ last_modified_begin=last_modified_begin,
+ last_modified_end=last_modified_end
+ )
+ if self._debug:
+ df.info(verbose=True, memory_usage='deep')
+ logging.info(
+ u"\n"
+ f"Creation of dataframe {path} took: {time() - start}"
+ u"\n"
+ )
+ except NoFilesFound as err:
+ logging.error(f"No parquets found.\n{err}")
+ except EmptyDataFrame as err:
+ logging.error(f"No data.\n{err}")
+
+ self._data = df
+ return df
+
+ def check_datasets(self, days: int=None):
+ """Read structure from parquet.
+
+ :param days: Number of days back to the past for which the data will be
+ read.
+ :type days: int
+ """
+ self._get_list_of_files(path=self._get_path("trending"), days=days)
+ self._get_list_of_files(path=self._get_path("statistics"), days=days)
+
+ def read_stats(self, days: int=None) -> tuple:
+ """Read statistics from parquet.
+
+ It reads from:
+ - Suite Result Analysis (SRA) partition,
+ - NDRPDR trending partition,
+ - MRR trending partition.
+
+ :param days: Number of days back to the past for which the data will be
+ read.
+ :type days: int
+ :returns: tuple of pandas DataFrame-s with data read from specified
+ parquets.
+ :rtype: tuple of pandas DataFrame-s
+ """
+
+ l_stats = lambda part: True if part["stats_type"] == "sra" else False
+ l_mrr = lambda part: True if part["test_type"] == "mrr" else False
+ l_ndrpdr = lambda part: True if part["test_type"] == "ndrpdr" else False
+
+ return (
+ self._create_dataframe_from_parquet(
+ path=self._get_path("statistics"),
+ partition_filter=l_stats,
+ columns=self._get_columns("statistics"),
+ days=days
+ ),
+ self._create_dataframe_from_parquet(
+ path=self._get_path("statistics-trending-mrr"),
+ partition_filter=l_mrr,
+ columns=self._get_columns("statistics-trending-mrr"),
+ days=days
+ ),
+ self._create_dataframe_from_parquet(
+ path=self._get_path("statistics-trending-ndrpdr"),
+ partition_filter=l_ndrpdr,
+ columns=self._get_columns("statistics-trending-ndrpdr"),
+ days=days
+ )
+ )
+
+ def read_trending_mrr(self, days: int=None) -> DataFrame:
+ """Read MRR data partition from parquet.
+
+ :param days: Number of days back to the past for which the data will be
+ read.
+ :type days: int
+ :returns: Pandas DataFrame with read data.
+ :rtype: DataFrame
+ """
+
+ lambda_f = lambda part: True if part["test_type"] == "mrr" else False
+
+ return self._create_dataframe_from_parquet(
+ path=self._get_path("trending-mrr"),
+ partition_filter=lambda_f,
+ columns=self._get_columns("trending-mrr"),
+ days=days
+ )
+
+ def read_trending_ndrpdr(self, days: int=None) -> DataFrame:
+ """Read NDRPDR data partition from iterative parquet.
+
+ :param days: Number of days back to the past for which the data will be
+ read.
+ :type days: int
+ :returns: Pandas DataFrame with read data.
+ :rtype: DataFrame
+ """
+
+ lambda_f = lambda part: True if part["test_type"] == "ndrpdr" else False
+
+ return self._create_dataframe_from_parquet(
+ path=self._get_path("trending-ndrpdr"),
+ partition_filter=lambda_f,
+ columns=self._get_columns("trending-ndrpdr"),
+ days=days
+ )
+
+ def read_iterative_mrr(self, release: str) -> DataFrame:
+ """Read MRR data partition from iterative parquet.
+
+ :param release: The CSIT release from which the data will be read.
+ :type release: str
+ :returns: Pandas DataFrame with read data.
+ :rtype: DataFrame
+ """
+
+ lambda_f = lambda part: True if part["test_type"] == "mrr" else False
+
+ return self._create_dataframe_from_parquet(
+ path=self._get_path("iterative-mrr").format(release=release),
+ partition_filter=lambda_f,
+ columns=self._get_columns("iterative-mrr")
+ )
+
+ def read_iterative_ndrpdr(self, release: str) -> DataFrame:
+ """Read NDRPDR data partition from parquet.
+
+ :param release: The CSIT release from which the data will be read.
+ :type release: str
+ :returns: Pandas DataFrame with read data.
+ :rtype: DataFrame
+ """
+
+ lambda_f = lambda part: True if part["test_type"] == "ndrpdr" else False
+
+ return self._create_dataframe_from_parquet(
+ path=self._get_path("iterative-ndrpdr").format(release=release),
+ partition_filter=lambda_f,
+ columns=self._get_columns("iterative-ndrpdr")
+ )
Code Review / csit.git / blobdiff