Why not adopt me?

This distribution is up for adoption! If you're interested then please contact the PAUSE module admins via email.

NAME

Stream::Aggregate - generate aggregate information from a stream of data

SYNOPSIS

 use Stream::Aggregate;

 my $af = generate_aggregation_func(
	$agg_config, 
	$extra_parameters, 
	$user_extra_parameters);

 while ($log = ???) {
	@stats = $af->($log);
 }
 @stats = $af->(undef);

DESCRIPTION

Stream::Aggregate is a general-purpose aggregation module that will aggregate from a stream of perl objects. While it was written specifically for log processing, it can be used for other things too.

Aggregation has two key elements: how you group things and what you aggregate. This module understands two different ways to group things: nested and cross-product.

Nested groupings come from processing sorted input: if you have three fields you are considering your context, the order in which the data is sorted must match the order in which these fields make up your context. If you want to count things by URL, then you must sort your input by URL.

Cross-product groupings come from processing unsorted input. Each combination of values of the fields that make up your context is another context. This can lead to memory exhaustion so you must specify the maximum number of values for each of the fields.

Nested groupings

Nested groups are most easily illustrated with a simple example: aggregating by year, month, and day. The input data must be sorted by year, month, and day. The current context is defined by the triplet: (year, month, day). That triplet must be returned by the context code. It is stored in the @current_context array. When a context is finished, it must be converted into a hash by context2columns.

Doing it this way, you can, for example, get the average of some data item per day, per month, and per year in one pass though your data.

Cross-Product grouping

Cross Product grouping does not depend on the sort order of the input and can have many contexts active at the same time.

For example, if you're aggregating sales figures for shoes and want statistics for the combinations of size, width, and color there isn't a sort or nesting order that will answer your questions.

Use crossproduct to limit yourself to a certain number of values for each variable (say 10 sizes, 3 widths, and 5 colors).

API

The configuration for Stream::Aggregate is compiled into a perl function which is then called once for each input object. Each time it is called, it may produce one or more aggregate objects as output. When there is no more input data, call the aggregation function with undef as its input.

The generate-the-function routine, generate_aggregation_func takes three parameters. The first is the configuration object (defined below). The configuration object is expected (but not required) to come from a YAML file. All examples are in YAML format. The second and third arguments provide extra information. Currently they are only used to get a description of what this aggregation is trying to do using the name field. Eg:

generate_aggregation_func($agg_config, $extra, $user_extra);

