Верблюд: как разделить, а затем агрегировать, когда количество элементов меньше размера партии - PullRequest
Новая
       9

Верблюд: как разделить, а затем агрегировать, когда количество элементов меньше размера партии

0 голосов
/ 03 января 2019

У меня есть маршрут Camel, который читает файл из S3 и обрабатывает входной файл следующим образом:

  1. Разбор каждой строки в POJO (Student) с использованием Bindy
  2. Разделить вывод по телу ()
  3. Агрегировать по атрибуту тела (.semester) и размер пакета 2
  4. Invokeслужба сохранения для загрузки в БД в заданных пакетах

Проблема заключается в том, что при размере пакета 2 и нечетном количестве записей всегда существует одна запись, которая не сохраняется.

Предоставляется код Kotlin , но он не должен сильно отличаться от эквивалентного кода Java (за исключением косой черты перед "\ $ {simple expression}" или отсутствия точек с запятой для завершения операторов.

Если я установлю размер пакета на 1, то каждая запись будет сохранена, в противном случае последняя запись никогда не будет сохранена.

Я несколько раз проверял документацию для обработчика сообщений , нопохоже, он не охватывает этот конкретный сценарий.

Я также установил [completionTimeout | completionInterval] в дополнение к completionSize, но это не имеет значения.

Кто-нибудь сталкивался с этой проблемой раньше? 

val csvDataFormat = BindyCsvDataFormat(Student::class.java)

from("aws-s3://$student-12-bucket?amazonS3Client=#amazonS3&delay=5000")
    .log("A new Student input file has been received in S3: '\${header.CamelAwsS3BucketName}/\${header.CamelAwsS3Key}'")
    .to("direct:move-input-s3-object-to-in-progress")
    .to("direct:process-s3-file")
    .to("direct:move-input-s3-object-to-completed")
    .end()

from("direct:process-s3-file")
    .unmarshal(csvDataFormat)
    .split(body())
    .streaming()
    .parallelProcessing()
    .aggregate(simple("\${body.semester}"), GroupedBodyAggregationStrategy())
    .completionSize(2)
    .bean(persistenceService)
    .end()

С входным CSV-файлом, включающим семь (7) записей, это сгенерированный вывод (с некоторыми добавленными журналами отладки): 

WARN 19540 --- [student-12-move] c.a.s.s.internal.S3AbortableInputStream  : Not all bytes were read from the S3ObjectInputStream, aborting HTTP connection. This is likely an error and may result in sub-optimal behavior. Request only the bytes you need via a ranged GET or drain the input stream after use.
 INFO 19540 --- [student-12-move] student-workflow-main                    : A new Student input file has been received in S3: 'student-12-bucket/inbox/foo.csv'
 INFO 19540 --- [student-12-move] move-input-s3-object-to-in-progress      : Moving S3 file 'inbox/foo.csv' to 'in-progress' folder...
 INFO 19540 --- [student-12-move] student-workflow-main                    : Moved input S3 file 'in-progress/foo.csv' to 'in-progress' folder...
 INFO 19540 --- [student-12-move] pre-process-s3-file-records              : Start saving to database...
DEBUG 19540 --- [read #7 - Split] c.b.i.d.s.StudentPersistenceServiceImpl  : Saving record to database: Student(id=7, name=Student 7, semester=2nd, javaMarks=25)
DEBUG 19540 --- [read #7 - Split] c.b.i.d.s.StudentPersistenceServiceImpl  : Saving record to database: Student(id=5, name=Student 5, semester=2nd, javaMarks=81)
DEBUG 19540 --- [read #3 - Split] c.b.i.d.s.StudentPersistenceServiceImpl  : Saving record to database: Student(id=6, name=Student 6, semester=1st, javaMarks=15)
DEBUG 19540 --- [read #3 - Split] c.b.i.d.s.StudentPersistenceServiceImpl  : Saving record to database: Student(id=2, name=Student 2, semester=1st, javaMarks=62)
DEBUG 19540 --- [read #2 - Split] c.b.i.d.s.StudentPersistenceServiceImpl  : Saving record to database: Student(id=3, name=Student 3, semester=2nd, javaMarks=72)
DEBUG 19540 --- [read #2 - Split] c.b.i.d.s.StudentPersistenceServiceImpl  : Saving record to database: Student(id=1, name=Student 1, semester=2nd, javaMarks=87)
 INFO 19540 --- [student-12-move] device-group-workflow-main               : End pre-processing S3 CSV file records...
 INFO 19540 --- [student-12-move] move-input-s3-object-to-completed        : Moving S3 file 'in-progress/foo.csv' to 'completed' folder...
 INFO 19540 --- [student-12-move] device-group-workflow-main               : Moved S3 file 'in-progress/foo.csv' to 'completed' folder...

1 Ответ

0 голосов
/ 03 января 2019

Если вам нужно немедленно завершить ваше сообщение, вы можете указать предикат завершения, который основан на свойствах обмена, установленных сплиттером. Я не пробовал это, но я думаю 

.completionPredicate( simple( "${exchangeProperty.CamelSplitComplete}" ) )

обработает последнее сообщение.

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

...