code: def test_reduce_irls_eqn(spark):
      lvl1df = spark.read.parquet(
          f"{data_root}/bt_reduceded_1part.snappy.parquet"
      )
      sample_blocks_df = (
          spark.read.parquet(f"{data_root}/groupedIDs.snappy.parquet")
          .withColumn(
              "sample_block", f.col("sample_block").cast("string")
          )
          .withColumn(
              "sample_ids",
              f.expr("transform(sample_ids, v -> cast(v as string))"),
          )
      )
      sample_blocks = {
          r.sample_block: r.sample_ids
          for r in sample_blocks_df.collect()
      }
      with open(f"{data_root}/test_reduce_irls_eqn.json") as json_file:
          test_values = json.load(json_file)
  
      map_key_pattern = ["sample_block", "label", "alpha_name"]
      reduce_key_pattern = [
          "header_block",
          "header",
          "label",
          "alpha_name",
      ]
  
      map_udf = f.pandas_udf(
          lambda key, pdf: map_irls_eqn(
              key,
              map_key_pattern,
              pdf,
              labeldf,
              sample_blocks,
              covdf,
              beta_cov_dict,
              maskdf,
              alphas,
          ),
          irls_eqn_struct,
          PandasUDFType.GROUPED_MAP,
      )
  
      reduce_udf = f.pandas_udf(
          lambda key, pdf: reduce_irls_eqn(
              key, reduce_key_pattern, pdf
          ),
          irls_eqn_struct,
          PandasUDFType.GROUPED_MAP,
      )
  
      reducedf = (
          lvl1df.withColumn(
              "alpha_name",
              f.explode(f.array([f.lit(n) for n in alphas.keys()])),
          )
          .groupBy(map_key_pattern)
          .apply(map_udf)
          .groupBy(reduce_key_pattern)
          .apply(reduce_udf)
      )
  
      outdf = (
          reducedf.filter(
              f'sample_block = "{test_sample_block}" AND label = "{test_label}" AND alpha_name = "{test_alpha}"'
          )
          .orderBy("sort_key", "header")
          .toPandas()
      )
  
      xtgx_glow = np.row_stack(outdf["xtgx"])
      xty_glow = outdf["xty"].ravel()
      beta_glow = outdf["beta"].ravel()
  
      assert (
          np.allclose(np.array(test_values["xtgx"]), xtgx_glow)
          and np.allclose(np.array(test_values["xty"]), xty_glow)
          and np.allclose(np.array(test_values["beta"]), beta_glow)
      )
  
file_path: glow/python/glow/wgr/tests/test_logistic_ridge_regression.py
imports: from pyspark.sql.functions import PandasUDFType
  from glow.wgr.logistic_udfs import *
  from glow.wgr.logistic_ridge_regression import *
  from pyspark.sql import functions
  import json
  import pandas
  import numpy
label_desc: Use image node_id 2 for calling a global function with example usage: test_reduce_irls_eqn(spark) without return types