src/main/java/org/apache/accumulo/testing/continuous/TimeBinner.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.accumulo.testing.continuous;

import static java.nio.charset.StandardCharsets.UTF_8;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.accumulo.testing.cli.ClientOpts.TimeConverter;
import org.apache.accumulo.testing.cli.Help;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.beust.jcommander.Parameter;

public class TimeBinner {
  private static final Logger log = LoggerFactory.getLogger(TimeBinner.class);

  enum Operation {
    AVG,
    SUM,
    MIN,
    MAX,
    COUNT,
    CUMULATIVE,
    AMM, // avg,min,max
    AMM_HACK1 // special case
  }

  private static class DoubleWrapper {
    double d;
  }

  private static DoubleWrapper get(long l, HashMap<Long,DoubleWrapper> m, double init) {
    DoubleWrapper dw = m.get(l);
    if (dw == null) {
      dw = new DoubleWrapper();
      dw.d = init;
      m.put(l, dw);
    }
    return dw;
  }

  static class Opts extends Help {
    @Parameter(names = "--period", description = "period", converter = TimeConverter.class,
        required = true)
    long period = 0;
    @Parameter(names = "--timeColumn", description = "time column", required = true)
    int timeColumn = 0;
    @Parameter(names = "--dataColumn", description = "data column", required = true)
    int dataColumn = 0;
    @Parameter(names = "--operation", description = "one of: AVG, SUM, MIN, MAX, COUNT",
        required = true)
    String operation;
    @Parameter(names = "--dateFormat",
        description = "a SimpleDataFormat string that describes the data format")
    String dateFormat = "MM/dd/yy-HH:mm:ss";
  }

  public static void main(String[] args) throws Exception {
    Opts opts = new Opts();
    opts.parseArgs(TimeBinner.class.getName(), args);

    Operation operation = Operation.valueOf(opts.operation);
    SimpleDateFormat sdf = new SimpleDateFormat(opts.dateFormat);

    BufferedReader in = new BufferedReader(new InputStreamReader(System.in, UTF_8));

    String line;

    HashMap<Long,DoubleWrapper> aggregation1 = new HashMap<>();
    HashMap<Long,DoubleWrapper> aggregation2 = new HashMap<>();
    HashMap<Long,DoubleWrapper> aggregation3 = new HashMap<>();
    HashMap<Long,DoubleWrapper> aggregation4 = new HashMap<>();

    while ((line = in.readLine()) != null) {

      try {
        String[] tokens = line.split("\\s+");

        long time = (long) Double.parseDouble(tokens[opts.timeColumn]);
        double data = Double.parseDouble(tokens[opts.dataColumn]);

        time = (time / opts.period) * opts.period;

        switch (operation) {
          case AMM_HACK1: {
            if (opts.dataColumn < 2) {
              throw new IllegalArgumentException("--dataColumn must be at least 2");
            }
            double data_min = Double.parseDouble(tokens[opts.dataColumn - 2]);
            double data_max = Double.parseDouble(tokens[opts.dataColumn - 1]);

            updateMin(time, aggregation3, data, data_min);
            updateMax(time, aggregation4, data, data_max);

            increment(time, aggregation1, data);
            increment(time, aggregation2, 1);
            break;
          }
          case AMM: {
            updateMin(time, aggregation3, data, data);
            updateMax(time, aggregation4, data, data);

            increment(time, aggregation1, data);
            increment(time, aggregation2, 1);
            break;
          }
          case AVG: {
            increment(time, aggregation1, data);
            increment(time, aggregation2, 1);
            break;
          }
          case MAX: {
            updateMax(time, aggregation1, data, data);
            break;
          }
          case MIN: {
            updateMin(time, aggregation1, data, data);
            break;
          }
          case COUNT: {
            increment(time, aggregation1, 1);
            break;
          }
          case SUM:
          case CUMULATIVE: {
            increment(time, aggregation1, data);
            break;
          }
        }

      } catch (Exception e) {
        log.error("Failed to process line: {} {}", line, e.getMessage());
      }
    }

    AtomicReference<Double> cumulative = new AtomicReference<>((double) 0);
    aggregation1.entrySet().stream().sorted().forEach(entry -> {
      final String value;
      final var currentKey = entry.getKey();
      final var currentValue = entry.getValue();

      switch (operation) {
        case AMM_HACK1:
        case AMM: {
          DoubleWrapper countdw = aggregation2.get(currentKey);
          value = "" + (currentValue.d / countdw.d) + " " + aggregation3.get(currentKey).d + " "
              + aggregation4.get(currentKey).d;
          break;
        }
        case AVG: {
          DoubleWrapper countdw = aggregation2.get(currentKey);
          value = "" + (currentValue.d / countdw.d);
          break;
        }
        case CUMULATIVE: {
          cumulative.updateAndGet(v -> v + currentValue.d);
          value = "" + cumulative;
          break;
        }
        default:
          value = "" + currentValue.d;
      }

      log.info("{} {}", sdf.format(new Date(currentKey)), value);
    });
  }

  private static void increment(long time, HashMap<Long,DoubleWrapper> aggregation, double amount) {
    get(time, aggregation, 0).d += amount;
  }

  private static void updateMax(long time, HashMap<Long,DoubleWrapper> aggregation, double data,
      double new_max) {
    DoubleWrapper maxdw = get(time, aggregation, Double.NEGATIVE_INFINITY);
    if (data > maxdw.d)
      maxdw.d = new_max;
  }

  private static void updateMin(long time, HashMap<Long,DoubleWrapper> aggregation, double data,
      double new_min) {
    DoubleWrapper mindw = get(time, aggregation, Double.POSITIVE_INFINITY);
    if (data < mindw.d)
      mindw.d = new_min;
  }
}