when using the pyspark decorator, do i need to put pyspark as a dependency of my flow? it doesn't seem to be declared as a dependency in the metaflow-pyspark package

Yep, I believe you'll need it in the environment of the @step that @pyspark annotates.

the example seems to run without it though? so is it actually necessary?

@victorious-lawyer-58417 can you fact check me here? what is the intention with regard to deps on your example?

well i dunno if it's related, but i'm seeing something about the following. the job starts but then it's crashing on the

df.rdd.map

line

affinity = df.rdd.map(lambda x: map_affinity_to_propensity(current_timestamp,x)).toDF()
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 115, in toDF
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 1276, in createDataFrame
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 1316, in _create_dataframe
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 931, in _createFromRDD
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 874, in _inferSchema
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2869, in first
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2836, in take
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/context.py", line 2322, in runJob
  File "/usr/lib/spark/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/errors/exceptions/captured.py", line 169, in deco
  File "/usr/lib/spark/python/lib/py4j-0.10.9.7-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1.0 (TID 4) ([2600:1f18:6b7e:901:68d2:4953:bd43:cbb6] executor 12): org.apache.spark.SparkException: 
Bad data in pyspark.daemon's standard output. Invalid port number:
  538976288 (0x20202020)
Python command to execute the daemon was:
  /home/hadoop/metaflow-python.sh -m pyspark.daemon
Check that you don't have any unexpected modules or libraries in
your PYTHONPATH:
  /usr/lib/spark/python/lib/pyspark.zip:/usr/lib/spark/python/lib/py4j-0.10.9.7-src.zip:/usr/lib/spark/jars/spark-core_2.12-3.4.0-amzn-0.jar

btw, are you using Spark Serverless? did you get it working?

yea using EMR serverless, and it's working now. The example was always working, but when I tried writing some of my own pyspark functions it was throwing the error above

ok, good to hear it's working now. Suggestions for (other) improvements (too) are welcome!

there are a few ways this EMR workflow can be streamlined: 1. setting up correct permissions on the ECS role to include emr-serverless permissions etc in the CFN template 2. get rid of the custom Docker image which I'm currently using, and instead use something like

spark.archives

to bundle the trampoline code (see my PR for how it currently works) 3. set all of the relevant environment variables via the extension (I'm currently doing it manually)

ack, good points. Will take a look at the PR. Env vars should be easily doable (hopefully)