Why won't my application start with pandas_udf and PySpark+Flask?

user_9411423 0
Viewed 428

When my Flask+PySpark application has a function with @udf or @pandas_udf annotation, it will not start. If I simply remove the annotation, it does start. If I try to start my application with Flask, the first pass of lexical interpretation of the script is executed. For example, the debugger stops at import lines such as from pyspark.sql.functions import pandas_udf, udf, PandasUDFType . However no statement is executed at all, including the initial app = Flask(name) statement. (Could it be some kind of hidden exception? ) If I start my application without Flask, with the same exact function and with the same imports, it does work.

These are the imports:

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

This is the function:

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

This is the statement that is not executed iff @pandas_udf is there:

app = Flask(__name__)

This is how IntelliJ starts 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

I'm running MacOS in my own computer.

pyspark pandas flask
1 Answers

I found the problem. The problem was that the @pandas_udf annotation required a Spark session at the time that the module is loaded (some kind of "first pass parsing" in Python). To solve the problem, I first called my code that creates a Spark session. Then I imported the module that has the function with the @pandas_udf annotation after. I imported it right inside the caller function and not at the header. To troubleshoot, I set a breakpoint over the @pandas_udf function (in PyCharm) and stepped into the functions. With that I could inspect the local variables. One of the variables referred to something like "sc" or "_jvm". I knew from a past problem that that happened if the Spark session was not initialized.

user_94114230
Related