code: def test_apply_model(spark):
      indexdf = spark.read.parquet(
          f"{data_root}/groupedIDs.snappy.parquet"
      )
      blockdf = spark.read.parquet(
          f"{data_root}/blockedGT.snappy.parquet"
      )
      testGroup = "0"
      testBlock = "chr_1_block_0"
      ids = (
          indexdf.filter(f"sample_block = {testGroup}")
          .select("sample_ids")
          .head()
          .sample_ids
      )
      headers = [
          r.header
          for r in blockdf.filter(
              f'header_block = "{testBlock}" AND sample_block = {testGroup}'
          )
          .orderBy("sort_key")
          .select("header")
          .collect()
      ]
  
      X_in = X0[headers].loc[ids, :]
      X_out = X0[headers].drop(ids, axis="rows")
      Y_out = labeldf.drop(ids, axis="rows")
  
      XtX_out = X_out.to_numpy().T @ X_out.to_numpy()
      XtY_out = X_out.to_numpy().T @ Y_out.to_numpy()
      B = np.column_stack(
          [
              (
                  np.linalg.inv(
                      XtX_out + np.identity(XtX_out.shape[1]) * a
                  )
                  @ XtY_out
              )
              for a in alphas
          ]
      )
      X1_in = X_in.to_numpy() @ B
  
      sample_blocks = __get_sample_blocks(indexdf)
      map_key_pattern = ["header_block", "sample_block"]
      reduce_key_pattern = ["header_block", "header"]
      transform_key_pattern = ["header_block", "sample_block"]
      map_udf = pandas_udf(
          lambda key, pdf: map_normal_eqn(
              key,
              map_key_pattern,
              pdf,
              labeldf,
              sample_blocks,
              covdf_empty,
          ),
          normal_eqn_struct,
          PandasUDFType.GROUPED_MAP,
      )
      reduce_udf = pandas_udf(
          lambda key, pdf: reduce_normal_eqn(
              key, reduce_key_pattern, pdf
          ),
          normal_eqn_struct,
          PandasUDFType.GROUPED_MAP,
      )
      model_udf = pandas_udf(
          lambda key, pdf: solve_normal_eqn(
              key, map_key_pattern, pdf, labeldf, alphaMap, covdf_empty
          ),
          model_struct,
          PandasUDFType.GROUPED_MAP,
      )
      transform_udf = pandas_udf(
          lambda key, pdf: apply_model(
              key,
              transform_key_pattern,
              pdf,
              labeldf,
              sample_blocks,
              alphaMap,
              covdf_empty,
          ),
          reduced_matrix_struct,
          PandasUDFType.GROUPED_MAP,
      )
  
      modeldf = (
          blockdf.groupBy(map_key_pattern)
          .apply(map_udf)
          .groupBy(reduce_key_pattern)
          .apply(reduce_udf)
          .groupBy(map_key_pattern)
          .apply(model_udf)
      )
  
      columns = ["values"]
      rows = (
          blockdf.join(
              modeldf.drop("sort_key"),
              ["header_block", "sample_block", "header"],
          )
          .groupBy(transform_key_pattern)
          .apply(transform_udf)
          .filter(
              f'header LIKE "%{testBlock}%" AND sample_block = {testGroup}'
          )
          .select(*columns)
          .collect()
      )
      outdf = pd.DataFrame(rows, columns=columns)
  
      X1_in_lvl = np.column_stack(outdf["values"])
  
      assert np.allclose(X1_in_lvl, X1_in)
  
imports: from glow.wgr.ridge_reduction import *
  from glow.wgr.ridge_regression import *
  from glow.wgr.ridge_udfs import *
  from glow.wgr.model_functions import _fill_na_and_standardize
  import numpy
  import pandas
  from pyspark.sql.functions import *
  import math
label_desc: Use image node_id 6 for calling a global function with example usage: test_apply_model(spark) without return types