New tool dcdbcsvimport to import data vom CSV files to DCDB

DCDBCSVImport/Makefile

+include ../config.mk
+
+CXXFLAGS =	-O2 -ggdb --std=c++11 -Wall -Wno-unused-local-typedefs -Wno-unknown-warning-option -fmessage-length=0 -I$(DCDBDEPLOYPATH)/include/ -I$(DCDBBASEPATH)/include/ -DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG
+OBJS =		dcdbcsvimport.o
+LIBS =		-L$(DCDBDEPLOYPATH)/lib/ -ldcdb -lcassandra -luv -lboost_random -lboost_system -lboost_date_time -lboost_regex -lssl -lcrypto -lpthread
+TARGET =	dcdbcsvimport
+
+.PHONY : clean install
+
+$(TARGET): $(OBJS)
+	$(CXX) -o $(TARGET) $(OBJS) $(LIBS)
+
+all:	$(TARGET)
+
+clean:
+	rm -f $(TARGET)
+	rm -f $(OBJS)
+
+install: $(TARGET)
+	install $(TARGET) $(DCDBDEPLOYPATH)/bin/

DCDBCSVImport/dcdbcsvimport.cpp

+/*
+ * dcdbbenchmark.cpp
+ *
+ *  Created on: Nov 05, 2015
+ *      Author: Axel Auweter
+ */
+
+#define SENSORS 20
+#define INSERTS_PER_SENSOR 100000
+#define INSERTS_INTERVAL 1000000000
+#define THREADS 4
+#define USE_SIN_DATA
+
+#include <iostream>
+#include <cstdio>
+#include <cstdint>
+#include <cmath>
+#include <fstream>
+#include <iomanip> 
+#include <sstream> 
+#include <vector> 
+#include <boost/tokenizer.hpp>
+#include <algorithm>
+#include <exception>
+
+#include <pthread.h>
+
+#include <dcdb/connection.h>
+#include <dcdb/sensorconfig.h>
+#include <dcdb/sensordatastore.h>
+#include <dcdb/sensorid.h>
+#include <dcdb/timestamp.h>
+#include <dcdb/c_api.h>
+
+typedef struct {
+	std::string name;
+	std::string topic;
+	std::string publicName;
+} sensor_t;
+
+int main(int argc, char** argv)
+{
+  /* Connect to the data store */
+  std::cout << std::endl;
+  std::cout << "Connecting to the data store..." << std::endl;
+
+  DCDB::Connection* connection;
+  connection = new DCDB::Connection();
+  connection->setHostname("127.0.0.1");
+  if (!connection->connect()) {
+      std::cout << "Cannot connect to database." << std::endl;
+      return 1;
+  }
+
+  /* Initialize the SensorConfig and SensorDataStore interfaces */
+  DCDB::SensorConfig sensorConfig(connection);
+  DCDB::SensorDataStore sensorDataStore(connection);
+
+  /* Insert test data */
+  std::ifstream fs;
+  std::string s;
+  std::vector<std::string> vec;
+  std::vector<sensor_t> sensors;
+  std::string prefix = argv[2];
+  
+
+  fs.open (argv[1], std::fstream::in);
+  std::getline(fs, s);
+
+  boost::tokenizer<boost::escaped_list_separator<char> > tk(s, boost::escaped_list_separator<char>('\\', ',', '\"'));
+  int topics = 0;
+  for (boost::tokenizer<boost::escaped_list_separator<char> >::iterator i=tk.begin(); i!=tk.end();++i) 
+  {
+	  if (i == tk.begin()) {
+		  continue;
+	  }
+	sensor_t sensor;
+	sensor.name = *i;
+	std::stringstream ss;
+	if (prefix.size() < 12) {
+		ss << std::setfill('0') << std::setw((12-prefix.size())*2) << "";
+	}
+	const char *c = prefix.substr(0, 12).c_str();
+	while (*c) {
+		uint32_t x = *c++;
+		ss << std::setfill('0') << std::setw(2) << std::hex << x;
+	}
+	ss << "/" << std::setfill('0') << std::setw(8) << std::hex << topics;
+	sensor.topic = ss.str();
+	sensor.publicName = prefix + "." + sensor.name;
+	std::replace(sensor.publicName.begin(), sensor.publicName.end(), ' ', '_');
+	sensors.push_back(sensor);
+    topics++;
+  }
+  
+  while (std::getline(fs, s)) {
+    int col = 0;
+	uint64_t ts = 0;
+	boost::tokenizer<boost::escaped_list_separator<char> > tk(s, boost::escaped_list_separator<char>('\\', ',', '\"'));
+    for (boost::tokenizer<boost::escaped_list_separator<char> >::iterator i=tk.begin(); i!=tk.end();++i) 
+    {
+		if (0 == col) {
+			ts = std::stoull(*i) * 1000000;
+		} else {
+			std::cout << ts << " " << sensors[col-1].topic << ":" << *i << std::endl;
+			try {
+				DCDB::SensorId sid(sensors[col-1].topic);
+				sensorDataStore.insert(&sid, ts, std::stoll(*i));
+			}
+			catch (std::exception &e) {
+				std::cout << e.what() << ":" << *i << std::endl;
+			}
+		}
+		col++;
+    }	  
+  }
+
+  /* Create public sensor names */
+  std::cout << std::endl;
+  std::cout << "Publishing sensors..." << std::endl;
+
+  std::vector<sensor_t>::iterator it;
+  for (it = sensors.begin(); it != sensors.end(); it++) {
+	  std::cout << it->name << " " << it->topic << " " << it->publicName << std::endl;
+	  sensorConfig.publishSensor(it->publicName.c_str(), it->topic.c_str());
+  }
+
+  /* Disconnect */
+  delete connection;
+
+  return 0;
+}
+