tools/heron/tracker_stats.py

# Copyright 2018 Twitter, Inc.
# Licensed under the Apache License, Version 2.0
# http://www.apache.org/licenses/LICENSE-2.0

""" Script for displaying statistics for all topologies registered with a
Heron Tracker instance."""

import os
import sys
import logging
import argparse
import pickle

import datetime as dt

from typing import List, Dict, Union, Any

import requests

import pandas as pd

from caladrius import logs
from caladrius.common.heron import tracker
from caladrius.metrics.heron.topology import groupings

LOG: logging.Logger = logging.getLogger("caladrius.tools.heron.tracker_stats")


def summarise_groupings(tracker_url: str,
                        topologies: pd.DataFrame = None) -> pd.DataFrame:
    """ Summarises the stream grouping counts of all topologies registered with
    the supplied Tracker instance.

    Arguments:
        tracker_url (str):  The URL for the Heron Tracker API
        topologies (pd.DataFrame):  The topologies summary from the heron
                                    tracker can be supplied, if not it will
                                    fetched fresh from the trackerAPI.

    Returns:
        A DataFrame with columns for:
        topology: The topology ID
        cluster: The cluster the topology is running on
        environ: The environment the topology is running in
        user: The user that uploaded the topology
        A column for each type of stream grouping as well as combinations of
        stream grouping (incoming grouping)->(outgoing grouping) and their
        associate frequency count for each topology.
    """
    if topologies is None:
        topologies = tracker.get_topologies(tracker_url)
    output: pd.DataFrame = None

    for (cluster, environ), data in topologies.groupby(["cluster", "environ"]):
        for topology in data.topology:

            try:
                grouping_summary: Dict[str, int] = \
                    groupings.summary(tracker_url, topology, cluster, environ)
            except requests.HTTPError:
                LOG.warning("Unable to fetch grouping summary for topology: "
                            "%s, cluster: %s, environ: %s", topology, cluster,
                            environ)
            else:
                grouping_summary["topology"] = topology
                grouping_summary["cluster"] = cluster
                grouping_summary["environ"] = environ
                grouping_df: pd.DataFrame = pd.DataFrame([grouping_summary])

                if output is None:
                    output = grouping_df
                else:
                    output = output.append(grouping_df)

    output = output.merge(topologies, on=["topology","environ","cluster"])
    return output


def add_pplan_info(tracker_url: str,
                   topologies: pd.DataFrame = None) -> pd.DataFrame:
    """ Combines information from the topology summary DataFrame with
    information from the physical plan of each topology.

    Arguments:
        tracker_url (str):  The URL for the Heron Tracker API
        topologies (pd.DataFrame):  The topologies summary from the heron
                                    tracker can be supplied, if not it will
                                    fetched fresh from the Tracker API.

    Returns:
        pandas.DataFrame:   The topologies summary DataFrame with physical plan
        information added. This will return a new DataFrame and will not modify
        the supplied DataFrame
    """
    if topologies is None:
        topologies = tracker.get_topologies(tracker_url)

    output: List[Dict[str, Union[str, float, List[int]]]] = []

    for (cluster, environ, user), data in topologies.groupby(["cluster",
                                                              "environ",
                                                              "user"]):
        for topology_id in data.topology:

            try:
                pplan: Dict[str, Any] = tracker.get_physical_plan(
                    tracker_url, cluster, environ, topology_id)
            except requests.HTTPError:
                # If we cannot fetch the plan, skip this topology
                continue

            # Add information from the configuration dictionary
            config: Dict[str, str] = pplan["config"]
            row: Dict[str, Union[str, float, List[int]]] = {}
            row["topology"] = topology_id
            row["cluster"] = cluster
            row["environ"] = environ
            row["user"] = user
            for key, value in config.items():

                # Some of the custom config values are large dictionaries or
                # lists so we will skip them
                if isinstance(value, (dict, list)):
                    continue

                # Replace "." with "_" in the key name so we can use namespace
                # calls on the DataFrame
                new_key: str = "_".join(key.split(".")[1:])

                # Try to convert any values that numeric so we can do summary
                # stats
                try:
                    new_value: Union[str, float] = float(value)
                except ValueError:
                    new_value = value
                except TypeError:
                    LOG.error("Value of key: %s was not a string or number it"
                              " was a %s", key, str(type(value)))

                row[new_key] = new_value

            # Add instances stats for this topology
            row["total_instances"] = len(pplan["instances"])
            row["instances_per_container_dist"] = \
                [len(pplan["stmgrs"][stmgr]["instance_ids"])
                 for stmgr in pplan["stmgrs"]]
            row["total_bolts"] = len(pplan["bolts"])
            row["total_spouts"] = len(pplan["spouts"])
            row["total_components"] = len(pplan["bolts"]) + len(pplan["spouts"])

            output.append(row)

    return pd.DataFrame(output)


