adrian-bialy / adrian-bialy/trading-data-warehouse

Refactor BaseIngestor.read() to process API requests on worker nodes

Open
#2 0 comments 0 reactions 0 assignees View on GitHub
enhancement
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.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.