code: def test_score_logistic_model(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_score_logistic_model.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",
      ]
      model_key_pattern = ["sample_block", "label", "alpha_name"]
      score_key_pattern = ["sample_block", "label"]
  
      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,
      )
  
      model_udf = f.pandas_udf(
          lambda key, pdf: solve_irls_eqn(
              key, model_key_pattern, pdf, labeldf, alphas, covdf
          ),
          model_struct,
          PandasUDFType.GROUPED_MAP,
      )
  
      score_udf = f.pandas_udf(
          lambda key, pdf: score_models(
              key,
              score_key_pattern,
              pdf,
              labeldf,
              sample_blocks,
              alphas,
              covdf,
              maskdf,
              metric="log_loss",
          ),
          cv_struct,
          PandasUDFType.GROUPED_MAP,
      )
  
      modeldf = (
          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)
          .groupBy(model_key_pattern)
          .apply(model_udf)
          .withColumn(
              "alpha_label_coef",
              f.expr(
                  "struct(alphas[0] AS alpha, labels[0] AS label, coefficients[0] AS coefficient)"
              ),
          )
          .groupBy(
              "header_block",
              "sample_block",
              "header",
              "sort_key",
              f.col("alpha_label_coef.label"),
          )
          .agg(
              f.sort_array(f.collect_list("alpha_label_coef")).alias(
                  "alphas_labels_coefs"
              )
          )
          .selectExpr(
              "*",
              "alphas_labels_coefs.alpha AS alphas",
              "alphas_labels_coefs.label AS labels",
              "alphas_labels_coefs.coefficient AS coefficients",
          )
          .drop("alphas_labels_coefs", "label")
      )
  
      cvdf = (
          lvl1df.drop("header_block", "sort_key")
          .join(modeldf, ["header", "sample_block"], "right")
          .withColumn(
              "label",
              f.coalesce(f.col("label"), f.col("labels").getItem(0)),
          )
          .groupBy(score_key_pattern)
          .apply(score_udf)
      )
  
      outdf = cvdf.filter(
          f'sample_block = "{test_sample_block}" AND label = "{test_label}"'
      ).toPandas()
      scores_glow = outdf["score"].to_numpy()
  
      assert np.allclose(np.array(test_values["scores"]), scores_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 4 for calling a global function with example usage: test_score_logistic_model(spark) without return types