GAE에서 대형 파일의 처리를 처리하는 방법은 무엇입니까?

Question

Google App Engine에서 대용량 파일 처리를 처리하는 강력하고 빠른 방법을 찾고 있습니다. 그것은 (끝 단순화 워크 플로우) 다음과 같은 작동

:

1. 고객은 우리의 서버 라인으로 라인을 치료하는 것, CSV 파일을 보낼 수 있습니다.
2. 파일이 업로드되면 NDB 데이터 스토어 Uploads에 항목이 CSV 이름, 파일 경로 (Google 스토리지 용) 및 몇 가지 기본 정보가 추가됩니다. 그런 다음 "전처리"라 불리는 작업이 생성됩니다.
3. 전처리 작업은 수백만 개가 될 수있는 CSV 파일의 모든 행을 반복하며 각 행에 대해 CSV ID, 회선, 추출/추출 할 데이터가있는 UploadEntries 모델에 NDB 항목을 추가합니다. ("is_treating", "is_done")
4. 전처리 작업이 종료되면 정보가 클라이언트 "XXX lines"에 업데이트됩니다. 처리됩니다 "
5. Uploads.next()에 대한 호출이 이루어집니다. next 방법은 것입니다 :
   • 거짓에서 is_treating 및 is_done이있는 UploadEntries,
   • 찾은 다음 라인에 대한 레디 스 데이터 저장소에서 작업을 추가합니다 검색합니다.
   • Process-healthcheck 작업에 새 항목을 만듭니다 (이 작업은 5 분 후에 실행되고 7 번 항목이 올바르게 실행되었는지 확인합니다).이 작업은 올바르게 실행되었습니다.이 작업은 Google에서 관리하지 않는 서버에서 수행되므로 Redis 데이터 저장소가 사용됩니다. 그렇지 않은 경우 Redis/Outside 서버가 실패했으며 결과가없는 7)과 동일하게 처리합니다 (대신 "오류").
   • 그런 다음 해당 항목에 대해 UploadEntries.is_treating을 True로 업데이트합니다.
6. 외부 서버는 데이터를 처리하고 서버의 끝점으로 POST 요청을하여 결과를 반환합니다.
7. 해당 끝 점이 데이터 저장소의 UploadEntries 항목 ("is_treating"및 "is_done"포함)을 업데이트하고 다음 줄을 시작하려면 Uploads.next()으로 전화하십시오.
8. Uploads.next에서 다음 항목을 검색 할 때 아무 것도 반환되지 않습니다. 파일을 최종적으로 처리한다고 생각하고 처리 된 데이터로 CSV를 다시 작성하는 post-process 작업을 호출하여 고객에게 반환합니다. 내가 레디 스 마련했다 왜
   1. 실제 작업은 구글 AppEngine에 외부에있는 수행하는 서버, 그건 :

   ---

   여기서 명심해야 할 몇 가지 있습니다.

9. 현재 수행중인 방법으로 처리 할 병렬 엔트리 수에 유연성을 제공합니다. 5)에서 Uploads.next() 메서드는 인수를 포함하고 있기 때문에 n 프로세스를 병렬로 검색 할 수 있습니다. 1, 5, 20, 50 일 수 있습니다.
작업의 모든 줄을 Redis becase에 직접 추가 할 수는 없습니다. 다음 고객은 첫 번째 파일 처리가 완료되기를 기다려야합니다. 그러면이 작업이 너무 오래 걸릴 것입니다.

---

하지만이 시스템은 다양한 문제를 가지고, 나는 당신의 도움에 의존하고있어 그 이유는 :

1. 때때로,이 시스템은 데이터 저장소가 아직 제대로 때 업데이트되지 않도록 빠른 Uploads.next()을 호출하면 항목이 이미 처리 중입니다. (entry.is_treating = True은 데이터베이스에 아직 푸시되지 않았습니다.)
2. 처리가 완료되지 않은 상태에서 Redis 또는 내 서버 (실제로 모르겠 음) 언젠가는 작업이 느슨하거나 POST 요청이 느슨합니다. 작업은 절대로 is_done = True으로 이동하지 않습니다. 그래서 저는 Healcheck 시스템을 구현해야만했습니다. 라인이 올바르게 처리되었는지 확인하기 위해서입니다. 이것은 두 가지 이점이 있습니다 : 그 작업의 이름은 CSV ID와 라인을 포함합니다. 파일마다 고유하게 만듭니다. 데이터 저장소가 최신 상태가 아니며 동일한 작업이 두 번 실행되는 경우 동일한 이름이 이미 존재하므로 동의 확인 문제가 있음을 알리기 때문에 healthcheck 생성이 실패합니다. 따라서이 작업은 무시됩니다. 데이터 저장소가 아직 최신 상태가 아닙니다.

---

나는 라인으로 하나 개의 독립적 인 과정에서 파일, 라인을 실행하는 방법에 대한 생각 initiall하지만,이 병렬로 여러 줄을 실행 할 수 없다는 큰 단점이있다. 또한 Google은 전용 타겟 (기본값은 아님)에 대해 작업 실행을 24 시간으로 제한하고 파일이 실제로 클 때 24 시간 이상 실행할 수 있습니다.

: 도움이된다면
내용

, 내가

---

파이썬을 사용하고 그리고 워크 플로우를 단순화하기 위해, 여기에 내가 할 수있는 최선의 방법으로 달성하기 위해 노력하고있어입니다

• 큰 파일을 처리하고, 여러 개의 paralllel 프로세스를 한 행에 하나씩 실행하십시오.
• Redis를 사용하여 외부 서버로 작업을 보냅니다. 일단 외부 서버가 주 서버
• 메인 서버가 그 라인에 대한 정보를 업데이트에 POST 요청을 통해 결과를 반환하고 다음 줄

정말 사람이 있다면 감사하겠습니다가는 것을 수행 이렇게하는 더 좋은 방법이있었습니다. 나는 정말로 내가 그런 종류의 일을하는 첫번째 사람이 아니라고 믿는다. 그리고 나는 내가 올바르게 그것을하고 있지 않다라고 꽤 확신한다.

(Stackoverflow는 알고리즘 질문이기 때문에 Stack Exchange의 가장 좋은 섹션이라고 생각합니다.하지만 더 나은 네트워크를 찾지 못했을 수도 있습니다. 그렇다면 미안 해요. 그).

Answer 1

실제 작업이

가 대신 큰 파일을 처리 Google Cloud Dataflow를 사용하여 고려 가지고 구글 AppEngine에 외부에있는 않습니다 서버를? 파일 분할 및 처리를 관리하는 관리 서비스입니다.

• 사용자 업로드 파일 signed urls 또는 AppEngine에에서 요청이를 시작하는 작은 컴퓨팅 엔진 인스턴스를 시작
• Blob 저장소 API를 사용하여 클라우드 스토리지를 구글에 직접 : 여기 초기의 생각을 바탕으로

  는 윤곽 과정 차단 요청 (BlockingDataflowPipelineRunner)이 데이터 흐름 작업을 시작합니다. (샌드 박스 및 I/O 문제를 차단하기 때문에 컴퓨팅 인스턴스 여야합니다.)

• 데이터 흐름 작업이 완료되면 컴퓨팅 엔진 인스턴스의 차단이 해제되고 pubsub에 메시지가 게시됩니다.
• pubsub 메시지는 작업 상태를 '진행 중'에서 '완료'로 변경하는 AppEngine 서비스의 Webhook을 호출하므로 사용자가 결과를 가져올 수 있습니다.