<jupyter_start><jupyter_text>Analysis of LSST Batch Activity at CC-IN2P3*Source: [https://github.com/airnandez/alba](https://github.com/airnandez/alba)**Author: <NAME>, CC-IN2P3*  Introduction The purpose of this notebook is to analyse accounting information emitted by GridEngine about the activity of LSST batch jobs executed at CC-IN2P3.<jupyter_code>import pathlib
import datetime
import sys
import re
import collections

import IPython.display
print_md = IPython.display.Markdown

import pandas as pd
import numpy as np
import bokeh
import bokeh.plotting as bkh
import bokeh.models as bkhmodels
bkh.output_notebook()<jupyter_output><empty_output><jupyter_text>DependenciesThis notebook uses the packages below listed below. The links point to their documentation:* bokeh* numpy* pandas* python<jupyter_code>table = f"""
These are the versions of those dependencies you are currently using:

| Component | Version              |
| --------: | -------------------- |
|  bokeh    | {bokeh.__version__}  |
|  numpy    | {np.version.version} |
|  pandas   | {pd.__version__}     |
|  python   | {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro} |
"""
print_md(table)<jupyter_output><empty_output><jupyter_text>Load the accounting records into a Pandas dataframe Set the data directory, where JSON accounting files are located. The repository of accounting data at CC-IN2P3 is `/sps/hep/ccin2p3/data/BatchFiles`. In that directory one can find one JSON file per month (e.g. `GE.accounting.2018.11.json`) which contains records of batch jobs of all groups using CC-IN2P3's batch service.Records for jobs relevant for LSST can be found at CC-IN2P3 at `/sps/lsst/groups/accounting`. Those records are copied at the LSST Science Platform under `/path/to/location/at/ncsa`.There is a JSON file per month (e.g. `GE.accounting.2018.11.lsst.json.gz`) which only contains accounting records for jobs submitted by members of the Unix group `lsst`:<jupyter_code># The locations below are searched for accounting data. The first location found is used.
search_paths = (
    # For development, we assume that data files are located in the './data' directory,
    # relative to the location of this notebook
    pathlib.Path.cwd().joinpath('data'),
    
    # Location at CC-IN2P3
    pathlib.Path('/sps/lsst/groups/accounting'),
    
    # Location at the LSST Science Platform at NCSA
    pathlib.Path('/lsst/jhome/fabioh/DATA/accounting'),
    
    # Location at NERSC
    pathlib.Path('/global/projecta/projectdirs/lsst/global/in2p3/accounting'),
)

data_dir = None
for p in search_paths:
    if p.exists():
        data_dir = p
        break
        
if not data_dir:
    raise ValueError("Could not find the location of the accounting data")

d = data_dir if data_dir != pathlib.Path.cwd().joinpath('data') else './data'
print_md(f'Data directory: `{d}`')<jupyter_output><empty_output><jupyter_text>Import all the files available in the data directory:<jupyter_code># To analyse ALL the accounting files in the directory use
#    pattern = 'GE.accounting.*.json*'
# To analyse data of all months for a single year use:
#    pattern = 'GE.accounting.2018.*.json*'
# To analyse data of some months of a given year use:
#   pattern = 'GE.accounting.2018.1[0-1]*.json*'
patterns = ('GE.accounting.2018.*.json*', 'GE.accounting.2019.*.json*')

df = pd.DataFrame()
for p in sorted(patterns):
    for path in sorted(data_dir.glob(p)):
        print(f'Loading data file {pathlib.PurePosixPath(path).name}')
        df = df.append(pd.read_json(path, lines=True))
    
rows = df.shape[0]
if rows == 0:
    raise Exception("no job records could be loaded")
    
print_md(f'Loaded **{rows}** job accounting records')<jupyter_output>Loading data file GE.accounting.2018.01.lsst.json.gz
Loading data file GE.accounting.2018.02.lsst.json.gz
Loading data file GE.accounting.2018.03.lsst.json.gz
Loading data file GE.accounting.2018.04.lsst.json.gz
Loading data file GE.accounting.2018.05.lsst.json.gz
Loading data file GE.accounting.2018.06.lsst.json.gz
Loading data file GE.accounting.2018.07.lsst.json.gz
Loading data file GE.accounting.2018.08.lsst.json.gz
Loading data file GE.accounting.2018.09.lsst.json.gz
Loading data file GE.accounting.2018.10.lsst.json.gz
Loading data file GE.accounting.2018.11.lsst.json.gz
Loading data file GE.accounting.2018.12.lsst.json.gz
Loading data file GE.accounting.2019.01.lsst.json.gz
Loading data file GE.accounting.2019.02.lsst.json.gz
Loading data file GE.accounting.2019.03.lsst.json.gz<jupyter_text>Dataframe column renamingIn this section we modify the dataframe to rename some columns so to make their contents explicit and intuitive in the remaining of this notebook, if needed.<jupyter_code>if False:
    df.rename(inplace=True, columns={
        # Modify column names as below
        'old_name':  'new_name',

        # Rename some fields to make explicit it is an extension to the raw data
        'x_worker_might_sectohs06sec': 'x_worker_hs06sec_factor',
    })<jupyter_output><empty_output><jupyter_text>Dataframe contents overview Each row in the dataframe contains information about a single job. This information is emitted by GridEngine, preprocessed, extended and reformatted in JSON to make it easier to analyse (see [here](https://gitlab.in2p3.fr/Batch-tools/GE-batch-system)).Below is the list of columns in the dataframe, shown in alphabetical order. The columns prefixed by `x_` (e.g. `x_cputime_hs06sec`) are extensions to the data provided by GridEngine.<jupyter_code>num_rows, num_columns = df.shape
columns = sorted(list(df))
print_md(f'Dataframe size: **{num_rows} rows, {num_columns} columns:**\n\n `{", ".join(columns)}`')<jupyter_output><empty_output><jupyter_text>The semantics of each column is described in the table below:| Column                         | Semantics                                                                                    ||:-------------------------------|:---------------------------------------------------------------------------------------------|| **`account`**                  | account string associated to the job, specified at submission time via `qsub(1)`, see `accounting(5)` || **`ar_submission_time`**       | advance reservation submission time, see `accounting(5)` || **`arid`**                     | advance reservation identifier, see `accounting(5)` || **`category`**                 | a string specifying the job category, see `accounting(5)` || **`cpu`**                      | number of **scaled CPU seconds (integrated over all the CPU slots allocated for this job)** this job was in executing state. Note that the scaling factor is specific for each worker node and can be retrieved via `qconf -se ` (field `usage_scaling`). The value of the scaling factor is available in column `x_worker_hs06sec_factor`. The number of slots allocated for this job is available in the column `slots` (see below). See `accounting(5)`| **`cwd`**                      | working  directory the job ran in,  see `accounting(5)` || **`department`**               | the department which was assigned to the job, see `accounting(5)` || **`exit_status`**              | exit status of the job script, see `accounting(5)` || **`failed`**                   | indicates the problem which occurred in case a job could not be started, see `accounting(5)` || **`granted_pe`**               | parallel environment which was selected for that job, see `accounting(5)` || **`group`**                    | effective group id of the job owner when executing the job, see `accounting(5)` || **`hostname`**                 | name of the execution host, see `accounting(5)` || **`io`**                       | amount of read and written by the job, in GBytes,  see `accounting(5)` || **`ioops`**                    | number of io operations, see `accounting(5)` || **`iow`**                      | I/O wait time in seconds, see `accounting(5)` || **`job_class`**                | if the job has been running in a job class, the name of the job class, see `accounting(5)` || **`job_name`**                 | job name (or `QLOGIN` if this is an interactive job), see `accounting(5)` || **`job_number`**               | job identifier, see `accounting(5)` || **`maxpss_GB`**                | maximum proportional set size in GBytes, see `accounting(5)` || **`maxrss_GB`**                | maximum resident set size in GBytes, see `accounting(5)` || **`maxvmem_GB`**               | the maximum vmem size in GBytes, see `accounting(5)` || **`mem`**                      | integral memory usage in GBytes CPU seconds, see `accounting(5)` || **`slots`**                    | number of (job) slots which were dispatched to the job by the scheduler, see `accounting(5)` || **`owner`**                    | user name who submitted the job, see `accounting(5)` || **`pe_taskid`**                | if this identifier is set the task was part of a parallel job, see `accounting(5)` || **`priority`**                 | priority value assigned to the job, see `accounting(5)` || **`project`**                  | the project which was assigned to the job, see `accounting(5)` || **`qdel_info`**                | if the job has been deleted via `qdel`, username@hostname, else `NONE`, see `accounting(5)` || **`queue_name`**               | name of the cluster queue in which the job has run, see `qname` in `accounting(5)` || **`ru_*`**                     | standard UNIX rusage structure as described in `getrusage(2)`. Fields are: `ru_idrss`, `ru_inblock`, `ru_ismrss`, `ru_isrss`, `ru_ixrss`, `ru_majflt`, `ru_maxrss_GB`, `ru_minflt`, `ru_msgrcv`, `ru_msgsnd`, `ru_nicsw`, `ru_nsignals`, `ru_nswap`, `ru_nvcsw`, `ru_oublock`, `ru_stime`, `ru_utime`. See `accounting(5)` || **`ru_wallclock_sec`**         | wallclock execution time of the job, in seconds, see `accounting(5)` || **`submit_cmd`**               | the command line used for job submission, see `accounting(5)` || **`submit_host`**              | submit host name, see `accounting(5)` || **`task_number`**              | array job task index number, see `accounting(5)` || **`wallclock`**                | wallclock time this job spent in running state. Unit is seconds. See `accounting(5)` || ||  **EXTENDED COLUMN**            | **SEMANTICS** || **`x_cputime_hs06sec`**         | number of normalized seconds this job was in execution state, integrated over all the slots allocated for this job (see `slots`). The unit is HS06.sec || **`x_cputime_sec`**             | CPU time (in seconds) consumed by this job, integrated over all the slots allocated to the job (see `slots`). **Warning**: this CPU time is not normalized. See also `x_cputime_hs06sec` || **`x_limits_s_cpu`**            | ??? || **`x_limits_s_rss_gb`**         | ??? || **`x_limits_s_vmem_gb`**        | ??? || **`x_limits_s_vmem_gb`**        | ??? || **`x_mem_max_pss_gb`**          | maximum proportional set size in GBytes, see `accounting(5)` and `getrusage(2)` || **`x_mem_max_rss_gb`**          | maximum resident set size in GBytes, see `accounting(5)` and `getrusage(2)` || **`x_mem_max_vmem_gb`**         | the maximum vmem size in GBytes, see `accounting(5)` and `getrusage(2)` || **`x_mem_ru_maxrss_gb`**        | ??? || **`x_scheduler`**               | Identifier of the batch job scheduler (typically 'UGE') || **`x_worker_might_sectohs06sec`**   | CPU scaling factor for the worker node this job executed on. This factor is used to compute the execution time of this job in normalized units, namely in HS06.sec. || **`x_worker_model_type`**       | CPU model of the worker node where the job executed, e.g. 'Intel(R) Xeon(R) CPU E5-2680 v2 @ 2.80GHz' || **`x_worker_physical_virtual`** | is the worker node a 'physical' or 'virtual' machine? || **`x_worker_processors_thread_count`**        | number of (maybe virtual) CPU cores in this worker node || **`x_worker_product_name`**     | worker node product name, as assigned by vendor e.g. 'PowerEdge C6220 II' |<jupyter_code>df.head()<jupyter_output><empty_output><jupyter_text>Dataframe extension Extend the dataframe by computing convenient values and make sure the columns referring to a timestamp are interpreted by Pandas as a timestamp:<jupyter_code># Make sure columns 'submission_time', 'start_time' and 'end_time' are timestamps
for col in ('submission_time', 'start_time', 'end_time'):
    df[col] = df[col].astype('datetime64[s]')

# Compute a new column which contains the waiting time in queue for each job
# We prefix each new column name with '_' to visually distinguish it from the columns in the
# original data files
df['_waiting_time_sec'] = (df['start_time'] - df['submission_time']).apply(lambda d: d.total_seconds())

# Modify the CPU model to remove '(R)'
df['x_worker_model_type'] = df['x_worker_model_type'].str.replace('(R)', '', regex=False)<jupyter_output><empty_output><jupyter_text>Filter dataframe contentsOnly consider jobs by owners and groups of interest. At CC-IN2P3, jobs for the organized data processing activities of LSST and DESC are executed as users `lsstprod` and `descprod`, respectively. The accounts for members of LSST/DESC belong to the group `lsst`.<jupyter_code># Select the users and groups of interest by specifying them in a list. Use an empty list
# if no selection is needed
users = ('descprod',)
groups = ('lsst',)

if users:
    df = df[df['owner'].isin(users)]

if groups:
    df = df[df['group'].isin(groups)]
    
def user_filter_description(users, groups):
    res = '(all users' if not users else f"(users: {','.join(users)}"
    res += ', all groups)' if not groups else f", groups: {','.join(groups)})"
    return res

print_md(f'Number of jobs for {user_filter_description(users, groups)}: **{df.shape[0]}**')<jupyter_output><empty_output><jupyter_text>Select the time interval of jobs of interest. We only consider jobs submitted within that interval.<jupyter_code>def filter_by_date(dframe, start=None, end=None):
    """Select the job records in the dataframe 'dframe' which submission time is between 'start' and 'end'
    """
    if start:
        dframe = dframe[dframe['submission_time'] >= start]
    if end:
        dframe = dframe[dframe['submission_time'] <= end]
    return dframe


# Use a specific date such as
#    datetime.datetime(2018, 10, 25)
# to specify the start and end dates of interest. Alternatively, use None for processing
# all the available job records.
start_date = None
end_date = None
df = filter_by_date(df, start_date, end_date)
print_md(f'Number of job records in the selected period: **{df.shape[0]}**')<jupyter_output><empty_output><jupyter_text>Helper functions and variables<jupyter_code>def format_pct_hours(v):
    """Format v to be displayed as a percentile, in hours. Argument v is expected to be seconds.
    """
    return "<1" if v < 3600.0 else f'{int(v/3600.0)}'

def format_pct_min(v):
    """Format v to be displayed as a percentile, in minutes. Argument v is expected to be seconds.
    """
    return "<1" if v < 60.0 else f'{int(v/60.0)}'

def style_figure(fig, x_font_style='bold', x_label_standoff=15, y_font_style='bold', y_label_standoff=15):
    """Style some attributes of a Bokeh figure
    """
    # Figure size (in screen units)
    fig.plot_width = 800
    fig.plot_height = 600
    
    # Autohide figure toolbar
    fig.toolbar.autohide = True

    # Figure background fill color and alpha
    fig.background_fill_color = 'whitesmoke'
    fig.background_fill_alpha = 0.8
    
    # Axis properties
    fig.xaxis.axis_label_text_font_style = x_font_style
    fig.xaxis.axis_label_standoff =  x_label_standoff
    fig.yaxis.axis_label_text_font_style = y_font_style
    fig.yaxis.axis_label_standoff =  y_label_standoff
    
# Definition of the batch queues
batch_queues = {
    # Interactive queues
    'interactive': ('interactive', 'mc_interactive'),
    
    # Queues which accept single-node jobs (either single CPU core or multi-CPU core)
    'single-core': ('huge', 'long', 'longlasting'),
    'multi-core':  ('mc_highmem_huge', 'mc_highmem_long', 'mc_huge', 'mc_long', 'mc_longlasting'),
    
    # Queues which accept multi-node jobs
    'multi-node':  ('pa_gpu_long', 'pa_long', 'pa_longlasting', 'pa_medium'),
    'gpu':         ('mc_gpu_interactive', 'mc_gpu_long', 'mc_gpu_longlasting', 'mc_gpu_medium'),
}<jupyter_output><empty_output><jupyter_text>--- Batch system serviceIn this section we analyse the behavior of jobs and the batch system in general terms.  Activity overviewIn this sub-section we present the periods of batch activity<jupyter_code>start_period, end_period = df['submission_time'].min(), df['submission_time'].max()
num_jobs = df.shape[0]
print_md(f'From {start_period.date()} to {end_period.date()} there were {num_jobs} jobs submitted by {user_filter_description(users, groups)}')
# Count submitted and terminated jobs per day
df['_submission_date'] = df['submission_time'].apply(lambda dt: dt.date())
submitted = df.groupby(['_submission_date']).size()

df['_end_date'] = df['end_time'].apply(lambda dt: dt.date())
terminated = df.groupby(['_end_date']).size()

fig = bkh.figure(
    title = f'BATCH ACTIVITY OVERVIEW {user_filter_description(users, groups)}',
    x_axis_label = 'date',
    x_axis_type = 'datetime',
    y_axis_label = 'job count'
)
style_figure(fig)

color = 'steelblue'
fig.circle(x=submitted.index, y=submitted.values, fill_color=color, size=8, legend="submitted")
fig.line(x=submitted.index, y=submitted.values, line_color=color, legend="submitted")

color = 'red'
fig.circle(x=terminated.index, y=terminated.values, fill_color=color, size=8, legend="terminated")
fig.line(x=terminated.index, y=terminated.values, line_color=color, legend="terminated")

fig.xaxis.formatter = bkhmodels.formatters.DatetimeTickFormatter(
    days=["%Y-%m-%d"], months=["%Y-%m-%d"], years=["%Y-%m-%d"]
)
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('date', '@x{%F}'),
        ('jobs', '@y{0,000}'),
    ],
    formatters = {
        'x': 'datetime',
    },
    mode = 'mouse',
))
fig.legend.location = 'top_left'
fig.legend.click_policy = 'hide'
bkh.show(fig)
print_md(f"""*The figure above shows the number of jobs submitted and terminated per day.*""")<jupyter_output><empty_output><jupyter_text>Execution queuesShow the activity per batch queue. Each queue has its own characteristics in terms of CPU time, memory size, etc. You can see the details of each queue [here](http://cctools.in2p3.fr/mrtguser/info_sge_queue.php).<jupyter_code>queues = df.groupby(['qname']).size().sort_values(ascending=False)
labels = [n for n in queues.index]
values = queues.get_values()

fig = bkh.figure(
    title = f'NUMBER OF JOBS PER EXECUTION QUEUE {user_filter_description(users, groups)}',
    x_axis_label = 'queue',
    y_axis_label = 'job count',
    x_range = labels
)
style_figure(fig)
color, alpha = 'darkorange', 0.7
fig.vbar(x=labels, top=values, width=0.7, alpha=alpha, fill_color=color, line_color=color)
fig.xgrid.grid_line_color = None
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('queue', '@x'),
        ('jobs', '@top{0,000}'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
print_md(f"""*The figure above shows the number of jobs executed per batch system queue. Only queues
    where at least one job was put for execution are shown.*""")<jupyter_output><empty_output><jupyter_text>Successful and failed jobsCompute the percentiles of job which succeeded and failed. A job is considered failes if its exit status is not zero.<jupyter_code>def format_failed_pct(v):
    return '{:.0%}'.format(v) if v > 0.0 else 'n/a'

def format_job_count(count):
    return '{:,}'.format(count)

d = {}
queues = df.groupby(['qname']).size().sort_values(ascending=False)
for q in queues.index:
    job_count = queues[q]
    succeeded = df.loc[(df['qname'] == q) & (df['exit_status'] == 0)].shape[0]
    d[q] = {
        'job_count': job_count,
        'succeeded': succeeded/job_count if job_count > 0 else 0,
        'failed':    (job_count - succeeded)/job_count if job_count > 0 else 0,
    }

table = """
### Percentage of failed jobs (`exit_status != 0`)

| Queue | Job Count | Succeeded | Failed |
| ----: | --------: | --------: | -----: |
"""

for k, v in d.items():
    table += f'| **{k}** | {format_job_count(v["job_count"])} | {format_failed_pct(v["succeeded"])} | {format_failed_pct(v["failed"])} |\n'

table += """
*The table above shows the fraction of succeeded and failed jobs per queue. A job is considered failed if its `exit_status` is not zero.*
"""    
print_md(table)<jupyter_output><empty_output><jupyter_text>Waiting timeShow the distribution of the time the jobs spent waiting in the queue before being put in execution in a compute node.<jupyter_code>fig = bkh.figure(
    title = f'DISTRIBUTION OF JOB WAITING TIME IN QUEUE {user_filter_description(users, groups)}',
    x_axis_label = 'waiting time (min)',
    y_axis_label = 'job count',
)
style_figure(fig)
color, alpha = 'firebrick', 0.7
hist, edges = np.histogram(df['_waiting_time_sec']/60, density=False, bins=30)
fig.quad(top=hist, bottom=0, left=edges[:-1], right=edges[1:], alpha=alpha, fill_color=color, line_color=color)
fig.xaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('jobs', '@top{0,000}'),
        ('low',  '@left{0.0} min'),
        ('high', '@right{0.0} min'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
print_md(f"""*The figure above shows the distribution of the time the jobs waited in queue before
    being put in execution by the job scheduler.*""")
d = {}
queues = df.groupby(['qname']).size().sort_values(ascending=False)
for q in queues.index:
    selection = df.loc[df['qname'] == q]
    d[q] = {
        'job_count': selection.shape[0],
        'pcts': np.percentile(selection['_waiting_time_sec'], [50, 80, 98, 100])
    }

table = """
### Waiting time percentiles (*minutes*)

| Queue | Job Count | 50pct | 80pct | 98pct | 100pct |
| ----: | --------: | ----: | ----: | ----: | -----: |
"""

for k, v in d.items():
    table += f'| **{k}** | {format_job_count(v["job_count"])} | {format_pct_min(v["pcts"][0])} | {format_pct_min(v["pcts"][1])} | {format_pct_min(v["pcts"][2])} | {format_pct_min(v["pcts"][3])} |\n'    

table += """
*The table above shows some percentiles for waiting time for each execution queue (unit is minutes).*
"""    
print_md(table)<jupyter_output><empty_output><jupyter_text>Execution timeShow the distribution of the wallclock time the jobs spent in execution.<jupyter_code>fig = bkh.figure(
    title = f'DISTRIBUTION OF JOB EXECUTION WALLCLOCK TIME {user_filter_description(users, groups)}',
    x_axis_label = 'execution wallclock time (hours)',
    y_axis_label = 'job count'
)
style_figure(fig)
color, alpha = 'darkseagreen', 0.7
hist, edges = np.histogram(df['wallclock']/3600, density=False, bins=40)
fig.quad(top=hist, bottom=0, left=edges[:-1], right=edges[1:], alpha=alpha, fill_color=color, line_color=color)
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('jobs', '@top{0,000}'),
        ('low',  '@left{0.0} hours'),
        ('high', '@right{0.0} hours'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
d = {}
queues = df.groupby(['qname']).size().sort_values(ascending=False)
for q in queues.index:
    selection = df.loc[df['qname'] == q]
    d[q] = {
        'job_count': selection.shape[0],
        'pcts': np.percentile(selection['wallclock'], [50, 80, 98, 100])
    }

table = """
### Execution time percentiles (*hours*)

| Queue | Job Count | 50pct | 80pct | 98pct | 100pct |
| ----: | --------: | ----: | ----: | ----: | -----: |
"""

for k, v in d.items():
    table += f'| **{k}** | {format_job_count(v["job_count"])} | {format_pct_hours(v["pcts"][0])} | {format_pct_hours(v["pcts"][1])} | {format_pct_hours(v["pcts"][2])} | {format_pct_hours(v["pcts"][3])} | \n'

table += """
*The table above shows some percentiles for job execution wallclock time per execution queue (unit is hours)*
"""
print_md(table)<jupyter_output><empty_output><jupyter_text>CPU efficiencyIn this section we compute the CPU efficiency of the jobs. CPU efficiency is the ratio CPU time to wallclock time, that is, the fraction of the wallclock time the job actually used the CPU.<jupyter_code># Compute the non-normalized CPU time for each job slot for jobs satisfying the following conditions:
#    1) The exit status of the jobs is zero
#    2) The job executed in a single node batch queue, ie. do not include jobs executed in the 'interactive' queue
#    3) The job spent time in execution at least equivalent to the 5th percentile of the jobs satistying
#       the two previous conditions
queues_to_include = batch_queues['single-core'] + batch_queues['multi-core']
job_set = df.loc[(df['exit_status'] == 0) & (df['qname'].isin(queues_to_include))]
wallclock_threshold = np.percentile(job_set['wallclock'], [5])[0]
job_set = job_set.loc[job_set['wallclock'] >= wallclock_threshold]

# Compute the CPU efficiency
cputime_per_slot = job_set['cpu'] / (job_set['x_worker_might_sectohs06sec'] * job_set['slots'])
cpueff = cputime_per_slot / job_set['wallclock']

hist, edges = np.histogram(cpueff, density=False, bins=40)
fig = bkh.figure(
    title = f'DISTRIBUTION OF CPU EFFICIENCY {user_filter_description(users, groups)}',
    x_axis_label = 'CPU efficiency (CPU time / wallclock time)',
    y_axis_label = 'job count'
)
style_figure(fig)
color, alpha = 'saddlebrown', 0.7
fig.quad(top=hist, bottom=0, left=edges[:-1], right=edges[1:], alpha=alpha, fill_color=color, line_color=color)
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('jobs', '@top{0,000}'),
        ('low',  '@left{0%}'),
        ('high', '@right{0%}'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
print_md(f"""*The figure above shows the distribution of the time the CPU efficiency of selected jobs. Note that there are some jobs which CPU efficiency is greater than 1.0: this is likely
due to some measurement error of the actual CPU time used by the job as collected by GridEngine or an under-estimation of the number of CPU cores actually used by the job*""")<jupyter_output><empty_output><jupyter_text>CPU cores per jobShow the number of job slots that the batch system allocated for each job. This number is a proxy of the number of CPU cores the job requested<jupyter_code>fig = bkh.figure(
    title = f'DISTRIBUTION OF JOB SLOTS {user_filter_description(users, groups)}',
    x_axis_label = 'number of slots',
    y_axis_label = 'job count'
)
style_figure(fig)
color, alpha = 'lightseagreen', 0.7
hist, edges = np.histogram(df['slots'], density=False, bins=10)
fig.quad(top=hist, bottom=0, left=edges[:-1], right=edges[1:], alpha=alpha, fill_color=color, line_color=color)
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('jobs', '@top{0,000}'),
        ('low',  '@left{0.0} slots'),
        ('high', '@right{0.0} slots'),
    ],
    mode = 'mouse',
))
bkh.show(fig)<jupyter_output><empty_output><jupyter_text>Memory utilizationIn this section we plot the distribution of the usage of RAM by the jobs.<jupyter_code># Only consider successful jobs executed in single- and multi-core queues
queues_to_include = batch_queues['single-core'] + batch_queues['multi-core']
successful_jobs = df.loc[(df['exit_status'] == 0) & (df['qname'].isin(queues_to_include))]

# Create the figure
fig = bkh.figure(
    title = f'DISTRIBUTION OF MEMORY UTILIZATION PER BATCH QUEUE {user_filter_description(users, groups)}',
    x_axis_label = 'gigabyte',
    y_axis_label = 'job count'
)
style_figure(fig)

# Build a histogram per queue
colors = ('dodgerblue', 'darkmagenta', 'mediumseagreen', 'orange', 'crimson', 'sienna')
queues = successful_jobs.groupby(['qname']).size().sort_values(ascending=False)
for q, color in zip(queues.index, colors):
    # Select the jobs in this queue which wallclock time was higher than the 5th percentile,
    # to exclude jobs which may have terminated early
    job_set = successful_jobs.loc[successful_jobs['qname'] == q]
    wallclock_threshold = np.percentile(job_set['wallclock'], [5])[0]
    memory = job_set.loc[job_set['wallclock'] >= wallclock_threshold, 'x_mem_max_rss_gb']
    hist, edges = np.histogram(memory, density=False, bins=20)
    fig.quad(top=hist, bottom=0, left=edges[:-1], right=edges[1:], alpha= 0.5, fill_color=color, line_color=color, legend=q)

# Finalize the figure
fig.legend.click_policy = 'hide'
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('jobs', '@top{0,000}'),
        ('low',  '@left{0.0} GB'),
        ('high', '@right{0.0} GB'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
print_md(f"""*The figure above shows the distribution of RAM the jobs actually used in execution (in gigabyte), as collected by GridEngine. Click on the legend to hide / show information.*""")
def format_mem_pct(v):
    return '<1' if v < 1.0 else f'{int(v)}'

queues = df.groupby(['qname']).size().sort_values(ascending=False)
d = {}
for q in queues.index:
    selection = df[df['qname'] == q]
    d[q] = {
        'job_count': selection.shape[0],
        'pcts': np.percentile(selection['x_mem_max_rss_gb'], [50, 80, 98, 100])
    }

table = """
### Memory usage percentiles (gigabytes)

| Queue | Job Count | 50pct | 80pct | 98pct | 100pct |
| ----: | --------: | ----: | ----: | ----: | -----: |
"""

for k, v in d.items():
    table += f'| {k} | {format_job_count(v["job_count"])} | {format_mem_pct(v["pcts"][0])} | {format_mem_pct(v["pcts"][1])} | {format_mem_pct(v["pcts"][2])} | {format_mem_pct(v["pcts"][3])} | \n'

table += """
*The table above shows some percentiles of memory consumption **per job** for each execution queue (unit is GB).*
"""    
print_md(table)<jupyter_output><empty_output><jupyter_text>Estimation of required number of CPU coresIn this section we estimate the number of CPU cores that would be needed to perform the same activity over the considered period. We consider the most powerful CPU available in the farm.<jupyter_code># Get the value of the maximum CPU time conversion factor
max_might = df['x_worker_might_sectohs06sec'].max()

# Get the model name of a CPU with the max conversion factor (i.e. the most powerful CPU model used for the set of jobs considered)
cpu_model = df.loc[df['x_worker_might_sectohs06sec'] == max_might, 'x_worker_model_type'].unique()[0]

# Get the total amount of normalized CPU time consumed by all the relevant jobs
total_consumed_cputime_hs06sec = df['x_cputime_hs06sec'].sum()

# Compute the number of elapsed (wallclock) seconds in the period
start_period = df['submission_time'].min()
end_period = df['end_time'].max()
elapsed_days = (end_period - start_period).days
elapsed_secs = (end_period - start_period).total_seconds()

# The number of CPU cores (of the most powerful model we have) to perform the same activity
# over the same period of time is computed by dividing the normalized CPU time consumed by
# all the relevant jobs by the scale factor for the most powerful CPU model and dividing
# this by the number of elapsed seconds in the period.
# We assume a CPU efficiency < 1.0 to be more realistic
estimated_cpu_efficiency = 0.6
required_cpu_cores = total_consumed_cputime_hs06sec / (max_might * estimated_cpu_efficiency) / elapsed_secs
print_md(f'The number of cores of **{cpu_model}** required to deliver the same computing capacity over {elapsed_days} days is **{int(required_cpu_cores)}**')<jupyter_output><empty_output><jupyter_text>DESC DC2 Run1.2p jobs In this section we focus on the jobs for processing the DESC DC2 Run1.2p. All the jobs of that exercise were executed by used  `descprod` and the last iteration started by 2018-10-01.  Select the jobs of interest<jupyter_code># Select the period of interest
start_period = datetime.datetime(2018, 10, 1)
end_period = datetime.datetime(2018, 12, 31)

# DESC jobs run as user 'descprod'
dc2_jobs = filter_by_date(df, start=start_period, end=end_period)
dc2_jobs = dc2_jobs[dc2_jobs['owner'] == 'descprod']

# Add a label and a period for annotating the plots
dc2_label = 'DESC DC2 Run1.2p'
dc2_period = f'{start_period.strftime("%Y-%m-%d")} to {end_period.strftime("%Y-%m-%d")}'<jupyter_output><empty_output><jupyter_text>Label jobs according to their role in the pipelineFirst, we label the jobs according to their purpose, using the submission command which contains some clues on the type of job. Note that there are some jobs which contain the type of the job in the `account` field, but this was implemented late in the process, so not all jobs were labeled this way.We add a column which contains the kind of job, named `_job_category`.<jupyter_code>dc2_jobs['_job_category'] = ''

patterns = {
    'ingest_flat':             re.compile(r'.*/ingest_flat/.*'),
    'ingest_bias':             re.compile(r'.*/ingest_bias/.*'),
    'ingest_dark':             re.compile(r'.*/ingest_dark/.*'),
    'ingestRefCat':            re.compile(r'.*/ingestRefCat/.*'),
    'ingestData':              re.compile(r'.*/ingestData/.*'),
    'setup_ingest':            re.compile(r'.*/setup_ingest/.*'),
    'run_ingestData':          re.compile(r'.*/run_ingestData/.*'),

    'setup_calib':             re.compile(r'.*/setup_calib/.*'),
    'run_bias':                re.compile(r'.*/run_bias/.*'),
    'run_flat':                re.compile(r'.*/run_flat/.*'),
    'run_dark':                re.compile(r'.*/run_dark/.*'),
    'setup_bias':              re.compile(r'.*/setup_bias/.*'),
    'setup_flat':              re.compile(r'.*/setup_flat/.*'),
    'setup_dark':              re.compile(r'.*/setup_dark/.*'),
    'finish_calib':            re.compile(r'.*/finish_calib/.*'),

    'setup_SingleFrameDriver': re.compile(r'.*/setup_SingleFrameDriver/.*'),
    'run_SingleFrameDriver':   re.compile(r'.*/run_SingleFrameDriver/.*'),
    'setup_coadd':             re.compile(r'.*/setup_coadd/.*'),
    'run_coaddDriver':         re.compile(r'.*/run_coaddDriver/.*'),
    'setup_multiBand':         re.compile(r'.*/setup_multiBand/.*'),
    'run_multiBandDriver':     re.compile(r'.*/run_multiBandDriver/.*'),
    'makeSkyMap':              re.compile(r'.*/makeSkyMap/.*'),
}

def get_job_type(submit_cmd):
    for cat, rex in patterns.items():
        if rex.search(submit_cmd):
            return cat
    return ''

dc2_jobs['_job_category'] = dc2_jobs['submit_cmd'].apply(lambda cmd: get_job_type(cmd))
unlabeled = dc2_jobs[dc2_jobs['_job_category'] == ''].shape[0]
print_md(f'There are {unlabeled} unlabeled jobs, out of {dc2_jobs.shape[0]} DC2 jobs')<jupyter_output><empty_output><jupyter_text>Number of jobs per category<jupyter_code>job_categories = dc2_jobs.groupby('_job_category').size().sort_values(ascending=False)
labels = [n for n in job_categories.index]
values = job_categories.get_values()
fig = bkh.figure(
    title = f'{dc2_label} — NUMBER OF JOBS PER CATEGORY [{dc2_period}]',
    x_axis_label = 'job category',
    y_axis_label = 'job count',
    y_axis_type = 'log',
    x_range = labels
)
style_figure(fig)
color, alpha = 'orange', 0.8
fig.vbar(x=labels, top=values, bottom=0.1, width=0.6, alpha=alpha, fill_color=color, line_color=color)
fig.xgrid.grid_line_color = None
fig.xaxis.major_label_orientation = 0.9
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('job type', '@x'),
        ('jobs', '@top{0,000}'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
print_md(f"""*The figure above shows the number of DC2 jobs per job category.
Hover the mouse over each vertical bar to get the exact number of jobs*""")<jupyter_output><empty_output><jupyter_text>CPU utilization per job category<jupyter_code>def get_most_powerful_CPU(df):
    """Returns a tuple with the conversion factor of the most powerful CPU and its CPU model
    """
    # Get the value of the maximum CPU time conversion factor
    max_might = df['x_worker_might_sectohs06sec'].max()
    # Get the model name of the CPU with the max conversion factor
    cpu_model = df.loc[df['x_worker_might_sectohs06sec'] == max_might, 'x_worker_model_type'].unique()[0]
    return max_might, cpu_model
    
# Get the value of the maximum CPU time conversion factor and the CPU model
(max_might, cpu_model) = get_most_powerful_CPU(df)

# Get the aggregated CPU time per job category and normalize it to the most powerful CPU used
total_cpu = dc2_jobs['x_cputime_hs06sec'].sum() / (max_might * 3600)
cpu_time_per_category = dc2_jobs.groupby('_job_category')['x_cputime_hs06sec'].sum() / (max_might * 3600)
cpu_time_per_category = cpu_time_per_category.sort_values(ascending=False)
cpu_time_per_category = cpu_time_per_category[cpu_time_per_category >= 1.0]
data = bkhmodels.ColumnDataSource({
    'categories':              cpu_time_per_category.index.values,
    'cpu_time':                cpu_time_per_category.get_values(),
    'percent_total_cpu_time':  [v / total_cpu for v in cpu_time_per_category.get_values()],
})
fig = bkh.figure(
    title = f'{dc2_label} — DISTRIBUTION OF CPU UTILIZATION [{dc2_period}]',
    x_axis_label = 'job category',
    y_axis_label = f'Hours on {cpu_model}',
    y_axis_type = 'log',
    x_range = labels
)
style_figure(fig)
color, alpha = 'dodgerblue', 0.8
fig.vbar(x='categories', top='cpu_time', source=data, width=0.6, bottom=1, alpha=alpha, fill_color=color, line_color=color)
fig.xgrid.grid_line_color = None
fig.xaxis.major_label_orientation = 0.9
fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,0")
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('job type', '@categories'),
        ('CPU time', '@cpu_time{0,0.000}'),
        ('% of total CPU time', '@percent_total_cpu_time{%0.0}'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
print_md(f"""*The figure above shows the aggregated CPU time used by DC2 jobs per job category.
Only the job categories for which the CPU consumption is at least 1 hour are shown.
Hover the mouse over each vertical bar to get the CPU hours consumed by each category*""")<jupyter_output><empty_output><jupyter_text>CPU and memory utilization per job category<jupyter_code># Get the value of the maximum CPU time conversion factor and the CPU model
(max_might, cpu_model) = get_most_powerful_CPU(df)

# Only consider jobs with exit_status < 128 belonging to the selected categories. An exit status >= 128 means that the job
# was interrupted by GridEngine, perhaps because it has exceeded its CPU or memory allocation
categories = ('run_multiBandDriver', 'run_SingleFrameDriver', 'run_coaddDriver')
job_set = dc2_jobs.loc[(dc2_jobs['exit_status'] < 128) & (dc2_jobs['_job_category'].isin(categories))]
max_cpu_time = job_set['x_cputime_hs06sec'].max()/(max_might * 3600)

fig = bkh.figure(
    title = f'{dc2_label} — CPU & RAM UTILIZATION PER JOB CATEGORY [{dc2_period}]',
    x_axis_label = 'RAM (gigabyte)',
    y_axis_label = f'Hours on {cpu_model}',
)
style_figure(fig)

colors = ('dodgerblue', 'darkmagenta', 'mediumseagreen', 'orange', 'crimson', 'sienna')
for cat, color in zip(categories, colors):
    jobs_in_category = job_set.loc[job_set['_job_category'] == cat]
    if jobs_in_category.shape[0] > 0:
        data = bkhmodels.ColumnDataSource({
            'memory': jobs_in_category['x_mem_max_rss_gb'],
            'cpu':    jobs_in_category['x_cputime_hs06sec']/(max_might * 3600),
            'jobid':  jobs_in_category['job_number'],
            'queue':  jobs_in_category['qname'],
            'slots':  jobs_in_category['slots'],
        })
        fig.scatter(x='memory', y='cpu', source=data, fill_color=color, fill_alpha=0.5, line_color=color, legend=cat)

fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format="0,000")
fig.y_range.end = max_cpu_time*1.3 # make sure the legend does not hide data points
fig.legend.location = "top_left"
fig.legend.click_policy = 'hide'
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('JobID', '@jobid'),
        ('CPU',   '@cpu{0,0.00}'),
        ('RAM',   '@memory{0.0}'),
        ('Queue', '@queue'),
        ('Slots', '@slots'),
    ],
    mode = 'mouse',
))
bkh.show(fig)
print_md(f"""*The figure above shows the normalized CPU time used vs RAM used by DC2 jobs per job category.
Each dot represents one job. The CPU time is normalized to the concrete given CPU model and is aggregated over all the CPU slots requested by the job at submission time. Click on the legend to hide/show any category.
Hover the mouse over each dot to retrieve some details about that job.*""")<jupyter_output>1066.7641987518355<jupyter_text>Failed jobs per category<jupyter_code># Get the value of the maximum CPU time conversion factor and the CPU model
(max_might, cpu_model) = get_most_powerful_CPU(df)

# Only consider jobs with exit_status > 0
categories = ('run_multiBandDriver', 'run_SingleFrameDriver', 'run_coaddDriver')
failed_jobs = dc2_jobs.loc[(dc2_jobs['exit_status'] > 0) & (dc2_jobs['_job_category'].isin(categories))]
max_cpu_time = failed_jobs['x_cputime_hs06sec'].max()/(max_might * 3600)

fig = bkh.figure(
    title = f'{dc2_label} — FAILED JOBS PER JOB CATEGORY [{dc2_period}]',
    x_axis_label = 'RAM (gigabyte)',
    y_axis_label = f'Hours on {cpu_model}',
)
style_figure(fig)

colors = ('dodgerblue', 'darkmagenta', 'mediumseagreen', 'orange', 'crimson', 'sienna')
for cat, color in zip(categories, colors):
    jobs_in_category = failed_jobs.loc[failed_jobs['_job_category'] == cat]
    if jobs_in_category.shape[0] > 0:
        data = bkhmodels.ColumnDataSource({
            'memory':  jobs_in_category['x_mem_max_rss_gb'],
            'cpu':     jobs_in_category['x_cputime_hs06sec']/(max_might * 3600),
            'jobid':   jobs_in_category['job_number'],
            'queue':   jobs_in_category['qname'],
            'slots':   jobs_in_category['slots'],
            'status':  jobs_in_category['exit_status'],
        })
        fig.scatter(x='memory', y='cpu', source=data, fill_color=color, fill_alpha=0.5, line_color=color, legend=cat)

fig.yaxis.formatter = bkhmodels.formatters.NumeralTickFormatter(format='0,000')
fig.legend.location = 'top_left'
fig.legend.click_policy = 'hide'
fig.y_range.end = max_cpu_time*1.3 # make sure the legend does not hide data points
fig.add_tools(bkhmodels.HoverTool(
    tooltips = [
        ('JobID',  '@jobid'),
        ('CPU',    '@cpu{0,0.00}'),
        ('RAM',    '@memory{0.0}'),
        ('Queue',  '@queue'),
        ('Slots',  '@slots'),
        ('Status', '@status'),
    ],
    mode = 'mouse',
))

bkh.show(fig)
print_md(f"""*The figure above shows the normalized CPU time used vs RAM used by a total of **{failed_jobs.shape[0]} failed DC2 jobs**, presented per job category.
Each dot represents one failed job, that is, one which exit status is not zero. The CPU time is normalized to the concrete given CPU model and is aggregated over all the CPU slots requested by the job at submission time. Click on the legend to hide/show any category.
Hover the mouse over each dot to retrieve some details about that job, including its exit status.*""")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13981574846457054, 'fraction_numerical': 0.01576679805665047, 'mean_word_length': 4.062296983758701, 'pattern_counts': {'":': 0, '<': 114, '<?xml version=': 0, '>': 116, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '26', 'dir': 'jupyter-structured-clean-dedup', 'id': '152597', 'license': 'Apache-2.0', 'n_tokens_mistral': 14839, 'n_tokens_neox': 14255, 'n_words': 4051, 'path': 'ALBA.ipynb', 'repo_name': 'airnandez/alba', 'size': '43645'}