my $code = qq{#line 1 "FAKE-all-code-for-$extra->{name}"\n};

The configuration object for Stream::Aggregate is expected to be read from a YAML file but it does not have to be created that way.

For some of the code fields (below), marked as Closure/Config, you can provide a closure instead of code. To do that, have a BEGIN block assign a value (the closure) to the variable $coderef. If you do this, code outside the BEGIN block will only be compiled but will never be run. When evalutating the BEGIN block, the variable $agg_config will be set to the value of key_config (assuming the field was key).

The behavior of generate_aggregation_func in array context may change in the future to provide additional return values.

As the aggregator runs over the input, it needs to know the boundries of the contexts so that it knows when to generate an aggregation result record.

For example, if you were aggretgating information about URL with nested groupings, you need to sort your input by URL and you need to define a context that returns the URL (in YAML format, with $log as your input variable):

context: |
  return ($log->{url})

If you want to aggregate over both the URL and the web host, the context must return an array: host & URL (in YAML format):

context: |
  $log->{url} =~ m{(?:https?|ftp)://([^/:]+)}
  my $host = $1;
  return ($host, $log->{url})

When the context is has multiple levels like that, there will be a resultant aggregation record for each value at each level.

context

Code, Optional. Given a $log entry, return an array that describes the aggregation context. For example, for a URL, this array might be: domain name; host name (if different from domain name); each component of the path of the URL except the final filename. As Aggregate runs, it will generate an aggregation record for each element of the array.

This code will be invoked on every input record.

This is not required for cross-product aggregations but is required for nested aggregations.

context2columns

Code, Optional. Given a context, in @current_context, return additional key/value pairs for the resulting aggregation record. This is how the context gets described in the aggregation results records.

This code will be invoked to generate resultant values just before a context is closed.

If this code sets the variable $suppress_result, then this aggregation result will be discarded.

stringify_context

Code, Optional.

If the new context array returned by the context code (soon to become @current_context) is not an array of strings but rather an array of references, it will be turned into strings using YAML. These strings may not matter because you control how the context manifests in the result records with context2columns.

If this isn't what you want, use stringify_context to do something different. Unlike most of the other functions, stringify_context operates on @_.

This will be invoked for every input record.

crossproduct

Hash, Name->Number, Optional.

For crossproduct groupings, this defines the dimensions. The keys are the variables. The values are the maximum number of values for each variable to track.

The keys must be ephemeral0, ephemeral, or ephemeral2 column names.

combinations

Hash, Name->Code, Optional.

If you have crossproduct groupings, do you also want to synthesize contexts that exclude some or all of the crossproduct dimensions?

For each dimension, provide code that that answers the question: if you remove this dimension from the crossproduct, should this new context be considered? This code can look at $row to see what other dimensions are active for this potential context. To keep this context, the code must evaluate to a true value.

Combinations are evaluated in alphabetical order. If there is more than one path to a combination, only the first path will be considered. For example, if you have three crossproduct dimensions, a, b and c then the combinations are (in the order the'll be considered):

b and c, excluding a
a and c, excluding b
a and b, excluding c: These have no dependency and will be produced if the combinations code returns a true value.
c, excluding a and b: This possibility will only be explored coming from b and c. If the combination rule for a rejected the combination, then the c-only permuation will never be reached.
b, excluding a and c: This possibility will only be explored coming from b and c. If the combination rule for a rejected the combination, then the b-only permuation will never be reached.
a, excluding b and c: This possibility will only be explored coming from a and c. If the combination rule for b rejected the combination, then the a-only permuation will never be reached.
excluding a, b and c: This possibility will only be explored coming from c. If the combination rule for a or c rejected the combination, then the empty permuation will never be reached.

Using combinations can greatly expand your memory requirements. More than four dimensions of combinations is probably a bad idea.

If you want skip a dimension entirely, leave it out of the combinations key rather than adding key that evaluates to false. A key the evaluates to false will still have to be tested many times whereas a missing key won't have any code generated for it.

simplify

Hash, Name->Code, Optional.

When a cross-product key is exceeding its quota of values, the default replacement value is *. This hash allows you to override the code that chooses the new value.

finalize_result

Code, Optional, Closure/Config.

This code will be called after the resultant values for a context have been calculated. It is a last-chance to modify them or to suppress the results. The values can be found as a reference to a hash: $row. To suppress the results, set $suppress_results.

new_context

Code, Optional, Closure/Config.

This code will be called each time there is a new context. At the time it is called, $ps is a reference to the new context, but @current_context will not yet have been updated to the new value.

merge

Code, Optional, Closure/Config.

When using multiple levels of contexts, data is counted for the top-most context layer only. When that top-most layer finishes, the counts are merged into the next more-general layer.

During the merge there is both $ps and $oldps available to for code to reference. The default merge handles all of the pre-defined column types. If you are using $ps->{heap} storage for context data, you need to merge that data from $oldps to $ps yourself.

CONTROL FLOW CONFIGURATION

filter

Code, Optional. Before any of the columns are calculated or any of the values saved, run this filter code. If it returns a true value then proceed as normal. If it returns a false value, then do not count any of the values. The filter code can remember things in $ps-{heap}> that might effect how other things are counted.

In some situations, you many want to throw away most data and count things in the filter. When doing that, it may be that all of the columns come from output.

This may be redesigned in a future release in a way that is not backwards compatible.

filter_early

Boolean, Optional, default false. Check the filter early before figuring out contexts? If so, and the result is filtered, don't check to see if the context changed.

passthrough

Code, Optional. Add results to the output of the aggregation. A value of $log (assuming that's your input record variable) adds the input data stream to the output stream. The passthrough code is run before the input line is counted.

PARAMETERS CONFIGURATION

max_stats_to_keep

Number, Optional, default: 4000.

When aggregating large amounts of data, limit memory use by throwing away some of the data. When data is thrown away, keep this number of samples for statistics functions that need bulk data like standard_deviation.

reduce

Code, Optional, Closure/Config.

When max_stats_to_keep is exceeded, data will be thrown away. This function will be called when that has happened.

preprocess

Code, Optional. Code to preprocess the input $log objects.

item_name

String, Optional, default: $log. The default name for the input values to aggregate over is $log. If this name is not appropriate, you can use item_name to change the variable name of the input values to something else. Include the dollar sign ($) in the name.

debug

Boolean, Optional. Print out some debugging information, including the code that is generated for building the columns.

strict

Boolean, Optional, default: false. Enforce strict and warnings for user code.

AGGREGATE OUTPUT COLUMNS CONFIGURATION

Each of these (except ephemeral & keep) defines additional columns of output that will be included in each aggregation record. These are all optional and all are defined as key/value pairs where the keys are column names and the values are perl code. You can refer to other columns using the variable $column_column_name where column_name is the name of one of the other columns. When refering to other columns, the order in which columns are processed matters: ephemeral and keep are processed first and second respecively. Idential code fragments will be evaluated only once. Within a group, columns are evaluated alphabetically.

Some of the columns will have their code evaluated per-item and some are evaluated per-aggregation.

The input data is in $log unless overriden by item_name.

Per item callbacks

ephemeral

These columns will not be included in the aggregation data. Refer to them as $column_column_name. If you are using ephemeral to declare the column but do not want to assign it a value, set the value for the ephemeral code to be undef. In YAML, thats ~:

ephemeral:
  var1: ~
  var2: ~

ephemeral0

Same as ephemeral, will be evaluated before ephemeral.

ephemeral2

Same as ephemeral, will be evaluated after ephemeral.

counter

Keep a counter. Add one if the code returns true.

percentage

Keep a counter. Include the percentage of items for which the code returned true as an output column as opposed to the number of items where the code returned 0. A return value of undef does not get counted at all.

sum

Keep an accumulator. Add the return values.

mean

Keep an accumulator. Add the return values. Divide by the number of items before inserting into the results. Items whose value is undef do not count towards the number of items or the sum.

standard_deviation

Remeber the return values. Compute the standard deviation of the accumulated return values and insert that into the results. Items whose value is undef are removed before calculating the standard_deviation.

median

Remeber the return values. Compute the median of the accumulated return values and insert that into the results. Items whose value is undef are removed before calculating the median.

dominant

Remeber the return values. Compute the mode (most frequent) of the accumulated return values and insert that into the results. Items whose value is undef are removed before calculating the mode.

min

Keep a minimum numeric value. Replace it with the return value if the return value is less than the current value. Items whose value is undef are removed before calculating the min.

max

Keep a maximum numeric value. Replace it with the return value if the return value is greater than the current value. Items whose value is undef are removed before calculating the max.

minstr

Keep a minimum string value. Replace it with the return value if the return value is less than the current value. Items whose value is undef are removed before calculating the minstr.

maxstr

Keep a maximum string value. Replace it with the return value if the return value is greater than the current value. Items whose value is undef are removed before calculating the maxstr.

keep

Remember the return values. The return values are available at aggregation time as @{$ps->{keep}{column_name}}. Items whose value is undef are kept but they're ignored by Stream::Aggregate::Stats functions.

Per aggregation result record callbacks

For code that is per-aggregation, the saved aggregation state can be found in $ps. One item that is probably needed is $ps->{item_count}.

output: Extra columns to include in the output. This is where to save $ps->{item_count}.
stat: Use arbitrary perl code to compute statistics on remembered return values kept with keep. Write your own function or use any of the functions in Stream::Aggregate::Stats (the global variable is pre-loaded). No, there isn't any difference between this and output.

VARIALBES AVAILABLE FOR CODE SNIPPETS TO USE

The following variables are available for the code that generates per-item and per-aggregation statistics:

$log

The current input item (unless overridden by item_name)

$ps->{keep}{column_name}

An array of return values kept by keep.

$ps->{numeric}{column_name}

If Stream::Aggregate::Stats functions are called, they will grab the numeric values from $ps->{keep}{column_name} and store them in $ps->{numeric}{column_name}.

$ps->{random}

For each kept item in $ps->{keep}{column_name}, there is a corrosponding item in $ps->{random} that is a random number. These random numbers are used to determine which values to keep and which values to toss if there are too many values to keep them all.

$ps->{$column_type}{column_name}

For each type of column (output, counter, percentage, sum, min, standard_deviation, median, stat) the values that will be part of the final aggregation record.

$ps->{$tempoary_type}{column_name}

Some columns need temporary storage for their values: percentage_counter (the counter used by percentage); percentage_total (the number of total items); mean_sum (the sum used to compute the mean); mean_count (the number of items for the mean).

$ps->{heap}

A hash that can be used by the configured perl code for whatever it wants.

$ps->{item_counter}

The count of items.

$agg_config

The configuration object for Stream::Aggregate

$itemref

A reference to $log. It's always $itemref even if $log is something else.

@current_context

The current context as returned by context.

@context_strings

The string-ified version @current_context as returned by stringify_context or YAML.

@contexts

The array of context objects. $ps is always $context[-1].

@items_seen

An array that counts the number of rows of output from this aggregation. When the context is multi-level, the counter is multi-level. For example, if the context is domain, host, and URL; then $items_seen[0] is the number of domains (so far), and $items_seen[1] is the number of hosts for this domain (so far), and $items_seen[2] is the number of URLs for this host (so far).

Passthrough rows do not count.

XXX what about cross-product aggregations?

$row

When gathering results, the variable that holds them is a reference to a hash: $row.

$suppress_result

After gathering results, the $suppress_result variable is examined. If it's set the results (in $row) are discards.

To skip results that aren't crossproduct results, in finalize_result, set $suppress_result if $cross_count isn't true.

$cross_count

The number of currently active crossproduct accumulator contexts.

$extra, $user_extra

The additional paramerts (beyond $agg_config) that were passed to generate_aggregation_func().

%persist

This hash is not used by Stream::Aggregate. It's available for any supplied code to use however it wants.

$last_item

A refernece to the previous $log object. This is valid during finalize_result and context2columns.

There are more. Read the code.

HELPER FUNCTIONS

The following helper functions are available: everything in Stream::Aggregate::Stats and:

nonblank($value): Returns $value if $value is defined and not the empty string. Returns undef otherwise.

EXAMPLE1

This example will look at a set of records regarding health risks. Each record represents a person:

name<TAB>birthday<TAB>sex<TAB>number of hospital visits in the last year

We will generate the following aggregation records:

Average age of entire sample
Median number of hospital visits in the last year for the entire sample.
Median number of hospital visits in the last year by sex.
Median number of hospital visits in the last year by age, with a minimum sample size of five.
Median number of hospital vists in the last year by age and sex with a minimum sample size of five.

The code:

  #!/usr/bin/perl 

  #
  # Our input data is the raw strings from the input file.  Most of the
  # work is parsing them and reformatting the data.
  #

  use strict;
  use warnings;
  use Stream::Aggregate;
  use YAML;

  my $aconfig = Load(<<'END_ACONFIG');
  strict:                 1
  debug:                  0
  item_name:              $record
  max_stats_to_keep:      500
  filter_early:           1
  filter: |
    # ignore black lines and comments
    return 0 if $record =~ /^#/;
    return 0 if $record =~ /^$/;
    return 1;
  crossproduct:
    sex:                  3
    age:                  150
  combinations:
    sex:                  1
    age:                  1
  ephemeral0:
    #
    # We are using ephemeral0 to declare the column variables
    #
    name:                 ~
    birthday:             ~
    gender:               ~
    number_of_visits:     ~
  ephemeral:
    # 
    # We are using a fake column ($column_step1) in ephemeral to initialize 
    # the raw column variables we declared in ephemeral0
    #
    step1: |
       chomp($record);
       ($column_name, $column_birthday, $column_gender, $column_number_of_visits) = split(/\t/, $record);
  ephemeral2:
    #
    # We are using ephemeral2 to generate the computed input data
    #
    age: |
      use Time::ParseDate qw(parsedate);
      my $t = parsedate($column_birthday, NO_RELATIVE => 1, DATE_REQUIRED => 1, WHOLE => 1, GMT => 1);
      return undef unless $t;
      return int ((parsedate('2011-05-01', GMT => 1) - $t) / (365.24 * 86400))
    sex: |
      return 'M' if $column_gender =~ /^m/i;
      return 'F' if $column_gender =~ /^f/i;
      return undef;
    hospital_visits: |
      $column_number_of_visits =~ /^(\d+)$/;
      $1
  output:
    sample_size:          $ps->{item_counter}
  median:
    avg_hospital_visits:  $column_hospital_visits
  mean:                   
    avg_age:              $column_age
  finalize_result: |
    #
    # Don't generate result records unless there are at
    # least five items being aggregated.
    #
    $suppress_result = 1 if $ps->{item_counter} < 5;
  END_ACONFIG

  my $ag = generate_aggregation_func($aconfig, { 
	  name	=> 'Aggregate Hospital Visits',
  });

  my @results;

  while (<>) {
	  for my $result ($ag->($_)) {
		  # do something with the result records
	  }
  }

  for my $result ($ag->(undef)) {
	  # do something with the result records
  }

EXAMPLE2

Our example will count the following things: number of unique URLs per domain, average length of the URL.

  #!/usr/bin/perl 

  use strict;
  use warnings;
  use Stream::Aggregate;
  use YAML;

  my $aconfig = Load(<<'END_ACONFIG');
  strict:                 1
  debug:                  0
  item_name:              $item
  context:                $item->{domain}
  context2columns:        return (domain => $current_context[0])
  ephemeral:
    # 
    # The only persistent unstructured place to store data from 
    # one row to the next is $ps->{heap}.   $ps->{heap} 
    # is per-context, but that's okay for our usage.
    #
    is_different: |
      my $old = $ps->{heap}{last_item};
      $ps->{heap}{last_item} = $item;
      return 1 unless $old;
      return 0 if $old->{url} eq $item->{url};
      return 1;
  sum:
    unique_urls:          $column_is_different
  mean:                   
    avg_url_length:       length($item->{url})
  finalize_result: |
    # we don't want the roll-up context of all domains
    $suppress_result = 1 unless $row->{domain};
  END_ACONFIG

  my $ag = generate_aggregation_func($aconfig, { 
	  name	=> 'Aggregate URL data'
  });

  while(<>) {
	  # we'll parse the input here
	  my $item;
	  chomp;
	  next if /^$/;
	  next if /^#/;
	  die "'$_'" unless m{^\w+:\/\/([^/]+)(?:/.*)?};
	  $item = {
		  domain	=> $1,
		  url	=> $_,
	  };
	  for my $result ($ag->($item)) {
		  # do something with the result records
	  }
  }

  for my $result ($ag->(undef)) {
	# do something with the results records
  }

LICENSE

This package may be used and redistributed under the terms of either the Artistic 2.0 or LGPL 2.1 license.

To install Stream::Aggregate, copy and paste the appropriate command in to your terminal.

cpanm

cpanm Stream::Aggregate

CPAN shell

perl -MCPAN -e shell
install Stream::Aggregate

For more information on module installation, please visit the detailed CPAN module installation guide.

	Global
`s`	Focus search bar
`?`	Bring up this help dialog

	GitHub
`g` `p`	Go to pull requests
`g` `i`	go to github issues (only if github is preferred repository)

	POD
`g` `a`	Go to author
`g` `c`	Go to changes
`g` `i`	Go to issues
`g` `d`	Go to dist
`g` `r`	Go to repository/SCM
`g` `s`	Go to source
`g` `b`	Go to file browse

	Search terms
module: (e.g. module:Plugin)
distribution: (e.g. distribution:Dancer auth)
author: (e.g. author:SONGMU Redis)
version: (e.g. version:1.00)

Why not adopt me?

NAME

SYNOPSIS

DESCRIPTION

Nested groupings

Cross-Product grouping

API

CONTEXT-RELATED CONFIGURATION PARAMETERS

CONTROL FLOW CONFIGURATION

PARAMETERS CONFIGURATION

AGGREGATE OUTPUT COLUMNS CONFIGURATION

Per item callbacks

Per aggregation result record callbacks

VARIALBES AVAILABLE FOR CODE SNIPPETS TO USE

HELPER FUNCTIONS

EXAMPLE1

EXAMPLE2

LICENSE

Module Install Instructions