adrian-bialy / adrian-bialy/trading-data-warehouse
Refactor BaseIngestor.read() to process API requests on worker nodes
- Dominant language
- Python
- Stars
- 0
- Forks
- 1
- PR merge metrics
- No merged PRs in 30d
Description
Below method stores response on driver node as it's out of Spark's DataFrame context. There should be a method that processes such request within DataFrame on worker node. This is important as the requests might become huge or there will be multiple pages to process which can be evenly distributed on worker nodes instead of looping on 1 driver which may crash.
The most common approach is to use UDF that will be called in DataFrame.
```python
class BaseIngestor:
def read(self):
# ...
response = requests.get(
full_url,
headers=headers,
params=query_params,
timeout=self.source_config.get("variables", {}).get("timeout", 30),
)
```
**Acceptance criteria:**
- All API requests being processed on worker nodes (TODO: Process requests on worker nodes) [ ]
- Proper handling of succeeded/failed responses (TODO: Verify if API response is successful and handle errors) [ ]
- Proper pagination handling (TODO: Improve pagination handling - when to stop, how many pages to fetch, etc.) [ ]
- Documentation updated if needed [ ]
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.