2
\$\begingroup\$

I am trying to convert a SQL Server stored procedure to Pyspark code.

This is my SQL Server query:

    WITH A (ID, CHILDMSF, PARENTMSF, TOPPARENT, TYPE, QUANTITY) AS 
    (
        -- Base case: Select distinct rows from #BO satisfying the condition
        SELECT DISTINCT 
            ID,
            CAST(CHILDMSF AS VARCHAR(MAX)) AS CHILDMSF,
            PARENTMSF,
            PARENTMSF,
            CAST(CONCAT(PARENTMSF, '-->', CHILDMSF) AS VARCHAR(MAX)) AS TRANSITION,
            QUANTITY,
            TYPE
        FROM 
            #BO BO(NOLOCK)
        WHERE 
            TYPE IN (0, 1)
        UNION ALL
        -- Recursive step: Join with previous results and select rows satisfying the condition
        SELECT 
            CAST(B.CHILDMSF AS VARCHAR(MAX)),
            B.PARENTMSF,
            P.TOPPARENT, 
            B.TYPE, 
            CAST(CONCAT(A.TRANSITION, '-->', B.CHILDMSF) AS VARCHAR(MAX)) AS TRANSITION, 
            B.QUANTITY
        FROM 
            A
        JOIN 
            #BO B (NOLOCK) ON B.PARENTMSF = A.CHILDMSF
        WHERE  
            A.TYPE IN (0, 1)
    )
    -- Select distinct rows from the recursive CTE and store them into #SUPPLYPARTS
    SELECT DISTINCT *
    INTO #SUPPLYPARTS
    FROM A
    WHERE TYPE IN (0, 1)
    ORDER BY CHILDMSF 
    -- Set MAXRECURSION option to 0 to avoid any limit on recursion depth
    OPTION(MAXRECURSION 0);

Below is the Pyspark code that I developed using iteration. Considering there is already root_df present

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, concat, lit



def generate_hierarchy(root_df):
    """
    Generate hierarchy using iterative join.

    Parameters:
    - root_df (DataFrame): The root DataFrame containing the initial data.

    Returns:
    - DataFrame: The DataFrame containing the hierarchy.
    """
    # Filter the DataFrame based on the TYPE column
    df = root_df.filter(root_df.TYPE.isin([0, 1]))

    # Create a new DataFrame to hold the results
    df_parents = df.select(
        df.ChildMsf.cast("string").alias("CHILDMSF"),
        df.ParentMsf.alias('PARENTMSF'),
        df.ParentMsf.alias("TOPPARENT"),
        df.TYPE,
        df.ISSPARABLE,
        concat(df.ParentMsf, lit("-->"), df.ChildMsf).cast("string").alias("BOMPATH"),
        df.QUANTITY
    )

    # Iteratively join the DataFrame with itself until the size of the DataFrame stabilizes
    while True:
        # Store the current state of df_parents
        df_parents_old = df_parents
        # Perform a self join to get next level of hierarchy
        df_parents = df_parents.union(df_parents.alias("P")
                                      .join(df.alias("T"), col("T.PARENTMSF") == col("P.CHILDMSF"))
                                      .select(
                                          col("T.CHILDMSF").cast("string"),
                                          col("T.PARENTMSF"),
                                          col("P.TOPPARENT"),
                                          col("T.TYPE"),
                                          col("T.ISSPARABLE"),
                                          concat(col("P.TRANSITION"), lit("-->"), col("T.CHILDMSF")).cast("string").alias("TRANSITION"),
                                          col("T.QUANTITY")
                                      )
                                      .filter(col("T.TYPE").isin([0, 1]))
                                      )
        # Check if the DataFrame size has stabilized (no new rows added)
        if df_parents.count() == df_parents_old.count():
            break

    # Remove duplicates
    df_parents = df_parents.distinct()

    # Order by CHILDMSF
    df_parents = df_parents.orderBy("CHILDMSF")

    return df_parents

The code is running infinitely long. I need help to know if I am doing it correctly.

\$\endgroup\$
3
  • \$\begingroup\$ Welcome. The SQL query is quite long and complicated to understand at first glance. If you need help, then help us understand your code. \$\endgroup\$ Mar 12 at 19:59
  • 1
    \$\begingroup\$ @BillalBegueradj I have added comments wherever needed to better explain \$\endgroup\$ Mar 12 at 20:09
  • \$\begingroup\$ NOLOCK against a temporary table makes no sense; why have those there. Who else are you expecting to be accessing that table when that table is limited to only be accessible by the current scope (and any inner scopes)? (Not to mentioned that NOLOCK is almost always the wrong decision.) Have you determined if it's the SQL or the Python that's the problem? If not, please do so, and then advise back. \$\endgroup\$
    – Larnu
    Mar 13 at 11:53

1 Answer 1

1
\$\begingroup\$

Here is the Problem Statement as I heard it:
A SQL Server stored procedure was running slowly, so it was converted to PySpark. The resulting PySpark function is running slowly. (Or, based on "running infinitely long", perhaps it never produces any visible results so it's not yet clear that it faithfully reproduces the behavior of the original stored procedure.)

Not an encouraging beginning. @Larnu asked you "where does the time go?" Is it mostly spent on fetching SQL results, or on evaluating python bytecode? You declined to supply observed timing results.

You didn't describe the Business Domain. Based on {CHILD,PARENT}MSF identifiers, I choose to interpret it as producing bills of materials used in supply chain logistics for Médecins Sans Frontières. Feel free to further illuminate the code.

You elected not to disclose the schema / indexing of your tables. I choose to assume that CHILDMSF is of type CHAR(80), which the backend needs cast to VARCHAR for some reason. (Or was it of type INTEGER?) Applying a function to a column, such as casting, will typically disable the column's index(es), perhaps forcing the backend planner to resort to a table scan. Including EXPLAIN PLAN output would have been very helpful. Understanding how many billion rows are involved, typical tree depth, and how many terabytes of RAM the target host has would also be helpful.


efficient query plan

                      .join(df.alias("T"),
                            col("T.PARENTMSF") == col("P.CHILDMSF"))

The OP offers no clue about how root_df is organized. When we filter it down to a df containing our favorite pair of types, we have an opportunity to .orderBy("PARENTMSF"). (Does .alias("T") denote "type filtered"?)

Ideally the execution plan would exploit indexes on T.PARENTMSF and P.CHILDMSF to quickly accomplish a MERGE JOIN.

How many hosts are involved in storing the intermediate results? What is Adaptive Query Execution telling you about the runtime costs?

timing debug

Just before checking whether the df size has stabilized, you have a really good opportunity to display an elapsed time and the pair of row counts. That would let you nail down "infinitely long" to an actual measurement of cycle time, even if there's an infinite number of cycles. It would also let you identify if the expected number of new transitions was found in the first several cycles.

Consider writing out an ordered copy of df_parents at this point, to distinct CSV filenames. That lets you run /usr/bin/diff -u to see the specifics of which new transitions were picked up.

You really want to test and debug this on a small subset of the original data. Impose filters like WHERE PARENTMSF < x, so the results will be small enough that you can quickly go through an edit-run-debug cycle, learning a little more with each new run.

redundant operation

I don't understand this line:

                                      .filter(col("T.TYPE").isin([0, 1]))

Didn't we already do that when initializing?

    df = root_df.filter(root_df.TYPE.isin([0, 1]))

Simply elide the redundant filter operation.


This codebase fails to achieve its design goals. It does not include automated tests that verify timing and correctness properties.

I would not be willing to delegate or accept maintenance tasks on it.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.