code: def create_normalization_spec_spark(
      df, column, num_samples: int, seed: Optional[int] = None
  ):
      """Returns approximately num_samples random rows from column of df."""
  
      # assumes column has a type of map
      df = df.select(
          explode(col(column).alias("features")).alias(
              "feature_name", "feature_value"
          )
      )
  
      # calculate fractions
      counts_df = df.groupBy("feature_name").count()
      frac = {}
      for row in counts_df.collect():
          assert num_samples <= row["count"]
          frac[row["feature_name"]] = num_samples / row["count"]
  
      # TODO(T64843081): change to reservoir sampling, currently it approximates
      # perform sampling and collect them
      df = df.sampleBy("feature_name", fractions=frac, seed=seed)
      df = df.groupBy("feature_name").agg(
          collect_list("feature_value").alias("feature_values")
      )
      return df
  
file_path: ReAgent/reagent/workflow/identify_types_flow.py
imports: from typing import Dict, List, Optional
  import reagent.core.types
  from pyspark.sql.functions import col, collect_list, explode
  from reagent.data.spark_utils import get_spark_session
  from reagent.preprocessing.normalization import get_feature_norm_metadata, NormalizationParameters
  from .types import PreprocessingOptions, TableSpec
label_desc: Use image node_id 3 for calling a global function with example usage: create_normalization_spec_spark(df, column, num_samples, seed) and returns: df