Почему мое приложение не запускается с pandas_udf и PySpark + Flask?

Question

Если в моем приложении Flask + PySpark есть функция с аннотацией @udf или @pandas_udf, она не запустится. Если я просто удаляю аннотацию, она начинается. Если я пытаюсь запустить приложение с помощью Flask, выполняется первый этап лексической интерпретации сценария. Например, отладчик останавливается на строках импорта, таких как из pyspark.sql.functions import pandas_udf, udf, PandasUDFType , Однако ни один оператор не выполняется вообще, включая начальный оператор app = Flask ( name ). (Может ли это быть какое-то скрытое исключение?) Если я запускаю свое приложение без Flask, с той же самой функцией и с тем же импортом, оно работает.

Это импорт:

from pyspark.sql import SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf, udf, PandasUDFType
import pandas as pd

Это функция:

@pandas_udf('string', PandasUDFType.SCALAR)
def pandas_not_null(s):
    return s.fillna("_NO_NA_").replace('', '_NO_E_')

Это утверждение, которое не выполняется, если @pandas_udf находится там:

app = Flask(__name__)

Вот как IntelliJ запускает Flask:

FLASK_APP = app
FLASK_ENV = development
FLASK_DEBUG = 1
In folder /Users/vivaomengao/projects/dive-platform/cat-intel/divecatintel

/Users/vivaomengao/anaconda/bin/python /Applications/PyCharm.app/Contents/helpers/pydev/pydevd.py --module --multiproc --qt-support=auto --client 127.0.0.1 --port 56486 --file flask run

Я использую MacOS на своем собственном компьютере.

Answer 1

Я нашел проблему. Проблема заключалась в том, что аннотация @pandas_udf требовала сеанса Spark во время загрузки модуля (что-то вроде «первого анализа» в Python). Чтобы решить эту проблему, я сначала назвал свой код, который создает сеанс Spark. Затем я импортировал модуль, который имеет функцию с аннотацией @pandas_udf после. Я импортировал его прямо в функцию вызова, а не в заголовок. Чтобы устранить неполадки, я установил точку останова над функцией @pandas_udf (в PyCharm) и вошел в функции. С этим я мог проверить локальные переменные. Одна из переменных ссылается на что-то вроде «sc» или «_jvm». Из прошлой проблемы я знал, что это произошло, если сеанс Spark не был инициализирован.