Как обрабатывать строки CSV файла с помощью Groovy/GPars наиболее эффективно?

Вопрос простой, и я удивлен, что он не появился сразу, когда я его искал.

У меня есть файл CSV, потенциально очень большой, который нужно обработать. Каждая строка должна передаваться процессору до тех пор, пока все строки не будут обработаны. Для чтения CSV файла я буду использовать OpenCSV, который по существу предоставляет метод readNext(), который дает мне следующую строку. Если больше строк не доступно, все процессоры должны завершиться.

Для этого я создал действительно простой groovy script, определил синхронный метод readNext() (поскольку чтение следующей строки не занимает много времени), а затем создал несколько потоков, которые читают следующую строку и обрабатывать его. Он отлично работает, но...

Не должно быть встроенного решения, которое я мог бы использовать? Это не обработка коллекции gpars, потому что всегда предполагает наличие существующей коллекции в памяти. Вместо этого я не могу позволить себе прочитать все это в памяти, а затем обработать его, это приведет к исключению из памяти.

Итак... кто имеет хороший шаблон для обработки CSV файла "по очереди", используя пару рабочих потоков?

Ответ 1

Одновременный доступ к файлу может быть не очень хорошей идеей, и обработка/совместная обработка GPars предназначена только для данных (коллекций) в памяти. Мой sugesstion должен был бы прочитать файл последовательно в список. Когда список достигнет определенного размера, обработайте записи в списке одновременно с помощью GPars, очистите список, а затем перейдите к строке чтения.

Ответ 2

Это может быть хорошей проблемой для актеров. Синхронный читатель-актер мог передавать CSV-линии актерам параллельного процессора. Например:

@Grab(group='org.codehaus.gpars', module='gpars', version='0.12')

import groovyx.gpars.actor.DefaultActor
import groovyx.gpars.actor.Actor

class CsvReader extends DefaultActor {
    void act() {
        loop {
            react {
                reply readCsv()
            }
        }
    }
}

class CsvProcessor extends DefaultActor {
    Actor reader
    void act() {
        loop {
            reader.send(null)
            react {
                if (it == null)
                    terminate()
                else
                    processCsv(it)
            }
        }
    }
}

def N_PROCESSORS = 10
def reader = new CsvReader().start()
(0..<N_PROCESSORS).collect { new CsvProcessor(reader: reader).start() }*.join()

Ответ 3

Я просто завершаю реализацию проблемы, подобной этой, в Grails (вы не указываете, используете ли вы грабли, простой спящий режим, простой JDBC или что-то еще).

Нет ничего из коробки, которую вы можете получить, о которой я знаю. Вы можете посмотреть на интеграцию с Spring Batch, но в последний раз, когда я смотрел на нее, мне было очень тяжело (и не очень groovy).

Если вы используете простой JDBC, то, что рекомендует Кристоф, возможно, является самой простой задачей (читайте в N строках и используйте GPars для одновременного вращения этих строк).

Если вы используете grails или hibernate и хотите, чтобы ваши рабочие потоки имели доступ к контексту Spring для инъекции зависимостей, все становится немного сложнее.

То, как я решил, это использовать плагин Grails Redis (отказ от ответственности: я автор) и Jesque plugin, который представляет собой реализацию Java Resque.

Плагин Jesque позволяет создавать классы "Job", которые имеют метод "process" с произвольными параметрами, которые используются для обработки работы, установленной в очереди Jesque. Вы можете развернуть столько рабочих, сколько хотите.

У меня есть загрузка файла, которую пользователь admin может отправить в файл, он сохраняет файл на диск и завершает работу для созданного мной ProducerJob. Этот ProducerJob вращается через файл, для каждой строки он выдает сообщение для ConsumerJob, чтобы забрать. Сообщение представляет собой просто карту значений, считанных из файла CSV.

ConsumerJob берет эти значения и создает для него соответствующий объект домена и сохраняет его в базе данных.

Мы уже использовали Redis в производстве, поэтому использование этого в качестве механизма очередей имело смысл. У нас была старая синхронная загрузка, которая периодически запускалась через файлы. В настоящее время я использую одного продюсера и 4 потребителя и загружаю вещи таким образом, это более чем в 100 раз быстрее, чем старый груз (с гораздо большей обратной связью с конечным пользователем).

Я согласен с оригинальным вопросом, что, вероятно, есть место для того, чтобы что-то подобное упаковывалось, поскольку это относительно обычная вещь.

UPDATE: я разместил сообщение в блоге с помощью простого примера, делающего импорт с Redis + Jesque.