ZipfDataLoader.h

Go to the documentation of this file.

 
#ifndef _INCLUDE_TESTBENCH_ZIPFDATALOADER_H_
#define _INCLUDE_TESTBENCH_ZIPFDATALOADER_H_
#include <Utils/ConfigMap.hpp>
#include <Common/Tuples.h>
#include <assert.h>
#include <Utils/IntelliLog.h>
#include <TestBench/AbstractDataLoader.h>
#include <vector>
#include <Utils/MicroDataSet.hpp>
using namespace INTELLI;
namespace OoOJoin {
 
class ZipfDataLoader : public AbstractDataLoader {
 protected:
  tsType windowLenMs, timeStepUs, watermarkTimeMs, maxArrivalSkewMs, eventRateKTps;
  uint64_t zipfDataLoader_zipfKey, zipfDataLoader_zipfValue, zipfDataLoader_zipfEvent, zipfDataLoader_zipfSkew;
  vector<keyType> keyS, keyR;
  vector<valueType> valueS, valueR;
  vector<tsType> eventS, eventR;
  vector<tsType> arrivalS, arrivalR;
  size_t testSize, keyRange, valueRange;
  double zipfDataLoader_zipfKeyFactor, zipfDataLoader_zipfValueFactor, zipfDataLoader_zipfEventFactor,
      zipfDataLoader_zipfSkewFactor;
  MicroDataSet md;
  vector<tsType> genArrivalTime(vector<tsType> eventTime, vector<tsType> arrivalSkew) {
    vector<tsType> ru = vector<tsType>(eventTime.size());
    size_t len = (eventTime.size() > arrivalSkew.size()) ? arrivalSkew.size() : eventTime.size();
    for (size_t i = 0; i < len; i++) {
      ru[i] = eventTime[i] + arrivalSkew[i];
    }
    return ru;
  }
  vector<OoOJoin::TrackTuplePtr> genTuples(vector<keyType> keyS,
                                           vector<valueType> valueS,
                                           vector<tsType> eventS,
                                           vector<tsType> arrivalS) {
    size_t len = keyS.size();
    vector<OoOJoin::TrackTuplePtr> ru = vector<OoOJoin::TrackTuplePtr>(len);
    for (size_t i = 0; i < len; i++) {
      ru[i] = newTrackTuple(keyS[i], valueS[i], eventS[i], arrivalS[i]);
    }
    return ru;
  }
  void genKey();
  void genValue();
  void genEvent();
  void genArrival();
  void genFinal();
 public:
  ConfigMapPtr cfgGlobal;
  vector<TrackTuplePtr> sTuple, rTuple;
  ZipfDataLoader() {}
  ~ZipfDataLoader() {}
  virtual bool setConfig(ConfigMapPtr cfg);
  virtual bool setModConfig(ConfigMapPtr cfg) {
    return AbstractDataLoader::setModConfig(cfg);
  }
  virtual vector<TrackTuplePtr> getTupleVectorS();
  virtual vector<TrackTuplePtr> getTupleVectorR();
};
 
typedef std::shared_ptr<class ZipfDataLoader> ZipfDataLoaderPtr;
#define newZipfDataLoader std::make_shared<OoOJoin::ZipfDataLoader>
}
#endif //INTELLISTREAM_INCLUDE_TESTBENCH_ZipfDataLoader_H_