0

Появилась задача пофиксить баг на новой работе, пока ещё во все процессы не вник и встрял с этой задачей.

Airflow использует LivySessionOperator для того чтобы запустить PySpark внутри livy-сессии который заберёт данные из топика кафки и загрузит их в HDFS.

Если смотреть по логам внутри Airflow, краткий ответ выглядит так:

{
  "status": "error",
  "execution_count": 0,
  "ename": "StreamingQueryException",
  "evalue": "'Job aborted.\
=== Streaming Query ===\
Identifier: [id = b5e02046-7ec0-4588-9188-1f8c96526e04, runId = f2babd9f-79f1-4e8c-8efd-5227e61707d9]\
Current Committed Offsets: {KafkaV2[Subscribe[rosgidromet_grib]]: {\"rosgidromet_grib\":{\"2\":6005318,\"1\":6011947,\"0\":6010359}}}\
Current Available Offsets: {KafkaV2[Subscribe[rosgidromet_grib]]: {\"rosgidromet_grib\":{\"2\":6007818,\"1\":6014446,\"0\":6012859}}}\
\
Current State: ACTIVE\
Thread State: RUNNABLE\
\
Logical Plan:\
Repartition 1, true\
+- Filter ((isnotnull(filename#26) && isnotnull(bands#31)) && isnotnull(rast#35))\
   +- Project [dt#21, ts#22, filename#26, rast#25.gdalinfo AS bands#31, rast#35]\
      +- Generate explode(rast#25.rast), false, [rast#35]\
         +- Project [cast(timestamp#12 as date) AS dt#21, timestamp#12 AS ts#22, cast(key#7 as string) AS filename#26, openGDAL(value#8) AS rast#25]\
            +- StreamingExecutionRelation KafkaV2[Subscribe[rosgidromet_grib]], [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]\
'",
  "traceback": [
    "Traceback (most recent call last):
",
    "  File \"/srv/hadoop-yarn/nm-local/usercache/airflow/appcache/application_1668513493768_5742/container_e1677_1668513493768_5742_01_000001/pyspark.zip/pyspark/sql/streaming.py\", line 103, in awaitTermination
    return self._jsq.awaitTermination()
",
    "  File \"/srv/hadoop-yarn/nm-local/usercache/airflow/appcache/application_1668513493768_5742/container_e1677_1668513493768_5742_01_000001/py4j-0.10.7-src.zip/py4j/java_gateway.py\", line 1257, in __call__
    answer, self.gateway_client, self.target_id, self.name)
",
    "  File \"/srv/hadoop-yarn/nm-local/usercache/airflow/appcache/application_1668513493768_5742/container_e1677_1668513493768_5742_01_000001/pyspark.zip/pyspark/sql/utils.py\", line 75, in deco
    raise StreamingQueryException(s.split(': ', 1)[1], stackTrace)
",
    "pyspark.sql.utils.StreamingQueryException: 'Job aborted.\

Я не очень понимаю что конкретно случилось по traceback, но ниже есть более полные логи самой сессии (но без ошибки спарка). Последнее что происходит это сообщение от yarn об ожидании ресурсов:

INFO yarn.Client: 
INFO -   client token: N/A
INFO -   diagnostics: [Fri Feb 03 10:00:07 +0300 2023] Application is Activated, waiting for resources to be assigned for AM.  Details : AM Partition = <DEFAULT_PARTITION> ; Partition Resource = <memory:958464, vCores:156> ; Queue's Absolute capacity = 50.0 % ; Queue's Absolute used capacity = 0.0 % ; Queue's Absolute max capacity = 100.0 % ; Queue's capacity (absolute resource) = <memory:479232, vCores:78> ; Queue's used capacity (absolute resource) = <memory:0, vCores:0> ; Queue's max capacity (absolute resource) = <memory:958464, vCores:156> ; 
INFO -   ApplicationMaster host: N/A
INFO -   ApplicationMaster RPC port: -1
INFO -   queue: batch
INFO -   start time: 1675407607629
INFO -   final status: UNDEFINED
INFO -   tracking URL: (ради приличия url убрал)
user: airflow

И далее сразу в логах происходит:

INFO util.ShutdownHookManager: Shutdown hook called

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

Далее в поисках ошибки перешёл на саму страницу с Application, там все статусы FINISHED или SUCCEEDED, но как-будто бы задача не потребляла совсем никаких ресурсов application parameters

Можете подсказать в какую сторону копать чтобы починить проблему? По словам коллег из смежного отдела эта ошибка стала проявляться после того как у них однажды забился кластер из-за нехватки памяти и всё легло.

2
  • Выглядит как нехватка ресурсов. "По словам коллег из смежного отдела эта ошибка стала проявляться после того как у них однажды забился кластер из-за нехватки памяти и всё легло." - может пока лежал, в очереди накопилось много данных, с которыми кластер теперь не справляется? 10 фев 2023 в 15:49
  • 1
    Я очень долго разбирался в чём проблема, оказалось нечто странное и тривиальное, каким-то образом оказалось что у спарка слетел offset и пытался обратиться к файлу с индексом 0 которого уже как пару лет в кафке не было Поменяли оффсет и стало грузиться 12 фев 2023 в 13:08

0

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Посмотрите другие вопросы с метками или задайте свой вопрос.