def add_logical_plan_info(tracker_url: str,
                   topologies: pd.DataFrame = None) -> pd.DataFrame:
    """ Combines information from the topology summary DataFrame with
    information from the logical plan of each topology.

    Arguments:
        tracker_url (str):  The URL for the Heron Tracker API
        topologies (pd.DataFrame):  The topologies summary from the heron
                                    tracker can be supplied, if not it will
                                    fetched fresh from the Tracker API.

    Returns:
        pandas.DataFrame:   The topologies summary DataFrame with logical plan
        information added. This will return a new DataFrame and will not modify
        the supplied DataFrame
    """
    if topologies is None:
        topologies = tracker.get_topologies(tracker_url)
    output: List[Dict] = []

    for (cluster, environ, user), data in topologies.groupby(["cluster",
                                                              "environ",
                                                              "user"]):
        for topology_id in data.topology:

            try:
                logical_plan: Dict[str, Any] = tracker.get_logical_plan(
                    tracker_url, cluster, environ, topology_id)
            except requests.HTTPError:
                # If we cannot fetch the plan, skip this topology
                continue

            # there are two possible kinds of spouts:
            spout_single_output = 0
            spout_multiple_output = 0

            # there are six possible kinds of bolts:
            # two of which are sinks, and four are intermediate bolts
            # sink types
            bolt_single_in_zero_out = 0
            bolt_multiple_in_zero_out = 0

            # intermediate types:
            bolt_single_in_single_out = 0
            bolt_multiple_in_single_out = 0
            bolt_single_in_multiple_out = 0
            bolt_multiple_in_multiple_out = 0

            row: Dict = {}
            LOG.info("Topology ID: %s",  topology_id)
            for key in logical_plan["spouts"].keys():
                num_outputs = len(logical_plan["spouts"][key]["outputs"])
                if num_outputs == 1:
                    spout_single_output = spout_single_output + 1
                else:
                    spout_multiple_output = spout_multiple_output + 1
            for key in logical_plan["bolts"].keys():
                outputs = len(logical_plan["bolts"][key]["outputs"])
                inputs = len(logical_plan["bolts"][key]["inputs"])

                # sinks
                if outputs == 0:
                    if inputs == 1:
                        bolt_single_in_zero_out = bolt_single_in_zero_out + 1
                    elif inputs > 1:
                        bolt_multiple_in_zero_out = bolt_multiple_in_zero_out + 1
                elif outputs == 1:
                    if inputs == 1:
                        bolt_single_in_single_out = bolt_single_in_single_out + 1
                    elif inputs > 1:
                        bolt_multiple_in_single_out = bolt_multiple_in_single_out + 1
                elif outputs > 1:
                    if inputs == 1:
                        bolt_single_in_multiple_out = bolt_single_in_multiple_out + 1
                    elif inputs > 1:
                        bolt_multiple_in_multiple_out = bolt_multiple_in_multiple_out + 1

            row["topology"] = topology_id

            row["spout_single_output"] = spout_single_output
            row["spout_multiple_output"] = spout_multiple_output

            row["bolt_single_in_zero_out"] = bolt_single_in_zero_out
            row["bolt_multiple_in_zero_out"] = bolt_multiple_in_zero_out

            row["bolt_single_in_single_out"] = bolt_single_in_single_out
            row["bolt_multiple_in_single_out"] = bolt_multiple_in_single_out
            row["bolt_single_in_multiple_out"] = bolt_single_in_multiple_out
            row["bolt_multiple_in_multiple_out"] = bolt_multiple_in_multiple_out
            output.append(row)

    return pd.DataFrame(output)


