core/src/main/java/org/apache/hadoop/chukwa/hicc/DatasetMapper.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
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.chukwa.hicc;


import java.util.TreeMap;
import java.util.HashMap;
import java.util.ArrayList;
import java.util.List;
import java.sql.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.chukwa.util.ExceptionUtil;

public class DatasetMapper {
  private String jdbc;
  private static Log log = LogFactory.getLog(DatasetMapper.class);
  private TreeMap<String, TreeMap<String, Double>> dataset;
  private List<String> labels;

  public DatasetMapper(String jdbc) {
    this.jdbc = jdbc;
    this.dataset = new TreeMap<String, TreeMap<String, Double>>();
    this.labels = new ArrayList<String>();
  }

  public void execute(String query, boolean groupBySecondColumn,
      boolean calculateSlope, String formatTime, List<Object> parameters) {
    dataset.clear();
    try {
      // The newInstance() call is a work around for some
      // broken Java implementations
      org.apache.hadoop.chukwa.util.DriverManagerUtil.loadDriver().newInstance();
    } catch (Exception ex) {
      log.error("failed to load driver", ex);
      // handle the error
    }
    Connection conn = null;
    PreparedStatement stmt = null;
    ResultSet rs = null;
    labels.clear();
    double max = 0.0;
    long timeWindowSize=0;
    long previousTime=0;
    try {
      conn = org.apache.hadoop.chukwa.util.DriverManagerUtil.getConnection(jdbc);
      stmt = conn.prepareStatement(query);
      if(query.indexOf("?")!=-1) {
        for(int i=0;i<parameters.size();i++) {
          int index = i+1;
          stmt.setObject(index,parameters.get(i));
        }
      }
      // rs = stmt.executeQuery(query);
      if (stmt.execute()) {
        rs = stmt.getResultSet();
        ResultSetMetaData rmeta = rs.getMetaData();
        int col = rmeta.getColumnCount();
        double[] previousArray = new double[col + 1];
        for (int k = 0; k < col; k++) {
          previousArray[k] = 0.0;
        }
        int i = 0;
        java.util.TreeMap<String, Double> data = null;
        HashMap<String, Double> previousHash = new HashMap<String, Double>();
        HashMap<String, Integer> xAxisMap = new HashMap<String, Integer>();
        while (rs.next()) {
          String label = "";
          if (rmeta.getColumnType(1) == java.sql.Types.TIMESTAMP) {
            long time = rs.getTimestamp(1).getTime();
            if(timeWindowSize==0) {
              timeWindowSize=1;
              previousTime=time;
            } else if(time!=previousTime) {
              timeWindowSize=(time-previousTime)/60000;
              previousTime=time;
            }
            label = "" + time;
          } else {
            label = rs.getString(1);
          }
          if (!xAxisMap.containsKey(label)) {
            xAxisMap.put(label, i);
            labels.add(label);
            i++;
          }
          if (groupBySecondColumn) {
            String item = rs.getString(2);
            // Get the data from the row using the series column
            for (int j = 3; j <= col; j++) {
              item = rs.getString(2) + " " + rmeta.getColumnName(j);
              data = dataset.get(item);
              if (data == null) {
                data = new java.util.TreeMap<String, Double>();
              }
              if (calculateSlope) {
                double current = rs.getDouble(j);
                double tmp = 0L;
                if (data.size() > 1) {
                  tmp = (current - previousHash.get(item).doubleValue())/timeWindowSize;
                  if(timeWindowSize<=0) {
                    tmp = Double.NaN;
                  }
                } else {
                  tmp = 0;
                }
                if (tmp < 0) {
                  tmp = Double.NaN;
                }
                previousHash.put(item, current);
                if (tmp > max) {
                  max = tmp;
                }
                data.put(label, tmp);
              } else {
                double current = rs.getDouble(j);
                if (current > max) {
                  max = current;
                }
                data.put(label, current);
              }
              dataset.put(item, data);
            }
          } else {
            for (int j = 2; j <= col; j++) {
              String item = rmeta.getColumnName(j);
              // Get the data from the row using the column name
              double current = rs.getDouble(j);
              if (current > max) {
                max = current;
              }
              data = dataset.get(item);
              if (data == null) {
                data = new java.util.TreeMap<String, Double>();
              }
              if (calculateSlope) {
                double tmp = current;
                if (data.size() > 1) {
                  tmp = (tmp - previousArray[j])/timeWindowSize;
                  if(timeWindowSize<=0) {
                    tmp = Double.NaN;
                  }
                } else {
                  tmp = 0.0;
                }
                if (tmp < 0) {
                  tmp = Double.NaN;
                }
                previousArray[j] = current;
                data.put(label, tmp);
              } else {
                data.put(label, current);
              }
              dataset.put(item, data);
            }
          }
        }
      } else {
        log.error("query is not executed.");
      }
      // Now do something with the ResultSet ....
    } catch (SQLException ex) {
      // handle any errors
      log.error("SQLException: " + ex.getMessage() + " on query: " + query);
      log.error("SQLState: " + ex.getSQLState());
      log.error("VendorError: " + ex.getErrorCode());
    } catch (Exception ex) {
      log.debug(ExceptionUtil.getStackTrace(ex));
    } finally {
      // it is a good idea to release
      // resources in a finally{} block
      // in reverse-order of their creation
      // if they are no-longer needed
      if (rs != null) {
        try {
          rs.close();
        } catch (SQLException sqlEx) {
          log.debug(ExceptionUtil.getStackTrace(sqlEx));
        }
        rs = null;
      }
      if (stmt != null) {
        try {
          stmt.close();
        } catch (SQLException sqlEx) {
          log.debug(ExceptionUtil.getStackTrace(sqlEx));
        }
        stmt = null;
      }
      if (conn != null) {
        try {
          conn.close();
        } catch (SQLException sqlEx) {
          log.debug(ExceptionUtil.getStackTrace(sqlEx));
        }
        conn = null;
      }
    }
  }

  public List<String> getXAxisMap() {
    return labels;
  }

  public TreeMap<String, TreeMap<String, Double>> getDataset() {
    return dataset;
  }
}