Hi I am trying to return a spark data frame but having this

Question

Hi I am trying to return a spark data frame but having this issue user exceptions FlyteAssertion flytekit exceptions user FlyteAssertion When calling tasks only keyword args are supported Aborting execution as detected 1 positional args FlyteSchema class getitem lt locals gt TypedSchema remote path= tmp flyte 20220322 204000 raw 82eaa7c6b77680e044d794beaf4e01cb I have defined column names with their data types as my schema = FlyteSchema kwtypes ID=str START T=datetime datetime STOP=datetime datetime PATIENT=str ENCOUNTER=str CODE=str intDESCRIPTION=str REASONCODE=int REASONDESCRIPTION=str Following this <https docs flyte org projects cookbook en latest auto integrations kubernetes k8s spark dataframe passing html> guide

Ketan (kumare3) · Answer

Can you share a code snippet

komal azram · Answer

```my schema = FlyteSchema kwtypes ID=str START T=datetime datetime STOP=datetime datetime PATIENT=str ENCOUNTER=str CODE=str intDESCRIPTION=str REASONCODE=int REASONDESCRIPTION=str task def spark start gt my schema spark = SparkSession builder master local appName snowflake test config spark jars jar snowflake jdbc 3 13 15 jar jar spark snowflake 2 12 2 10 0 spark 3 2 jar getOrCreate df = spark read format SNOWFLAKE SOURCE NAME options sfOptions option query SELECT FROM CAREPLANS load return df```

Ketan (kumare3) · Answer

This is not the problem

Ketan (kumare3) · Answer

The workflow

komal azram · Answer

haven t assigned workflow to it yet

Yee · Answer

how are you calling this task

Yee · Answer

can you paste the code where you call it

komal azram · Answer

```if name == main df=spark start df expect df ```

Yee · Answer

`df expect df=df `

komal azram · Answer

AttributeError TypedSchema object has no attribute persist

komal azram · Answer

getting this error now

Yee · Answer

can you show the whole stack trace

komal azram · Answer

asctime 2022 03 22 21 20 35 124 name flytekit levelname ERROR message Exception when executing TypedSchema object has no attribute persist exc info Traceback most recent call last n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 478 in dispatch execute n native outputs = self execute native inputs n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core python function task py line 160 in execute n return exception scopes user entry point self task function kwargs n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit exceptions scopes py line 198 in user entry point n return wrapped args kwargs n File home komal PycharmProjects spark ge main py line 46 in df expect n batch=SparkDFDataset df n File home komal PycharmProjects spark ge venv lib python3 8 site packages great expectations dataset sparkdf dataset py line 617 in init n self spark df persist nAttributeError TypedSchema object has no attribute persist Traceback most recent call last File home komal PycharmProjects spark ge main py line 64 in lt module gt df expect df=df File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 288 in call return flyte entity call handler self args kwargs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core promise py line 991 in flyte entity call handler result = cast LocallyExecutable entity local execute child ctx kwargs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 269 in local execute outputs literal map = self dispatch execute ctx input literal map File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 481 in dispatch execute raise e File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 478 in dispatch execute native outputs = self execute native inputs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core python function task py line 160 in execute return exception scopes user entry point self task function kwargs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit exceptions scopes py line 198 in user entry point return wrapped args kwargs File home komal PycharmProjects spark ge main py line 46 in df expect batch=SparkDFDataset df File home komal PycharmProjects spark ge venv lib python3 8 site packages great expectations dataset sparkdf dataset py line 617 in init self spark df persist AttributeError TypedSchema object has no attribute persist

Yee · Answer

< Samhita Alla> do you think you could help take a look i m not as familiar with great expectations code

Samhita Alla · Answer

Sure < komal azram> would you mind sharing me the `df expect ` task code

Samhita Alla · Answer

I think you may have to call `open ` on your spark dataframe within the `df expect ` code something like `df open pyspark DataFrame all `

komal azram · Answer

```def df expect df my schema gt str batch=SparkDFDataset df ID Null=batch expect column values to not be null ID ID res=print Null values not found if ID Null success else NUll values return ID res```