def _get_mg_summary(topo_pplan: pd.DataFrame, groupby_term: str):

    mg_summary: pd.DataFrame = pd.DataFrame(
        topo_pplan.groupby([groupby_term,
                            "reliability_mode"]).topology.count())
    mg_summary = mg_summary.reset_index()
    mg_summary.rename(index=str,
                      columns={"topology": f"mg_{groupby_term}_count"},
                      inplace=True)
    mg_summary = mg_summary.merge(
        (mg_summary.groupby(groupby_term)[f"mg_{groupby_term}_count"].sum()
         .reset_index().rename(
             index=str,
             columns={f"mg_{groupby_term}_count":
                      f"mg_{groupby_term}_total"})),
        on=groupby_term)

    mg_summary["overall_percentage"] = \
        (mg_summary[f"mg_{groupby_term}_count"] /
         topo_pplan.reliability_mode.count()
         * 100)
    mg_summary[f"{groupby_term}_percentage"] = \
        (mg_summary[f"mg_{groupby_term}_count"] /
         mg_summary[f"mg_{groupby_term}_total"] * 100)

    return mg_summary


def _create_parser() -> argparse.ArgumentParser:

    parser: argparse.ArgumentParser = argparse.ArgumentParser(
        description=("This program provides various statistics about the "
                     "topologies running on a Heron Tracker instance"))

    parser.add_argument("-t", "--tracker", required=False, type=str,
                        help=("The URL to the Heron Tracker API"))

    parser.add_argument("-r", "--reload", required=False, action="store_true",
                        help=("If supplied then fresh information will be "
                              "pulled from the Heron Tracker API. Otherwise "
                              "cached data will be used."))

    parser.add_argument("-cd", "--cache_dir", required=False, type=str,
                        default="/tmp/caladrius/heron/tracker/stats",
                        help=("The temporary directory to store Tracker "
                              "information."))

    parser.add_argument("-o", "--output", required=False, type=str,
                        help=("Output file path for the statistic summary to "
                              "be printed to. If not supplied out will be "
                              "sent to standard out."))

    parser.add_argument("-q", "--quiet", required=False, action="store_true",
                        help=("Optional flag indicating if console log output "
                              "should be suppressed"))

    parser.add_argument("--debug", required=False, action="store_true",
                        help=("Optional flag indicating if debug level "
                              "information should be displayed"))

    return parser


def _check_tracker(tracker_str: str) -> str:

    if "http://" in tracker_str:
        return tracker_str

    return "http://" + tracker_str


