Hi Im trying to use the basic PySpark example with ImageSpec

Question

Hi Im trying to use the basic PySpark example with ImageSpec when i run pyflyte run remote image imageSpec yaml flyte3 py my wf i get from web UI ``` 1 1 currentAttempt done Last Error USER core py 760 in invoke 760 return callback args kwargs usr local lib python3 10 site packages flytekit bin entrypoint py 508 in fast execute task cmd 508 subprocess run cmd check=True usr local lib python3 10 subprocess py 526 in run 526 raise CalledProcessError retcode process args CalledProcessError Command pyflyte execute inputs 6d n0 data inputs pb output prefix 6d n0 data 0 raw output data prefix checkpoint path prev checkpoint dynamic addl distro pt mode tar gz dynamic dest dir root resolver flytekit core python auto container default task resolver task module flyte3 task name hello spark returned non zero exit status 1 ``` Im not havving issues without remote flag pyflyte run flyte3 py my wf is running ok Id be soo grateful if you can guide me

Kevin Su · Answer

did you see any other error in the task pod

Kevin Su · Answer

kubectl logs lt pod name gt n flytesnacks development

Gabriel Molina · Answer

< Kevin Su> ``` bin bash warning setlocale LC ALL cannot change locale en US UTF 8 ++ id u + myuid=0 ++ id g + mygid=0 + set +e ++ getent passwd 0 + uidentry=root x 0 0 root root bin bash + set e + z root x 0 0 root root bin bash + z usr local openjdk 11 + SPARK CLASSPATH= opt spark jars + env + grep SPARK JAVA OPT + sort t k4 n + sed s ^= = 1 g + readarray t SPARK EXECUTOR JAVA OPTS + n + z + z + n + z + z + z x + SPARK CLASSPATH= opt spark conf opt spark jars + case $1 in Non spark on k8s command provided proceeding in pass through mode + echo Non spark on k8s command provided proceeding in pass through mode + CMD= $ + exec usr bin tini s pyflyte fast execute additional distribution dest dir root pyflyte execute inputs output prefix raw output data prefix checkpoint path prev checkpoint resolver flytekit core python auto container default task resolver task module flyte3 task name hello spark Traceback most recent call last File usr local bin pyflyte fast execute line 5 in lt module gt from flytekit bin entrypoint import fast execute task cmd File usr local lib python3 9 dist packages flytekit init py line 207 in lt module gt from flytekit core base sql task import SQLTask File usr local lib python3 9 dist packages flytekit core base sql task py line 4 in lt module gt from flytekit core base task import PythonTask TaskMetadata File usr local lib python3 9 dist packages flytekit core base task py line 34 in lt module gt from flytekit core interface import Interface transform interface to typed interface File usr local lib python3 9 dist packages flytekit core interface py line 14 in lt module gt from flytekit core type engine import TypeEngine File usr local lib python3 9 dist packages flytekit core type engine py line 24 in lt module gt from marshmallow enum import EnumField LoadDumpOptions ModuleNotFoundError No module named marshmallow enum ```

Kevin Su · Answer

you need to install marshmallow enum in the image spec

Kevin Su · Answer

cc < Eduardo Apolinario eapolinario > do we push the new flytekit image the latest image should contain marshmallow enum

Eduardo Apolinario (eapolinario) · Answer

yeah there s something weird going on I think it has to do with the alpha release we had last week cc < Yee>

Eduardo Apolinario (eapolinario) · Answer

driving to the office now will take a look when I get there

Yee · Answer

which version are you on

Yee · Answer

this was fixed already <https github com flyteorg flytekit pull 1746> basically an indirect dependency that stopped being an indirect dependency

Yee · Answer

try v1 8 1

Gabriel Molina · Answer

< Yee> was using 1 6 1 this is the tracebhack using 1 8 1 ```Traceback most recent call last File usr local lib python3 9 dist packages flytekit exceptions scopes py line 206 in user entry point return wrapped args kwargs File root flyte3 py line 33 in hello spark sess sparkContext parallelize range 1 n + 1 partitions map f reduce add File usr local lib python3 9 dist packages pyspark rdd py line 1905 in reduce vals = self mapPartitions func collect File usr local lib python3 9 dist packages pyspark rdd py line 1814 in collect sock info = self ctx jvm PythonRDD collectAndServe self jrdd rdd File usr local lib python3 9 dist packages pyspark rdd py line 5441 in jrdd wrapped func = wrap function File usr local lib python3 9 dist packages pyspark rdd py line 5243 in wrap function return sc jvm SimplePythonFunction Message JavaPackage object is not callable User error ```

Kevin Su · Answer

could you share your image spec

Gabriel Molina · Answer

< Kevin Su> sure ```python version 3 11 registry gmolinah92 packages pyspark env Debug True ```

Kevin Su · Answer

which version of flytekit are you using in the local venv

Gabriel Molina · Answer

1 8 1

Kevin Su · Answer

could you try to explicitly set spark base image in the image spec

Kevin Su · Answer

base image <http ghcr io flyteorg flytekit spark latest|ghcr io flyteorg flytekit spark latest>

Gabriel Molina · Answer

sure how can i do that should i run `docker pull <http ghcr io flyteorg flytekit py3 9 sqlalchemy 1 9 0a0|ghcr io flyteorg flytekit py3 9 sqlalchemy 1 9 0a0>`

Kevin Su · Answer

no envd will pull that automatically

Gabriel Molina · Answer

