Forum Discussion

mkj1213's avatar
mkj1213
Helper II
2 years ago
Solved

Pyspark code running twice causing a LIVY status = DEAD

Hello i am running the following cell (all packages has been imported and parameters set in a previous cells, and the code is running)

 

df = spark.sql(f"select * from LH.STOCK_Price where YEAR = {parameter_year} and MONTH <= {parameter_month}")

# Filter the DataFrame to include only rows where the year in TIMESTAMP is parameter
#df = df.filter((col("YEAR") == parameter_year) & (col("MONTH") <= parameter_month))

#df.cache()

# Ensure the data is partitioned appropriately
df = df.repartition("TYPE", "Vendor", col("TIMESTAMP").cast("date"))

# Define a custom function to count distinct IDs
distinct_count = F.expr("count(distinct ID)").alias("ID_DISTINCT_COUNT")

# Group by TYPE, Vendor, and TIMESTAMP (cast to date)
result = df.groupBy(
    col("TYPE"),
    col("Vendor"),
    col("TIMESTAMP").cast("date").alias("DATE")
).agg(
    count(col("ID")).alias("ID_COUNT"),
    distinct_count
)

result.cache()

# Write the result to a Delta table with partitioning
result.write.format("delta").mode("append").partitionBy("DATE").saveAsTable("LH.Count_Date_Wise_STOCKS")

 

The aim of my code is to read a STOCK_Price delta table from a lakehouse for a specific year month into a dataframe df, then group by different columns and count the ID and their distinct count. I am caching the result and then writing it to an aggregate delta table (Count_Date_Wise_Stocks), but i feel that part of my code is running twice as i can see from the spark diagnostic below the cell:


The second run is almost producing a session error with an error code LIVY state = Dead.

Some statistics about my tables:
STOCK_Price : around 700 Billion Rows
df : about 25 Billlions Rows
Count_Date_Wise_STOCKS: around 10,000 rows only.
the execution of the cell is taking sometimes up to 10 hours for 6 month of data and returning a Count_Date_Wise_STOCKS of around 4000 rows.

Questions: What part of my code is causing the second run? how can i avoid that if it is possible?
Regards



6 Replies

  • Anonymous's avatar
    Anonymous
    Not applicable

    Hi mkj1213 

     

    I feel like the number of Jobs doesn't mean that the code has run the same number of times. I made some test based on your code (have made some modifications to fit my data). In the following images, you will see that it experienced 7 Spark jobs. 

     

    When I query data from the result delta table, it doesn't show any duplicated rows. This means the code runs only once although it shows 7 Spark jobs. 

     

    According to my research, this "LIVY status = DEAD" error is more like a result of a lack of some kind of resource or a resource that goes out of limit. I find an Azure Synapse Analytics Blog related to this error as below. According to the solution in it, you can try to increase the node size of the Spark pool which is used to run the notebook. 

    Livy is dead and some logs to help. - Microsoft Community Hub

     

    Here are some Fabric documentations about setting Spark pool:

    Workspace administration settings in Microsoft Fabric - Microsoft Fabric | Microsoft Learn

    Manage settings for data engineering and science capacity - Microsoft Fabric | Microsoft Learn

     

    Hope this will be helpful!

     

    Best Regards,
    Jing
    If this post helps, please Accept it as Solution to help other members find it. Appreciate your Kudos!

    • mkj1213's avatar
      mkj1213
      Helper II

      Thanks for giving my problem some time on your end.
      i will look into the links that you had provided in more details and get back to you. The only difference that i had noticed between the structure of your code and my code was the agg part. In my code i had made two aggregation (count and distinct count) while in mine you had only done one. Is it possible to add another aggregation method (maybe a count) and share the spark jobs screenshot (the second image that you had included in your comments.

      On the spark pool settings, it turned out that i was using large size (maybe i could have increased it to X or XX size):

       


      Regards

      • Anonymous's avatar
        Anonymous
        Not applicable

        Hi mkj1213 This is my testing result with two aggregation methods. It includes 6 spark jobs the first time. 

        The second time:

  • Anonymous's avatar
    Anonymous
    Not applicable

    Hi mkj1213 

    Have you resolved this issue? If any of the answers provided were helpful, please consider accepting them as a solution. If you have found other solutions, we would greatly appreciate it if you could share them with us. Thank you!

     

    Best Regards,
    Jing