Commit c01aa31b authored by Matija Obreza's avatar Matija Obreza
Browse files

Added _parsed columns for DECLATITUDE, DECLONGITUDE

parent 78d58890
......@@ -59,6 +59,7 @@ public interface ProcessService {
symbols.setDecimalSeparator(decimalMark);
symbols.setGroupingSeparator(decimalMark == ',' ? '.' : ',');
decimalFormat.setDecimalFormatSymbols(symbols);
decimalFormat.setMinimumFractionDigits(8);
return decimalFormat;
}
}
......@@ -77,6 +77,18 @@ public class CountryProcessServiceImpl implements ProcessService {
outputHeaders.add(pos + 1, ApplicationUtils.HEADER_ORIGCTY_CHECK);
outputMapping.add(pos + 1, null);
}
// Add extra header HEADER_DECLATITUDE_CHECK
{
int pos = outputHeaders.indexOf(ApplicationUtils.HEADER_LATITUDE);
outputHeaders.add(pos + 1, ApplicationUtils.HEADER_DECLATITUDE_CHECK);
outputMapping.add(pos + 1, null);
}
// Add extra header HEADER_DECLONGITUDE_CHECK
{
int pos = outputHeaders.indexOf(ApplicationUtils.HEADER_LONGITUDE);
outputHeaders.add(pos + 1, ApplicationUtils.HEADER_DECLONGITUDE_CHECK);
outputMapping.add(pos + 1, null);
}
LOG.info("Output headers: {}", outputHeaders);
LOG.info("Source mapping indexes: {}", outputMapping);
......@@ -85,7 +97,7 @@ public class CountryProcessServiceImpl implements ProcessService {
// Thread pool with nThreads concurrent threads
int nThreads = Math.max(Runtime.getRuntime().availableProcessors() - 1, 1);
LOG.info("Using " + nThreads + " threads for " + Runtime.getRuntime().availableProcessors() + " CPUs.");
LOG.info("Using {} threads for {} CPUs.", nThreads, Runtime.getRuntime().availableProcessors());
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(nThreads * 2);
......@@ -108,7 +120,7 @@ public class CountryProcessServiceImpl implements ProcessService {
stopWatch.split();
long processingTimeSeconds = stopWatch.getSplitTime() / 1000;
if ((lineCount % 1000 == 0 && processingTimeSeconds > 0) || (lastLogTime != processingTimeSeconds && processingTimeSeconds % 10 == 0)) {
LOG.info("FYI, " + lineCount + " entries have been processed in " + processingTimeSeconds + " = " + (lineCount / processingTimeSeconds) + " lines/second");
LOG.info("FYI, {} entries have been processed in {}s = {} lines/second", lineCount, processingTimeSeconds, ((float)lineCount / processingTimeSeconds));
lastLogTime = processingTimeSeconds;
}
......@@ -142,6 +154,8 @@ public class CountryProcessServiceImpl implements ProcessService {
float longitude = decimalFormat.parse(declongitude).floatValue();
float latitude = decimalFormat.parse(declatitude).floatValue();
LOG.debug("Parsed DECLATITUDE/DECLONGITUDE ({}, {}) --> ({}, {})", declongitude, declatitude, longitude, latitude);
outputLine[outputHeaders.indexOf(ApplicationUtils.HEADER_DECLATITUDE_CHECK)] = decimalFormat.format(latitude);
outputLine[outputHeaders.indexOf(ApplicationUtils.HEADER_DECLONGITUDE_CHECK)] = decimalFormat.format(longitude);
try {
result = countryOfOriginService.getCountries(longitude, latitude, origCty, ApplicationUtils.ALLOWED_DISTANCE_MARGIN);
......@@ -183,12 +197,8 @@ public class CountryProcessServiceImpl implements ProcessService {
LOG.info("Shut down.");
stopWatch.stop();
long processingTimeSeconds = stopWatch.getTime() / 1000;
try {
LOG.info("FYI, " + lineCount + " entries have been processed in " + processingTimeSeconds + " = " + (lineCount / processingTimeSeconds) + " lines/second");
} catch (Exception e) {
LOG.info("FYI, " + lineCount + " entries have been processed in " + processingTimeSeconds);
}
long processingTimeSeconds = Math.max(1, stopWatch.getTime() / 1000);
LOG.info("FYI, {} entries have been processed in {}s = {}lines/ms = {}ms/line", lineCount, processingTimeSeconds, ((float) lineCount / stopWatch.getTime()), ((float) stopWatch.getTime() / lineCount));
}
return lines;
}
......
......@@ -73,9 +73,21 @@ public class LandOrSeaProcessServiceImpl implements ProcessService {
Map<String, Integer> sourceMapping = HeaderUtils.makeSourceMapping(headers, new String[] { ApplicationUtils.HEADER_LONGITUDE, ApplicationUtils.HEADER_LATITUDE });
// Add extra header
// Add extra header HEADER_DECLATITUDE_CHECK
{
int pos = Math.max(outputHeaders.indexOf(ApplicationUtils.HEADER_LONGITUDE), outputHeaders.indexOf(ApplicationUtils.HEADER_LATITUDE));
int pos = outputHeaders.indexOf(ApplicationUtils.HEADER_LATITUDE);
outputHeaders.add(pos + 1, ApplicationUtils.HEADER_DECLATITUDE_CHECK);
outputMapping.add(pos + 1, null);
}
// Add extra header HEADER_DECLONGITUDE_CHECK
{
int pos = outputHeaders.indexOf(ApplicationUtils.HEADER_LONGITUDE);
outputHeaders.add(pos + 1, ApplicationUtils.HEADER_DECLONGITUDE_CHECK);
outputMapping.add(pos + 1, null);
}
// Add extra header HEADER_LANDORSEA_CHECK
{
int pos = Math.max(outputHeaders.indexOf(ApplicationUtils.HEADER_DECLONGITUDE_CHECK), outputHeaders.indexOf(ApplicationUtils.HEADER_DECLATITUDE_CHECK));
outputHeaders.add(pos + 1, ApplicationUtils.HEADER_LANDORSEA_CHECK);
outputMapping.add(pos + 1, null);
}
......@@ -87,7 +99,7 @@ public class LandOrSeaProcessServiceImpl implements ProcessService {
// Thread pool with nThreads concurrent threads(TEST!!!!)
int nThreads = Math.max(Runtime.getRuntime().availableProcessors() - 1, 1);
LOG.info("Using " + nThreads + " threads for " + Runtime.getRuntime().availableProcessors() + " CPUs.");
LOG.info("Using {} threads for {} CPUs.", nThreads, Runtime.getRuntime().availableProcessors());
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(nThreads * 2);
......@@ -109,7 +121,7 @@ public class LandOrSeaProcessServiceImpl implements ProcessService {
stopWatch.split();
long processingTimeSeconds = stopWatch.getSplitTime() / 1000;
if ((lineCount % 1000 == 0 && processingTimeSeconds > 0) || (lastLogTime != processingTimeSeconds && processingTimeSeconds % 10 == 0)) {
LOG.info("FYI, " + lineCount + " entries have been processed in " + processingTimeSeconds + " = " + (lineCount / processingTimeSeconds) + " lines/second");
LOG.info("FYI, {} entries have been processed in {}s = {} lines/second", lineCount, processingTimeSeconds, ((float)lineCount / processingTimeSeconds));
lastLogTime = processingTimeSeconds;
}
......@@ -140,6 +152,8 @@ public class LandOrSeaProcessServiceImpl implements ProcessService {
float longitude = decimalFormat.parse(declongitude).floatValue();
float latitude = decimalFormat.parse(declatitude).floatValue();
LOG.debug("Parsed DECLATITUDE/DECLONGITUDE ({}, {}) --> ({}, {})", declongitude, declatitude, longitude, latitude);
outputLine[outputHeaders.indexOf(ApplicationUtils.HEADER_DECLATITUDE_CHECK)] = decimalFormat.format(latitude);
outputLine[outputHeaders.indexOf(ApplicationUtils.HEADER_DECLONGITUDE_CHECK)] = decimalFormat.format(longitude);
result = landOrSeaService.classifyLocation(longitude, latitude, ApplicationUtils.ALLOWED_DISTANCE_MARGIN);
} catch (Exception e) {
......@@ -174,7 +188,7 @@ public class LandOrSeaProcessServiceImpl implements ProcessService {
stopWatch.stop();
long processingTimeSeconds = Math.max(1, stopWatch.getTime() / 1000);
LOG.info("FYI, " + lineCount + " entries have been processed in " + processingTimeSeconds + " = " + (lineCount / processingTimeSeconds) + " lines/second");
LOG.info("FYI, {} entries have been processed in {}s = {}lines/ms = {}ms/line", lineCount, processingTimeSeconds, ((float) lineCount / stopWatch.getTime()), ((float) stopWatch.getTime() / lineCount));
}
return lines;
......
......@@ -81,4 +81,8 @@ public class ApplicationUtils {
/** The Constant HEADER_LANDORSEA_CHECK. */
public static final String HEADER_LANDORSEA_CHECK = "LANDorSEA_check";
public static final String HEADER_DECLATITUDE_CHECK = "DECLATITUDE_parsed";
public static final String HEADER_DECLONGITUDE_CHECK = "DECLONGITUDE_parsed";
}
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment