Блог

В разделе «Сведения о задании n» пользовательский интерфейс показывает «Статус: УСПЕШНО», однако на одном из этапов показано 22029 успешно выполненных задач из 59400 общих задач. Я запускаю это через ноутбук Python Jupyter под управлением Spark 3.0.1, и я еще не остановил контекст Spark, поэтому приложение все еще работает. Фактически, вкладка Этапы показывает, что рассматриваемый этап все еще активен. Я не понимаю, как этап все еще может быть активным, но задание указано в пользовательском интерфейсе как завершенное и успешное.

Соответствующий код (я думаю) приведен ниже, где я пытаюсь распараллелить как можно больше SQL-запросов, а затем объединить результирующие фреймы данных вместе. Наконец, я записываю их в облачное хранилище в parquet.

РЕДАКТИРОВАТЬ: я также могу видеть ту же информацию из REST API, используя конечные точки, описанные здесь в документах, и эти значения такие же, как я вижу в веб-интерфейсе.

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

Я видел в журналах много случаев Из Dropping event from queue appStatus. This likely means one of the listeners is too slow and cannot keep up with the rate at which tasks are being started by the scheduler. -за этого, я экспериментирую с параметром spark.scheduler.listenerbus.eventqueue.capacity , чтобы увеличить его и посмотреть, приведет ли это к разнице в отчетах об успешных и общих задачах для этого этапа.

При увеличении spark.scheduler.listenerbus.eventqueue.capacity значения по умолчанию с 10000 до 65000, по-видимому, наблюдается соответствующее уменьшение количества отброшенных событий, а также увеличение числа успешно выполненных задач, о которых сообщалось на этом этапе, до ~ 47 тыс. с ~ 22 тыс. Я также заметил, что разница в выполненных и общих задачах для этого этапа составляет порядка количества отброшенных событий в журнале, поэтому я посмотрю, может ли ограничение отброшенных событий устранить несоответствие.

 def make_df(query: str):
    df = spark.sql(query)
    return df

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

df_list = list(map(make_df, queries))

df = functools.reduce(lambda x, y: x.union(y), df_list)

df.repartition("col1", "col2")
    .write.partitionBy("col1", "col2")
    .mode("overwrite")
    .parquet(path)
 

Почему моя работа должна отображаться как успешная, если все еще остаются задачи, которые не выполнены успешно?