@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = getLine(value);
			lastKLines.add(line);
			while (lastKLines.size() > k) {
				lastKLines.poll();
			}

			if (line == null || line.isEmpty()) {
				return;
			}

			// Handle line breaks escaped by backslash
			if (lineBuffer != null) {
				// This line is a continuation of the previous
				line = lineBuffer + line;
				lineBuffer = null;
			}
			if (line.charAt(line.length() - 1) == '\\') {
				// This line continues on the next
				lineBuffer = line.substring(0, line.length() - 1);
				return;
			}

			// Detect abnormal lines
			String[] lineSplit = Tweet.split(line);
			if (lineSplit.length > Tweet.EXPECTED_COLUMNS) {
				printError(line, "Too many columns: " + lineSplit.length);
				line = value.toString();
			} else if (lineSplit.length < Tweet.EXPECTED_COLUMNS) {
				lineBuffer = line;
				return;
			}

			// Create tweet
			try {
				context.write(new Tweet(lineSplit), NULL);
			} catch (IllegalArgumentException e) {
				printError(line, e.getMessage());
			}
		}