Возможности для оптимизации производительности этой параллельной потоковой обработки CSV. ⇐ JAVA

[Anonymous]

Я пишу потоковый процессор csv. чтобы создать IdMapping, как показано. Здесь после перехода с обычного потока на параллельный и введения concurrent-map производительность значительно выросла. У меня около 500 тысяч строк в этом CSV-потоке, найдите пример строки:

Country,StudioNetwork,MediaId,Title,OriginalReleaseDate,MediaType,SeriesMediaId,SeasonMediaId,SeriesTitle, SeasonNumber,EpisodeNumber,LicenseType,ActualRetailPrice,OfferStartDate,OfferEndDate,Актеры,Режиссер,XboxLiveURL
СОЕДИНЕННЫЕ ШТАТЫ,,531b964f-0cb9-4968-9b77-e547f2435225,Зои 101: Заглядывает в
PCA,1 /23/ 2012 12:00:00
AM,TvSeason,c3eddf64-372f-4dfc-a763-2681cce65ac2,,Зои 101: Заглядывает в
PCA,1,,EST HD,12.99,30 января 2012 8 :01:00,01.01.3000 8:01:00
AM,,,https://video.xbox.com/tv- Season/531b964f-0cb9-4968-9b77-e547f2435225

Код: Выделить всё

 @Override
public List match(Matcher.CsvStream externalDb, MatchingContext matchingContext) {

// Map to store the ID mappings between internal and external databases
final Map idMappingMap = new ConcurrentHashMap();

// Extract column indices from the header row of the external database
final Map columnIndices = ParsingUtils.generateHeaderMap(externalDb.getHeaderRow());
int titleIdx = columnIndices.get("Title");
int dateIdx = columnIndices.get("OriginalReleaseDate");
int directorIdx = columnIndices.get("Director");
int actorIdx = columnIndices.get("Actors");
int mediaId = columnIndices.get("MediaId");

// Retrieve the internal movie index from the matching context
Map movieIndex = matchingContext.getMovieIndex();

//AtomicInteger count = new AtomicInteger();
externalDb.getDataRows().parallel().forEach(row -> {
//count.getAndIncrement();
try (CSVReader csvReader = new CSVReader(new StringReader(row))) {
String[] columns = csvReader.readNext();

if (columns != null) {
String externalTitle = columns[titleIdx].trim().toLowerCase();
String externalDate = columns[dateIdx].trim();
String externalDirector = columns[directorIdx].trim().toLowerCase();
List externalActors = Arrays.stream(columns[actorIdx].split(","))
.map(String::trim)
.map(String::toLowerCase)
.collect(Collectors.toList());

// Parse year from OriginalReleaseDate if needed
int externalYear = ParsingUtils.parseYear(extractYear(externalDate));

String key = externalTitle + "_" + externalYear;

// Find matching movie in internal DB
Optional matchedMovie = Optional.ofNullable(movieIndex.get(key))
.filter(movie -> isMatch(movie, externalTitle, externalYear, externalActors, externalDirector, matchingContext));

matchedMovie.ifPresent(movie -> {
// Use ConcurrentMap to handle concurrent access TODO
idMappingMap.putIfAbsent(movie.getId(), new Matcher.IdMapping(movie.getId(), columns[mediaId].trim()));
});
}
} catch (Exception e) {
LOGGER.error("Error processing row: {}", row, e);
}
});
//LOGGER.info("processed {} xbox rows", count);

return new ArrayList(idMappingMap.values());
}

Поскольку параллельные потоки не дают гарантии размера строки, будет ли это работать, когда количество строк увеличится, скажем, до 5 миллионов?
Есть ли способ оптимизировать это дальше?
Поскольку параллельные потоки не дают гарантии размера строки, будет ли это работать, когда количество строк увеличится, скажем, до 5 миллионов?
Есть ли способ оптимизировать это дальше?
Поскольку параллельные потоки не дают гарантии размера строки, будет ли это работать, когда количество строк увеличится, скажем, до 5 миллионов? п>

Подробнее здесь: https://stackoverflow.com/questions/788 ... ing-of-csv