public ReportWriterImpl() {
		String contactPointsStr = PropertyHelper.getProperty("contactPoints", "localhost");
		int creditThreads = Integer.parseInt(PropertyHelper.getProperty("creditThreads", "3"));
		int debitThreads = Integer.parseInt(PropertyHelper.getProperty("debitThreads", "3"));
		BATCH_SIZE = Integer.parseInt(PropertyHelper.getProperty("batchSize", "100000"));

		ExecutorService creditExecutors = Executors.newFixedThreadPool(creditThreads);
		ExecutorService debitExecutors = Executors.newFixedThreadPool(debitThreads);
		ExecutorService deadLetterExecutors = Executors.newSingleThreadExecutor();

		Cluster cluster = Cluster.builder().addContactPoints(contactPointsStr.split(",")).build();
		this.session = cluster.connect();

		this.transactionDao = new TransactionsDao(session);
		this.reportDao = new ReportDao(session);

		// Create collectors
		for (int i = 0; i < creditThreads; i++) {
			creditExecutors.execute(new CreditWriter(this.creditCollection, this));
		}
		for (int i = 0; i < creditThreads; i++) {
			debitExecutors.execute(new DebitWriter(this.debitCollection, this));
		}
		for (int i = 0; i < creditThreads; i++) {
			deadLetterExecutors.execute(new DeadLetterWriter(this.deadLetterCollection, this));
		}

		try {
			// Start processing data
			this.runReport();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		try {
			Thread.sleep(1000);
			closeAllFiles();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
			logger.warn("Problem closing files - please check : " + e.getMessage());
		}

		session.close();
		cluster.close();

		System.exit(0);
	}