updated imageSpec ```python version 3 11 registry gmolinah92 packages pyspark env Debug True base image <http ghcr io flyteorg flytekit spark latest|ghcr io flyteorg flytekit spark latest>``` output ```Traceback most recent call last File usr local lib python3 9 dist packages flytekit exceptions scopes py line 206 in user entry point return wrapped args kwargs File root flyte3 py line 33 in hello spark sess sparkContext parallelize range 1 n + 1 partitions map f reduce add File usr local lib python3 9 dist packages pyspark rdd py line 1905 in reduce vals = self mapPartitions func collect File usr local lib python3 9 dist packages pyspark rdd py line 1814 in collect sock info = self ctx jvm PythonRDD collectAndServe self jrdd rdd File usr local lib python3 9 dist packages pyspark rdd py line 5441 in jrdd wrapped func = wrap function File usr local lib python3 9 dist packages pyspark rdd py line 5243 in wrap function return sc jvm SimplePythonFunction Message JavaPackage object is not callable User error ```

Kevin Su · Answer

we use this <https github com apache spark blob master resource managers kubernetes docker src main dockerfiles spark bindings python Dockerfile|spark> image as our base image + flytekit + flytekitplugins

Kevin Su · Answer

do you use any jar that isn t installed in the spark base image

Gabriel Molina · Answer

thanks kevin no i m just trying the provided documentation example but we plan to use jdbc jar

Kevin Su · Answer

That might have some issues in the image investigating

Gabriel Molina · Answer

just in case this is the content of my flyte3 py file ```import datetime import random from operator import add import flytekit from flytekit import Resources task workflow from flytekitplugins spark import Spark from flytekit image spec image spec import ImageSpec spark image = ImageSpec registry= gmolinah92 task task config=Spark this configuration is applied to the spark cluster spark conf= spark driver memory 1000M spark executor memory 1000M spark executor cores 1 spark executor instances 2 spark driver cores 1 limits=Resources mem= 2000M cache version= 1 container image=spark image def hello spark partitions int gt float print Starting Sparkfk wifth Partitions format partitions n = 100000 partitions sess = flytekit current context spark session count = sess sparkContext parallelize range 1 n + 1 partitions map f reduce add pi val = 4 0 count n print Pi val is format pi val return pi val def f x = random random 2 1 y = random random 2 1 return 1 if x 2 + y 2 lt = 1 else 0 task cache version= 1 container image=spark image def print every time value to print float date triggered datetime datetime gt int print My printed value format value to print date triggered return 1 workflow def my wf triggered date datetime datetime = datetime datetime now gt float Using the workflow is still as any other workflow As image is a property of the task the workflow does not care about how the image is configured pi = hello spark partitions=50 print every time value to print=pi date triggered=triggered date return pi```

Gabriel Molina · Answer

how should i use that dockerfile can you please share suggested documentation

Kevin Su · Answer

thanks for sharing let me test it

Kevin Su · Answer

could you try `pyflyte register version v1 non fast flyte3 py` without passing imageSpec yaml

Gabriel Molina · Answer

successfully registered ```Running pyflyte register from home gabriel Documents with images ImageConfig default image=Image name= default fqn= tag= py3 10 1 8 1 images= Image name= default fqn= tag= py3 10 1 8 1 and image destination folder root on 1 package s home gabriel Documents flyte3 py Registering against localhost 30080 Detected Root home gabriel Documents using this to create deployable package Loading packages flyte3 under source root home gabriel Documents Image gmolinah92 flytekit Df4gvCXxRN3XkjdLXzPEVg found Skip building Image gmolinah92 flytekit Df4gvCXxRN3XkjdLXzPEVg found Skip building Successfully serialized 4 flyte objects Registration flyte3 hello spark type TASK successful with version v1 Registration flyte3 print every time type TASK successful with version v1 Registration flyte3 my wf type WORKFLOW successful with version v1 Registration flyte3 my wf type LAUNCH PLAN successful with version v1 Successfully registered 4 entities``` but then run remote generates same error did you tried it successfully ```Traceback most recent call last File usr local lib python3 9 dist packages flytekit exceptions scopes py line 206 in user entry point return wrapped args kwargs File root flyte3 py line 33 in hello spark sess sparkContext parallelize range 1 n + 1 partitions map f reduce add File usr local lib python3 9 dist packages pyspark rdd py line 1905 in reduce vals = self mapPartitions func collect File usr local lib python3 9 dist packages pyspark rdd py line 1814 in collect sock info = self ctx jvm PythonRDD collectAndServe self jrdd rdd File usr local lib python3 9 dist packages pyspark rdd py line 5441 in jrdd wrapped func = wrap function File usr local lib python3 9 dist packages pyspark rdd py line 5243 in wrap function return sc jvm SimplePythonFunction Message JavaPackage object is not callable User error ```

Kevin Su · Answer

hmm it works for me what image you re using for the spark task

Gabriel Molina · Answer

`gmolinah92 flytekit jauyJJTJIIQjfoWzFf3YaQ `

Kevin Su · Answer

could you try my image `pingsutw flytekit 4dRZqjpM2KKZ6XNRUL76ew `

Kevin Su · Answer

it works for me

Gabriel Molina · Answer

How can i use it

Kevin Su · Answer

pyflyte run image pingsutw flytekit 4dRZqjpM2KKZ6XNRUL76ew wf py my wf

Gabriel Molina · Answer

Apparently its needed to change `spark image = ImageSpec registry= gmolinah92 ` to `spark image = ImageSpec registry= pingsutw ` in flyte3 py Even that way running run remote is using a different image than specified both registry users as you can see in attached image Both executions failed with same previous error