Как использовать очереди задач Google App Engine (Java) для сохранения большого количества данных / строк в одном объекте - PullRequest
Новая
       0

Как использовать очереди задач Google App Engine (Java) для сохранения большого количества данных / строк в одном объекте

3 голосов
/ 16 февраля 2011

Я пытаюсь сохранить примерно 28 000 «строк» ​​в одной сущности, например СОТРУДНИК

По сути, моя цель состоит в том, чтобы избежать прерывания / тайм-аута с использованием значений PUT, превышающих 30 секунд - что может случиться, если я просто выполню 28 000 PUT, вызвав запрос doPost (), отправленный сервлету.

Так что я думаю об использовании задач, описанных в документации Google App Engine.

По сути, я хотел бы загрузить CSV-файл в каталог war с 28 000 "сотрудников". Затем создайте задачу, которая будет асинхронно ПРОВЕРИТЬ эти 28 000 строк сотрудников в объект EMPLOYEE.

  • В1: Это жизнеспособное решение или есть лучший способ? Опять же, цель состоит в том, чтобы выполнить PUT, чтобы избежать прерывания из-за 30-секундного ограничения.

  • Q2: Кроме того, какие конфигурации queue.xml я должен использовать, чтобы обеспечить максимально быстрое выполнение этих PUT?

  • Q3: Теперь я попробовал это, подобно записи в блоге: http://gaejexperiments.wordpress.com/2009/11/24/episode-10-using-the-task-queue-service/, но примерно через 23 секунды я получаю следующую ошибку: 

    SEVERE: Job default.task1 threw an unhandled Exception: 
com.google.apphosting.api.ApiProxy$ApplicationException: ApplicationError: 5: http method POST against URL http://127.0.0.1:8888/dotaskservlet timed out.
    at com.google.appengine.api.urlfetch.dev.LocalURLFetchService.fetch(LocalURLFetchService.java:236)
    at com.google.appengine.api.taskqueue.dev.LocalTaskQueue$UrlFetchServiceLocalTaskQueueCallback.execute(LocalTaskQueue.java:471)
    at com.google.appengine.api.taskqueue.dev.UrlFetchJob.execute(UrlFetchJob.java:77)
    at org.quartz.core.JobRunShell.run(JobRunShell.java:203)
    at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:520)
16/02/2011 12:12:55 PM org.quartz.core.ErrorLogger schedulerError
SEVERE: Job (default.task1 threw an exception.
org.quartz.SchedulerException: Job threw an unhandled exception. [See nested exception: com.google.apphosting.api.ApiProxy$ApplicationException: ApplicationError: 5: http method POST against URL http://127.0.0.1:8888/dotaskservlet timed out.]
    at org.quartz.core.JobRunShell.run(JobRunShell.java:214)
    at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:520)
* Nested Exception (Underlying Cause) ---------------
com.google.apphosting.api.ApiProxy$ApplicationException: ApplicationError: 5: http method POST against URL http://127.0.0.1:8888/dotaskservlet timed out.
    at com.google.appengine.api.urlfetch.dev.LocalURLFetchService.fetch(LocalURLFetchService.java:236)
    at com.google.appengine.api.taskqueue.dev.LocalTaskQueue$UrlFetchServiceLocalTaskQueueCallback.execute(LocalTaskQueue.java:471)
    at com.google.appengine.api.taskqueue.dev.UrlFetchJob.execute(UrlFetchJob.java:77)
    at org.quartz.core.JobRunShell.run(JobRunShell.java:203)
    at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:520)

  • Q4: я также проверил средство просмотра хранилища данных на http://localhost:8888/_ah/admin, и, похоже, в этом объекте было создано только 1000 результатов. 1000 лимит?

  • Q5: Как мне избавиться от этой ошибки?

  • В6: Можно ли подтвердить, что максимально допустимое время составляет 10 минут для задачи? или это все еще 30 секунд? Я наткнулся на это: http://code.google.com/appengine/docs/java/taskqueue/overview.html#Task_Execution

Ответы [ 3 ]

2 голосов
/ 17 февраля 2011

Вот пример / руководство по использованию mapreduce для синтаксического анализа CSV-файла, похоже на ваши потребности:

http://ikaisays.com/2010/08/11/using-the-app-engine-mapper-for-bulk-data-import/

0 голосов
/ 01 ноября 2013

Я бы сделал это с помощью пакетного сохранения через DeferredTask, примерно так: 

List<Employee> employees=...
EmployeeWriter qr = new EmployeeWriter (employees);
TaskHandle task = QueueFactory.getDefaultQueue().add(withPayload(qr));

, где 

public class EmployeeWriter implements DeferredTask {
   public EmployeeWriter () {    }
   public EmployeeWriter (List<Employee> employees) { 
          this.employees=new LinkedList(employees);  
   }

    private LinkedList<Employee> employees;

    @Override
    public void run() {
      Stopwatch sw = Stopwatch.createStarted();
      try {
        do {
           List employeesTosave=Pull100EmployeesFromLinkedList(employees)
           ofy().save(employeesTosave).now();

        } while (sw.elapsed(TimeUnit.MINUTES) < 9);
     finally {
        sw.stop();
        if (!employees.isEmpty()) {
            QueueFactory.getDefaultQueue().add(withPayload(this));
        }
    }

}
0 голосов
/ 16 февраля 2011

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

http://ikaisays.com/2010/06/10/using-the-bulkloader-with-java-app-engine/

...