Samhita Alla · Answer

Can you replace `batch=SparkDFDataset df ` with `batch=SparkDFDataset df open pyspark DataFrame all `

komal azram · Answer

AttributeError module pyspark has no attribute dataframe

Samhita Alla · Answer

Sorry can you try `pyspark sql DataFrame` < U02TU5RR0V8>

komal azram · Answer

AssertionError SPARK SESSION not available as a parameter in Flyte context are you in right task type

Samhita Alla · Answer

Are you using `task config=Spark ` for your task Example ``` task task config=Spark spark conf= spark driver memory 1000M spark executor memory 1000M spark executor cores 1 spark executor instances 2 spark driver cores 1 cache version= 1 def create spark df gt my schema ```

komal azram · Answer

No

komal azram · Answer

is it mandatory

Samhita Alla · Answer

Oh you ll have to to initialize the Spark session You can see <https docs flyte org projects cookbook en latest auto integrations kubernetes k8s spark dataframe passing html> example the one you re already referring to

Samhita Alla · Answer

Also refer to <https docs flyte org projects cookbook en latest auto case studies ml training spark horovod keras spark rossmann estimator html> for a more involved approach

komal azram · Answer

Thankyou < Samhita Alla>

komal azram · Answer

Hey I have added task config variables as mentioned in the tutorials but it gives following error

komal azram · Answer

raise AssertionError f attr name not available as a parameter in Flyte context are you in right task type AssertionError SPARK SESSION not available as a parameter in Flyte context are you in right task type

komal azram · Answer

In my use case I have to build a spark session and connect snowflake and spark to query data from snowflake

komal azram · Answer

I have added the task configuration in the task where I am starting the spark session

Samhita Alla · Answer

This error has to do with configuring Spark Can you paste the stacktrace I d like to see where exactly this error is popping up

komal azram · Answer

asctime 2022 03 23 14 40 00 136 name flytekit levelname ERROR message Exception when executing SPARK SESSION not available as a parameter in Flyte context are you in right task type exc info Traceback most recent call last n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 478 in dispatch execute n native outputs = self execute native inputs n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core python function task py line 160 in execute n return exception scopes user entry point self task function kwargs n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit exceptions scopes py line 198 in user entry point n return wrapped args kwargs n File home komal PycharmProjects spark ge main py line 58 in df expect n batch = SparkDFDataset df open spark sql DataFrame all n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekitplugins spark schema py line 27 in all n return ctx spark session read parquet self from path n File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core context manager py line 313 in getattr n raise AssertionError f attr name not available as a parameter in Flyte context are you in right task type nAssertionError SPARK SESSION not available as a parameter in Flyte context are you in right task type Traceback most recent call last File home komal PycharmProjects spark ge main py line 81 in lt module gt df expect df=df File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 288 in call return flyte entity call handler self args kwargs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core promise py line 991 in flyte entity call handler result = cast LocallyExecutable entity local execute child ctx kwargs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 269 in local execute outputs literal map = self dispatch execute ctx input literal map File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 481 in dispatch execute raise e File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core base task py line 478 in dispatch execute native outputs = self execute native inputs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core python function task py line 160 in execute return exception scopes user entry point self task function kwargs File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit exceptions scopes py line 198 in user entry point return wrapped args kwargs File home komal PycharmProjects spark ge main py line 58 in df expect batch = SparkDFDataset df open spark sql DataFrame all File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekitplugins spark schema py line 27 in all return ctx spark session read parquet self from path File home komal PycharmProjects spark ge venv lib python3 8 site packages flytekit core context manager py line 313 in getattr raise AssertionError f attr name not available as a parameter in Flyte context are you in right task type AssertionError SPARK SESSION not available as a parameter in Flyte context are you in right task type Process finished with exit code 1

Samhita Alla · Answer

Are you running this locally or in a configured Flyte environment with Spark backend

Samhita Alla · Answer

Can you also share me the whole code wherever you re initializing the task config

komal azram · Answer

running it locally

komal azram · Answer

give me a min

komal azram · Answer

