wip: modify ingester to use processes in a 'fan-out' pattern #1672
Conversation
gustavobtflores
added
the
Ingester
gustavobtflores
force-pushed
the
refactor/ingester-processing-architecture
branch
from
78c9926 to
a6443b2
Compare
gustavobtflores
force-pushed
the
refactor/ingester-processing-architecture
branch
2 times, most recently
from
4f8b996 to
648d89b
Compare
gustavobtflores
force-pushed
the
refactor/ingester-processing-architecture
branch
2 times, most recently
from
07f5a91 to
1978947
Compare
| processed, | ||
| stat_ok, | ||
| stat_fail, | ||
| counter_lock, |
There was a problem hiding this comment.
Can you type those parameters? I see that you can import the types from multiprocessing.synchronize since the multiprocessing classes are defined there.
https://stackoverflow.com/questions/71988089/python-typing-for-multiprocessing-value
| ) | ||
|
|
||
| batch_len = len(batch) | ||
| if batch_len >= 100 or batch_len >= total_files_count: |
There was a problem hiding this comment.
this 100 should be a constant
| cycle_start = time.time() | ||
| total_bytes = 0 | ||
| for f in json_files: | ||
| total_files_count = len(json_files) | ||
|
|
||
| process_queue: multiprocessing.Queue[Optional[list[SubmissionFileMetadata]]] = ( | ||
| multiprocessing.Queue(maxsize=INGEST_QUEUE_MAXSIZE) | ||
| ) |
There was a problem hiding this comment.
the cycle_start, total_files_count and process_queue don't need to be moved up here, they could be defined after the first out() is called
There was a problem hiding this comment.
i'm using these variables in the loop right below
| % ( | ||
| total_files, | ||
| writers = [] | ||
| for _ in range(max_workers): |
There was a problem hiding this comment.
why remove the try-except for keyboard interrupt?
There was a problem hiding this comment.
forgot to add it after changes, fixed
| FILES_INGESTER_COUNTER.labels(INGESTER_GRAFANA_LABEL).inc() | ||
|
|
||
| return True | ||
| if any(len(instances_dict[table]) for table in instances_dict): # type: ignore |
There was a problem hiding this comment.
mypy complains about:
TypedDict key must be a string literal; expected one of ("issues", "checkouts", "builds", "tests", "incidents")Mypyliteral-required
gustavobtflores
force-pushed
the
refactor/ingester-processing-architecture
branch
from
1978947 to
b6b8376
Compare
2 participants
This PR reviews the ingester architecture to use processes both for file processing and database insertion, allowing faster processing using multiple database connections doing batch inserts at the same time.
How to test
poetry run python manage.py monitor_submissions [...options]Closes #1673