ThreadPoolExecutor для параллелизма ⇐ Python

[Anonymous]

У меня есть код PySpark, который выполняет несколько вызовов POST API во внешнюю систему. Для каждой строки входного фрейма данных мне нужно запустить запрос POST API (с использованием кода Python), чтобы создать запись во внешней системе. Учитывая большой размер набора данных, этот процесс занимал значительное время.
Чтобы повысить производительность, запланируйте использование Python ThreadPoolExecutor для параллельной обработки строк (т. е. для POST API) (многопоточная обработка). ) на основе доступных ядер.

Код: Выделить всё

from concurrent.futures import ThreadPoolExecutor, as_completed

num_cores = spark.sparkContext.defaultParallelism

def process_all_rows(input_df):
results = []

with ThreadPoolExecutor(max_workers=num_cores) as executor:  # Adjust max_workers based on needs
futures = {executor.submit(process_row, row): row for row in input_df.collect()}

for future in as_completed(futures):
try:
result = future.result()
results.append(result)
except Exception as e:
logger.error(f"Error in thread execution: {e}")

return results

Просматривая это, мне сказали, что ThreadPoolExecutor в основном выполняет переключение контекста. Таким образом, если входной DataFrame имеет 100 строк, а для параметра num_cores установлено значение 8 (т. е. кластер имеет 8 ядер), код будет использовать только одно ядро ​​(не все доступные 8 ядер), последовательно запуская запросы POST с переключением контекста, т. е. запуск одного запроса POST API, затем следующего и так далее. Это правильное понимание? Будет ли ThreadPoolExecutor использовать все 8 ядер параллельно?

Подробнее здесь: https://stackoverflow.com/questions/791 ... arallelism