if __name__ == "__main__":

    ARGS: argparse.Namespace = _create_parser().parse_args()

    if not ARGS.quiet:
        logs.setup(debug=ARGS.debug)

    NO_CACHE_DIR: bool = False
    if not os.path.exists(ARGS.cache_dir):
        LOG.info("No cached tracker information present")
        NO_CACHE_DIR = True
        os.makedirs(ARGS.cache_dir)
        LOG.info("Created cache directory at: %s", ARGS.cache_dir)
    elif len(os.listdir(ARGS.cache_dir)) < 1:
        LOG.info("No cached tracker information present")
        # If the directory exists but there is nothing in it then we need to
        # load the tracker data
        NO_CACHE_DIR = True

    CREATE_TIME_FILE: str = os.path.join(ARGS.cache_dir, "created.pkl")
    TOPO_FILE: str = os.path.join(ARGS.cache_dir, "topo.pkl")
    TOPO_PPLAN_FILE: str = os.path.join(ARGS.cache_dir, "topo_pplan.pkl")
    TOPO_LPLAN_FILE: str = os.path.join(ARGS.cache_dir, "topo_lplan.pkl")
    GROUPING_SUMMARY_FILE: str = os.path.join(ARGS.cache_dir,
                                              "grouping_summary.pkl")

    if ARGS.reload or NO_CACHE_DIR:

        if not ARGS.tracker:
            err_msg: str = ("In order to load information from the Tracker API"
                            " the tracker URL must be supplied via the "
                            "-t / --tracker argument")
            if ARGS.quiet:
                print(err_msg, file=sys.stderr)
            else:
                LOG.error(err_msg)
            # Exit with error code 2 (usually means cli syntax error)
            sys.exit(2)

        TRACKER_URL: str = _check_tracker(ARGS.tracker)

        LOG.info("Fetching new data from the Heron Tracker API at %s",
                 TRACKER_URL)

        # Get the list of topologies registered with the heron tracker
        TOPOLOGIES: pd.DataFrame = tracker.get_topologies(TRACKER_URL)
        LOG.info("Caching topology data at %s", TOPO_FILE)
        TOPOLOGIES.to_pickle(TOPO_FILE)

        # Add physical plan options
        TOPO_PPLAN: pd.DataFrame = add_pplan_info(TRACKER_URL, TOPOLOGIES)
        LOG.info("Caching topology physical plan data at %s",
                 TOPO_PPLAN_FILE)
        TOPO_PPLAN.to_pickle(TOPO_PPLAN_FILE)

        # Add logical plan information
        TOPO_LPLAN: pd.DataFrame = add_logical_plan_info(TRACKER_URL, TOPOLOGIES)
        LOG.info("Caching topology logical plan data at %s",
                 TOPO_LPLAN_FILE)
        TOPO_LPLAN.to_pickle(TOPO_LPLAN_FILE)
        # Get the stream grouping summary
        GROUPING_SUMMARY: pd.DataFrame = summarise_groupings(TRACKER_URL,
                                                             TOPOLOGIES)
        LOG.info("Caching stream grouping summary data at %s",
                 GROUPING_SUMMARY_FILE)
        GROUPING_SUMMARY.to_pickle(GROUPING_SUMMARY_FILE)

        # Save a time stamp so we know how old the data is
        NOW: dt.datetime = dt.datetime.utcnow().replace(tzinfo=dt.timezone.utc)

        with open(CREATE_TIME_FILE, 'wb') as time_file:
            pickle.dump(NOW, time_file)

    else:

        with open(CREATE_TIME_FILE, 'rb') as time_file:
            timestamp: dt.datetime = pickle.load(time_file)

        NOW = dt.datetime.utcnow().replace(tzinfo=dt.timezone.utc)
        duration: dt.timedelta = NOW - timestamp

        LOG.info("Loading data that is %s hours old",
                 str(round(duration.total_seconds() / 3600, 2)))

        LOG.info("Loading topology data from %s", TOPO_FILE)
        TOPOLOGIES = pd.read_pickle(TOPO_FILE)

        LOG.info("Loading topology physical plan data from %s",
                 TOPO_PPLAN_FILE)
        TOPO_PPLAN = pd.read_pickle(TOPO_PPLAN_FILE)

        LOG.info("Loading topology logical plan data from %s",
                 TOPO_PPLAN_FILE)
        TOPO_LPLAN = pd.read_pickle(TOPO_LPLAN_FILE)

        LOG.info("Loading stream grouping summary data from %s",
                 GROUPING_SUMMARY_FILE)
        GROUPING_SUMMARY = pd.read_pickle(GROUPING_SUMMARY_FILE)

    # Overall topology counts
    TOTAL_TOPOS: int = TOPOLOGIES.topology.count()
    TOPOS_BY_CLUSTER: pd.DataFrame = pd.DataFrame(
        TOPOLOGIES.groupby("cluster").topology.count())
    TOPOS_BY_CLUSTER["percentage"] = ((TOPOS_BY_CLUSTER.topology /
                                       TOTAL_TOPOS) * 100)
    TOPOS_BY_CLUSTER = TOPOS_BY_CLUSTER.reset_index()
    TOPOS_BY_CLUSTER.rename(index=str,
                            columns={"topology": "topo_cluster_count"},
                            inplace=True)

    TOPOS_BY_ENV: pd.DataFrame = pd.DataFrame(
        TOPOLOGIES.groupby("environ").topology.count())
    TOPOS_BY_ENV["percentage"] = ((TOPOS_BY_ENV.topology /
                                   TOTAL_TOPOS) * 100)
    TOPOS_BY_ENV = TOPOS_BY_ENV.reset_index()

    MG_TOTAL_TOPOS: int = TOPO_PPLAN.reliability_mode.count()

    # Message Guarantee stats

    # Overall
    MG_OVERALL: pd.Series = \
        (TOPO_PPLAN.groupby("reliability_mode").topology.count())
    MG_OVERALL = MG_OVERALL.reset_index()
    MG_OVERALL.rename(index=str, columns={"topology": "mg_overall_count"},
                      inplace=True)
    MG_OVERALL["percentage"] = (MG_OVERALL.mg_overall_count /
                                MG_TOTAL_TOPOS * 100)

    # By Cluster
    MG_BY_CLUSTER: pd.DataFrame = _get_mg_summary(TOPO_PPLAN, "cluster")

    # By Environment
    MG_BY_ENV: pd.DataFrame = _get_mg_summary(TOPO_PPLAN, "environ")

    # Topology structure stats
    LINEAR_TOPOS = TOPO_LPLAN[(TOPO_LPLAN['spout_multiple_output'] == 0) &
                              (TOPO_LPLAN['bolt_multiple_in_zero_out'] == 0) &
                              (TOPO_LPLAN['bolt_multiple_in_multiple_out'] == 0) &
                              (TOPO_LPLAN['bolt_multiple_in_single_out'] == 0) &
                              (TOPO_LPLAN['bolt_single_in_multiple_out'] == 0)].reset_index().shape[0] \
                   / TOTAL_TOPOS * 100

    # We look for the most common type of operator
    SPOUT_MULTIPLE_OUTPUTS = \
        TOPO_LPLAN[(TOPO_LPLAN['spout_multiple_output'] > 0)].reset_index().shape[0] / TOTAL_TOPOS * 100
    SINK_MULTIPLE_INPUTS = \
        TOPO_LPLAN[(TOPO_LPLAN['bolt_multiple_in_zero_out'] > 0)].reset_index().shape[0] / TOTAL_TOPOS * 100
    BOLT_MULTIPLE_IN_OUT = \
        TOPO_LPLAN[(TOPO_LPLAN['bolt_multiple_in_multiple_out'] > 0)].reset_index().shape[0] / TOTAL_TOPOS * 100
    BOLT_MULTIPLE_IN_SINGLE_OUT = \
        TOPO_LPLAN[(TOPO_LPLAN['bolt_multiple_in_single_out'] > 0)].reset_index().shape[0] / TOTAL_TOPOS * 100
    BOLT_SINGLE_IN_MULTIPLE_OUT = \
        TOPO_LPLAN[(TOPO_LPLAN['bolt_single_in_multiple_out'] > 0)].reset_index().shape[0] / TOTAL_TOPOS * 100
    # We look for topologies that have both of the most common non-linear operator types:
    COMMON = TOPO_LPLAN[(TOPO_LPLAN['bolt_multiple_in_single_out'] > 0) &
                        (TOPO_LPLAN['bolt_multiple_in_zero_out'] > 0)].reset_index()
    # Grouping stats
    JUST_GROUPINGS: pd.DataFrame = \
        (GROUPING_SUMMARY.drop(["topology", "cluster", "environ", "user"],
                               axis=1))

    GROUPING_OVERALL: pd.Series = \
        (JUST_GROUPINGS.count() / TOTAL_TOPOS * 100)

    GROUPING_ENVIRON: pd.Series = \
        (GROUPING_SUMMARY.groupby(["environ"]).count() / TOTAL_TOPOS * 100)

    SINGLE_GROUPING: List[Dict[str, Union[int, float]]] = []
    for grouping in [g for g in list(JUST_GROUPINGS.columns) if "->" not in g]:
        grouping_only_count: int = \
            (JUST_GROUPINGS[JUST_GROUPINGS[grouping].notna() &
                            (JUST_GROUPINGS.isna().sum(axis=1) ==
                             len(JUST_GROUPINGS.columns) - 1)]
             [grouping].count())
        SINGLE_GROUPING.append({"Grouping": grouping,
                                "Frequency": grouping_only_count,
                                "% of Total":
                                (grouping_only_count / TOTAL_TOPOS * 100)})

    PERCENTILES: List[float] = [.10, .25, .5, .75, .95, .99]

    if ARGS.output:
        LOG.info("Saving output to file: %s", ARGS.output)
        OUT_FILE = open(ARGS.output, "w")
    else:
        LOG.info("Sending output to standard out")
        OUT_FILE = sys.stdout

    print("-------------------", file=OUT_FILE)
    print("Heron Tracker Stats", file=OUT_FILE)
    print("-------------------", file=OUT_FILE)

    print(f"\nTotal topologies: {TOTAL_TOPOS}", file=OUT_FILE)
    print("\nTotal topologies by cluster:\n", file=OUT_FILE)
    print(TOPOS_BY_CLUSTER.to_string(index=False), file=OUT_FILE)
    print("\nTotal topologies by environment:\n", file=OUT_FILE)
    print(TOPOS_BY_ENV.to_string(index=False), file=OUT_FILE)

    print("\n-------------------", file=OUT_FILE)
    print("Container stats:\n", file=OUT_FILE)

    print(TOPO_PPLAN.stmgrs.describe(percentiles=PERCENTILES).to_string(),
          file=OUT_FILE)

    print("\nTop 10 Largest topologies by container count:\n", file=OUT_FILE)
    print(TOPO_PPLAN.sort_values(by="stmgrs", ascending=False)
          [["topology", "cluster", "environ", "user", "total_instances",
            "stmgrs"]]
          .head(10).to_string(index=False), file=OUT_FILE)

    print("\n-------------------", file=OUT_FILE)
    print("Instance stats:\n", file=OUT_FILE)

    print("\nStatistics for total number of instances per topology:\n",
          file=OUT_FILE)
    print(TOPO_PPLAN.total_instances.describe(
        percentiles=PERCENTILES).to_string(), file=OUT_FILE)

    print("\nTop 10 Largest topologies by instance count:\n", file=OUT_FILE)
    print(TOPO_PPLAN.sort_values(by="total_instances", ascending=False)
          [["topology", "cluster", "environ", "user", "total_instances",
            "stmgrs"]].head(10).to_string(index=False), file=OUT_FILE)

    print("\nStatistics for instances per container:\n", file=OUT_FILE)
    output: List[int] = []
    for index, dist in TOPO_PPLAN.instances_per_container_dist.iteritems():
        output.extend(dist)
    print(pd.Series(output).describe(percentiles=PERCENTILES).to_string(),
          file=OUT_FILE)

    print("\n-------------------", file=OUT_FILE)
    print("Component stats:\n", file=OUT_FILE)

    print("\nStatistics for total number of components per topology:\n",
          file=OUT_FILE)
    print(TOPO_PPLAN.total_components.describe(
        percentiles=PERCENTILES).to_string(), file=OUT_FILE)

    print("\nTop 20 Largest topologies by component count:\n", file=OUT_FILE)
    print(TOPO_PPLAN.sort_values(by="total_components", ascending=False)
          [["topology", "cluster", "environ", "user", "total_components",
            "stmgrs", "total_spouts", "total_bolts", "total_instances"]]
          .head(20).to_string(index=False), file=OUT_FILE)

    print("\n-------------------", file=OUT_FILE)
    print("Topology structure stats:\n", file=OUT_FILE)
    print("\nPercentage of topologies with linear structure: ",
          LINEAR_TOPOS, file=OUT_FILE)
    print("\nPercentage of topologies with spouts with multiple outputs: ",
          SPOUT_MULTIPLE_OUTPUTS, file=OUT_FILE)
    print("\nPercentage of topologies with sinks with multiple inputs: ",
          SINK_MULTIPLE_INPUTS, file=OUT_FILE)
    print("\nPercentage of topologies with bolts with multiple inputs"
          " and outputs: ",
          BOLT_MULTIPLE_IN_OUT, file=OUT_FILE)
    print("\nPercentage of topologies with bolts with multiple inputs"
          " and a single output: ",
          BOLT_MULTIPLE_IN_SINGLE_OUT, file=OUT_FILE)
    print("\nPercentage of topologies with bolts with single input "
          "and multiple outputs: ",
          BOLT_SINGLE_IN_MULTIPLE_OUT, file=OUT_FILE)
    print("\nPercentage of topologies with both intermediate bolts with"
          " multiple inputs and sinks with multiple inputs: ",
          COMMON.shape[0]/TOTAL_TOPOS * 100, file=OUT_FILE)

    print("\n-------------------", file=OUT_FILE)
    print("Message guarantee stats:\n", file=OUT_FILE)

    print("\nTopologies with each message guarantee type - Overall\n",
          file=OUT_FILE)
    print(MG_OVERALL.to_string(), file=OUT_FILE)

    print("\nTopologies with each message guarantee type by Cluster\n",
          file=OUT_FILE)
    print(MG_BY_CLUSTER.set_index(["cluster",
                                   "reliability_mode"]).to_string(),
          file=OUT_FILE)

    print("\nTopologies with each message guarantee type by Environment\n",
          file=OUT_FILE)
    print(MG_BY_ENV.set_index(["environ", "reliability_mode"]).to_string(),
          file=OUT_FILE)

    print("\n-------------------", file=OUT_FILE)
    print("Stream grouping stats:\n", file=OUT_FILE)

    print("\nPercentage of topologies with each grouping - Overall:\n",
          file=OUT_FILE)
    print(GROUPING_OVERALL.to_string(), file=OUT_FILE)

    print("\nPercentage of topologies with each grouping - Per Environment:\n",
          file=OUT_FILE)
    print(GROUPING_ENVIRON.drop(["topology", "cluster", "user"],
                                axis=1).to_string(), file=OUT_FILE)

    print("\nTopologies with only a single grouping type:\n", file=OUT_FILE)
    print(pd.DataFrame(SINGLE_GROUPING).to_string(index=False), file=OUT_FILE)