code: def cross_validation(
      blockdf, modeldf, score_udf, score_key_pattern, alphas, metric
  ):
      """
      Performs cross-validated optimization over the shrinkage hyperparameters alpha, using the models in modeldf and
      the block feature matrix blockdf.
  
      Args:
          blockdf : Spark DataFrame representing a once- or twice-reduced block matrix.
          modeldf : Spark DataFrame produced by the LogisticRegression or RidgeRegression fit method, representing the
          reducer models fit using the shrinkage parameters alpha.
          score_udf : Pandas UDF used to apply the scoring function to blockdf and modeldf
          score_key_pattern : Pattern of column names used in the groupBy clause before applying score_udf
          alphas : dict of alpha_name : alpha_value
          metric : string specifying either the r2 or the log_loss metric for model scoring
  
      Returns:
          Spark DataFrame  containing the results of the cross validation procedure.
      """
      alpha_df = blockdf.sql_ctx.createDataFrame(
          [
              Row(alpha=k, alpha_value=float(v))
              for k, v in alphas.items()
          ]
      )
      if metric == "r2":
          window_spec = Window.partitionBy("label").orderBy(
              f.desc("r2_mean"), f.desc("alpha_value")
          )
      elif metric == "log_loss":
          window_spec = Window.partitionBy("label").orderBy(
              "log_loss_mean", f.desc("alpha_value")
          )
      else:
          raise ValueError(
              f'Metric should be either "r2" or "log_loss", found {metric}'
          )
  
      # Hint a sort-merge join to avoid an automatic broadcast join that may cause an OOM
      return (
          blockdf.drop("header_block", "sort_key")
          .join(
              modeldf.hint("merge"), ["header", "sample_block"], "right"
          )
          .withColumn(
              "label",
              f.coalesce(f.col("label"), f.col("labels").getItem(0)),
          )
          .groupBy(score_key_pattern)
          .apply(score_udf)
          .join(alpha_df, ["alpha"])
          .groupBy("label", "alpha", "alpha_value")
          .agg(f.mean("score").alias(f"{metric}_mean"))
          .withColumn("modelRank", f.row_number().over(window_spec))
          .filter("modelRank = 1")
          .drop("modelRank")
      )
  
imports: import itertools
  import math
  import re
  import warnings
  from typing import Any, Dict, Iterable, List, Tuple
  import numpy
  import pandas
  import pyspark.sql.functions
  import scipy.optimize
  from nptyping import Float, Int, NDArray
  from pyspark.sql import DataFrame, Row
  from pyspark.sql.window import Window
  from typeguard import typechecked
label_desc: Use image node_id 30 for calling a global function with example usage: cross_validation(blockdf, modeldf, score_udf, score_key_pattern, alphas, metric) and returns: blockdf