Используйте потоки для UDF в данных DataBricks, чтобы использовать новый токен API ⇐ Python

[Anonymous]

У меня есть дата, который необходимо использовать по строке для вызовов API. Поэтому я реализовал UDF для этого. Каждая строка использует данные из этой строки, чтобы сделать вызов API, и сохраняет идентификатор из ответа в результате DataFrame. Вызов выглядит примерно так: < /p>

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

processDf = FilesDf.withColumn("processed", row_udf(col("input_id")).select("processed.*").cache() 

Теперь проблема заключается в том, что обработка занимает более одного часа, а заголовки, которые я передаю в ROW_UDF для вызова API, содержат токен, который действителен только в течение часа. Я установил заголовки в ячейке на вершине ноутбука.

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

def get_new_token():
app = TokenApplication(...)
result = app.acquire_token_for_client(...)
...
headers = {"Authorization": token}
return headers

headers = get_new_token()

# Create a threading event to signal the thread to stop
stop_event = threading.Event()

def refresh_token_periodically():
global headers
while not stop_event.is_set():
time.sleep(55 * 60)  # Sleep for 55 minutes
headers = get_new_token()

# Start the background thread to refresh the token
token_refresh_thread = threading.Thread(target=refresh_token_periodically)
token_refresh_thread.daemon = True
token_refresh_thread.start()

Я проверил это с простыми операторами печати/сна в новой ячейке и увидел, что токен обновляется. Теперь, используя его внутри UDF, не работает.
Мне интересно, почему. Все строки «подготовлены» сразу и сложены в линию ожидания, чтобы начало всех рядов на самом деле одинаково, и они получают первый жетон, но более поздние ряды действительно передаются только позже, что затем приводит к неправильному токену?>

Подробнее здесь: https://stackoverflow.com/questions/796 ... -api-token