```import datetime from flytekit import task kwtypes from typing import Tuple import pyspark as spark from pyspark sql import SparkSession from pyspark sql types import from great expectations dataset import SparkDFDataset from flytekitplugins spark import Spark from flytekit types schema import FlyteSchema my schema = FlyteSchema kwtypes ID=str START T=datetime datetime STOP=datetime datetime PATIENT=str ENCOUNTER=str CODE=str intDESCRIPTION=str REASONCODE=int REASONDESCRIPTION=str task task config=Spark spark conf= spark driver memory 1000M spark executor memory 1000M spark executor cores 1 spark executor instances 2 spark driver cores 1 cache version= 1 def spark start gt my schema spark = SparkSession builder master local appName snowflake test config spark jars jar snowflake jdbc 3 13 15 jar jar spark snowflake 2 12 2 10 0 spark 3 2 jar getOrCreate f = spark read format SNOWFLAKE SOURCE NAME options sfOptions option query SELECT FROM CAREPLANS load return df task def df expect df my schema gt str df1 =df open df all batch = SparkDFDataset df open spark sql DataFrame all batch=SparkDFDataset df1 ID Null=batch expect column values to not be null ID ID res=print Null values not found if ID Null success else NUll values CODE Null=batch expect column values to not be null CODE CODE res=print Null values not found if CODE Null success else NUll values DESCRIPTION NUll=batch expect column values to not be null DESCRIPTION DESCRIPTION res=print Null values not found if DESCRIPTION NUll success else NUll values ID Unique= batch expect column values to be unique ID Unique res=print Column has Unique values if ID Unique success else duplicate values return ID res CODE res DESCRIPTION res Unique res if name == main df=spark start df expect df=df ```

Samhita Alla · Answer

In the `spark start` task can you replace `SparkSession` with `sess` ``` sess = flytekit current context spark session ```

komal azram · Answer

```def spark start gt my schema spark = flytekit current context sparkSession builder master local appName snowflake test config spark jars jar snowflake jdbc 3 13 15 jar jar spark snowflake 2 12 2 10 0 spark 3 2 jar getOrCreate ```

komal azram · Answer

you mean like that

Samhita Alla · Answer

Yes ``` sess = flytekit current context spark session spark = sess builder ```

komal azram · Answer

sertionError SPARK SESSION not available as a parameter in Flyte context are you in right task type

komal azram · Answer

still getting same error

Samhita Alla · Answer

Not sure why that s happening Can you share me your code once again

Samhita Alla · Answer

Can you merge both of them into a single task I think because we re trying to open our dataframe through the Spark mode it needs that session to be available

komal azram · Answer

ok

Samhita Alla · Answer

Thank you being patient Let me know after you run your modified code

komal azram · Answer

TypeError Failed to convert return value for var o0 for function main spark start with error lt class AssertionError gt Python value cannot be None expected lt class str gt simple STRING I got this error when I have returned one variable that contains print statement and when I have returned all three variables it gives following error TypeError Output o0 in task main spark start received a tuple None None None None instead of lt class str gt

komal azram · Answer

although I am getting results

Samhita Alla · Answer

< komal azram> in the `df expect` task the output datatype is `str` when it has to be a tuple with four values like `Tuple str ` You can use NamedTuple to assign names to the outputs Also rather than returning print statements you ll have to return the actual value cause `print` doesn t return anything

komal azram · Answer

it s working now

komal azram · Answer

but when I add workflow it gives following error TypeError new missing 3 required positional arguments CODE res DESCRIPTION res and Unique res ```wf outputs = typing NamedTuple op2 ID res=str CODE res=str DESCRIPTION res=str Unique res=str workflow def my wf gt wf outputs return wf outputs spark start if name == main print my wf ```

Samhita Alla · Answer

Can you call the `spark start` task like `return wf outputs spark start ` Also you can unbox the outputs like ```def my wf gt wf outputs task exec = spark start return wf outputs task exec ID res task exec CODE rest ``` considering your task is returning the same NamedTuple

komal azram · Answer

Thankyou for helping < Samhita Alla> it s working now