topic Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas() in Data Engineering

Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Sat, 02 Jul 2022 11:27:46 GMT

I wrote the following code:

data = spark.sql (" SELECT A_adjClose, AA_adjClose, AAL_adjClose, AAP_adjClose, AAPL_adjClose FROM deltabase.a_30min_delta, deltabase.aa_30min_delta, deltabase.aal_30min_delta, deltabase.aap_30min_delta ,deltabase.aapl_30min_delta ")
 
 
 
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
 
 
 
#This part keeps running command
 
data_pd = data.toPandas()
 
 
 
df_pct = data_pd.pct_change(1)

The code stucks in .toPandas() this part.

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Sat, 02 Jul 2022 11:28:52 GMT

Btw, this is Azure Databricks

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Hubert-Dudek — Sun, 03 Jul 2022 18:19:50 GMT

Try to replace .to_pandas() with .to_pandas_on_spark(). This way, you will ensure that the dataframe is processed distributed among the workers.

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Mon, 04 Jul 2022 20:59:54 GMT

I tried to replace .to_pandas() with .to_pandas_on_spark(), but there were 1 warning message and 1 error message:

/databricks/spark/python/pyspark/sql/dataframe.py:3407: FutureWarning: DataFrame.to_pandas_on_spark is deprecated. Use DataFrame.pandas_api instead.

warnings.warn(

SparkException: Job aborted due to stage failure: Task 0 in stage 21.0 failed 4 times, most recent failure: Lost task 0.3 in stage 21.0 (TID 24) (10.139.64.4 executor 3): ExecutorLostFailure (executor 3 exited caused by one of the running tasks) Reason: Executor heartbeat timed out after 176959 ms

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

User16753725469 — Thu, 07 Jul 2022 13:39:47 GMT

Hi @Cheuk Hin Christophe Poon can you please try increasing executors if it helps

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Fri, 08 Jul 2022 09:40:00 GMT

@Prasad Wagh @Kaniz Fatma I use Standard_DS3_V2 (14GB Memory 4 Cores) in Azure Databricks. Originally, I set Min Worker: 1 Max Workers: 8. Now, Min Worker: 1 Max Worker: 8.

But this still doesn't work. 1 worker should be able to finish the task. The data size of the task is small.

I guess the size of the VM cluster is not the cause.

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Hubert-Dudek — Fri, 08 Jul 2022 16:49:59 GMT

As Spark is using lazy evaluation, I bet that is not to_pands which case the issues but pct_change as stated in the quote from the documentation below. Also is better that panda datset has the unique index to run pct_change (if you have no time field, you can set increment id):

df.set_index('month')

"the current implementation of this API uses Spark’s Window without specifying partition specification. This leads to move all data into single partition in single machine and could cause serious performance degradation. Avoid this method against very large dataset."

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Sat, 09 Jul 2022 03:51:04 GMT

data = spark.sql (" SELECT A_adjClose, AA_adjClose, AAL_adjClose, AAP_adjClose, AAPL_adjClose FROM deltabase.a_30min_delta, deltabase.aa_30min_delta, deltabase.aal_30min_delta, deltabase.aap_30min_delta ,deltabase.aapl_30min_delta ")
 
display(data)
 
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
 
data_pd = data.toPandas()
 
#df_pct = data_pd.pct_change(1)
 
#display(df_pct)

@Hubert Dudek I don't think the problem is the data type.

In my original code, there was a date/time, but I am debugging now. And I realized removing the date/time column doesn't solve the problem.

Now, the data types of the data are just float.

Also, I removed pct_change. The problem still exists.

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Sat, 09 Jul 2022 04:09:47 GMT

@Prasad Wagh @Kaniz Fatma Is it possible to submit a full detailed log report to Databricks?

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Hubert-Dudek — Sat, 09 Jul 2022 12:14:12 GMT

How many rows does the dataset have?
Can you share excel from the display function with the sample?
What are the cluster-specific (worker type and runtime version)? Is it standard, high-concurrent, or single-machine?

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Sat, 09 Jul 2022 17:04:04 GMT

Each Delta table has about more than 100,000 rows, but each Delta table only has about 3.18 MB.

I upload the photo and there is a sample, but not using the python display() function.

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Sat, 09 Jul 2022 17:07:48 GMT

Cluster Model: Standard. Runtime version: 11.0(Includes Apache Spark 3.3.0, Scala 2.12). Worker Type: Standard_DS3_V2.

Spark Config:

spark.databricks.delta.autoCompact.enabled true
spark.databricks.delta.optimizeWrite.enabled true

Re: Failed to convert Spark.sql to Pandas Dataframe using .toPandas()

Dicer — Tue, 19 Jul 2022 06:39:47 GMT

I just discovered a solution.

Today, I opened Azure Databricks. When I imported python libraries. Databricks told me that toPandas() was deprecated and it suggested me to use toPandas.

The following solution works: Use toPandas instead of toPandas()

data = spark.sql (" SELECT A_adjClose, AA_adjClose, AAL_adjClose, AAP_adjClose, AAPL_adjClose FROM deltabase.a_30min_delta, deltabase.aa_30min_delta, deltabase.aal_30min_delta, deltabase.aap_30min_delta ,deltabase.aapl_30min_delta ")
 
display(data)
 
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
 
data_pd = data.toPandas

Yet, when I tried to calculate percentage change using pct_change(), it didn't work. pct_change() hasn't been put into pyspark.pandas

#This failed because pct_change() function has not been put into pyspark.pandas
df_pct = data_pd.pct_change(1)

Another solution is to use: pandas_api() to convert the spark dataframe to pandas-spark dataframe.

This allows me to use pct_change() after converting spark dataframe to pandas-spark dataframe

data_pd = data.pandas_api()
data_pd.pct_change()

Source: https://stackoverflow.com/questions/73061556/does-pyspark-pandas-support-pandas-pct-change-function/73061837?noredirect=1#comment129063659_73061837