@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		super.prepare(stormConf, context, collector);
		
		 _topWords = new SpaceSaving<String>(_numWords * 50);

		// Setup output writing at a fixed interval.
		Timer outputToFileTimer = new Timer();
		outputToFileTimer.scheduleAtFixedRate(new TimerTask() {

			@Override
			public void run() {
				if (_startDate != null) {
					float numSecondsSinceStart = 
							(new Date().getTime() - _startDate.getTime()) / 1000f;
					long throughput = 
							(long) (_numberOfWordsProcessesd / numSecondsSinceStart);

					Writer writer = null;

					try {
						writer = new BufferedWriter(new OutputStreamWriter(
								new FileOutputStream(OUTPUT_FILE_NAME), "utf-8"));
						writer.write("Top Words Statistics:\n");
						writer.write("\n");
						writer.write("# Words processed: " 
								+ _numberOfWordsProcessesd + "\n");
						writer.write("Time elapsed:       " 
								+ numSecondsSinceStart + "\n");
						writer.write("Throughput:         " 
								+ throughput + " wps\n");
						
						writer.write("\n");
						synchronized (_topWords) {
							for (Counter<String> counter : _topWords.topK(_numWords)) {
								writer.write("Item: " + counter.getItem() 
										+ " Count: " + counter.getCount() 
										+ " Error: " + counter.getError() 
										+ " Count Per Second: " + counter.getCount() / numSecondsSinceStart
										+ " Error Per Second: " + counter.getError() / numSecondsSinceStart
										+ "\n");
							}
						}
					} catch (IOException ex) {
						Logger.getLogger(StatusThroughputRecorderBolt.class).error(
								"Error while writing words statistics.");
					} finally {
						try {
							writer.close();
						} catch (Exception ex) {}
					}
				}
			}
		}, FILE_WRITE_DELAY, FILE_WRITE_PERIOD);
		
		// Read in all the stop words from the file.
		BufferedReader reader = null;
		try {
			reader = new BufferedReader(new FileReader(STOP_WORDS_FILE_NAME));
			String line;
			while ((line = reader.readLine()) != null) {
				_stopWords.add(line.replaceAll("\\s+", ""));
			}
		} catch (IOException e) {
			
		} finally {
			if (reader != null) {
				try { reader.close(); } catch (IOException e) { /* Ensure close happens. */ }
			}
		}
	}