code: def flatten_dataframe(
      df: DataFrame,
      separator: str = ":",
      replace_char: str = "_",
      sanitized_columns: bool = False,
  ) -> DataFrame:
      """Flattens the complex columns in the DataFrame.
  
      :param df: The input PySpark DataFrame.
      :type df: DataFrame
      :param separator: The separator to use in the resulting flattened column names, defaults to ":".
      :type separator: str, optional
      :param replace_char: The character to replace special characters with in column names, defaults to "_".
      :type replace_char: str, optional
      :param sanitized_columns: Whether to sanitize column names, defaults to False.
      :type sanitized_columns: bool, optional
      :return: The DataFrame with all complex data types flattened.
      :rtype: DataFrame
  
      .. note:: This function assumes the input DataFrame has a consistent schema across all rows. If you have files with
          different schemas, process each separately instead.
  
      .. example:: Example usage:
  
          >>> data = [
                  (
                      1,
                      ("Alice", 25),
                      {"A": 100, "B": 200},
                      ["apple", "banana"],
                      {"key": {"nested_key": 10}},
                      {"A#": 1000, "B@": 2000},
                  ),
                  (
                      2,
                      ("Bob", 30),
                      {"A": 150, "B": 250},
                      ["orange", "grape"],
                      {"key": {"nested_key": 20}},
                      {"A#": 1500, "B@": 2500},
                  ),
              ]
  
          >>> df = spark.createDataFrame(data)
          >>> flattened_df = flatten_dataframe(df)
          >>> flattened_df.show()
          >>> flattened_df_with_hyphen = flatten_dataframe(df, replace_char="-")
          >>> flattened_df_with_hyphen.show()
      """
  
      def sanitize_column_name(name: str, rc: str = "_") -> str:
          """Sanitizes column names by replacing special characters with the specified character.
  
          :param name: The original column name.
          :type name: str
          :param rc: The character to replace special characters with, defaults to '_'.
          :type rc: str, optional
          :return: The sanitized column name.
          :rtype: str
          """
          return re.sub(r"[^a-zA-Z0-9_]", rc, name)
  
      def explode_array(df: DataFrame, col_name: str) -> DataFrame:
          """Explodes the specified ArrayType column in the input DataFrame and returns a new DataFrame with the exploded column.
  
          :param df: The input PySpark DataFrame.
          :type df: DataFrame
          :param col_name: The column name of the ArrayType to be exploded.
          :type col_name: str
          :return: The DataFrame with the exploded ArrayType column.
          :rtype: DataFrame
          """
          return df.select(
              "*",
              F.explode_outer(F.col(f"`{col_name}`")).alias(col_name),
          ).drop(
              col_name,
          )
  
      fields = complex_fields(df.schema)
  
      while len(fields) != 0:
          col_name = next(iter(fields.keys()))
  
          if isinstance(fields[col_name], StructType):
              df = flatten_struct(
                  df, col_name, separator
              )  # noqa: PD901
  
          elif isinstance(fields[col_name], ArrayType):
              df = explode_array(df, col_name)  # noqa: PD901
  
          elif isinstance(fields[col_name], MapType):
              df = flatten_map(df, col_name, separator)  # noqa: PD901
  
          fields = complex_fields(df.schema)
  
      # Sanitize column names with the specified replace_char
      if sanitized_columns:
          sanitized_columns = [
              sanitize_column_name(col_name, replace_char)
              for col_name in df.columns
          ]
          df = df.toDF(*sanitized_columns)  # noqa: PD901
  
      return df
  
file_path: quinn/quinn/transformations.py
imports: from __future__ import annotations
  import re
  from collections.abc import Callable
  from pyspark.sql import DataFrame, SparkSession
  from pyspark.sql import functions
  from pyspark.sql.types import ArrayType, MapType, StructField, StructType
  from quinn.schema_helpers import complex_fields
label_desc: Use image node_id 8 for calling a global function with example usage: flatten_dataframe(df, separator, replace_char, sanitized_columns) and returns: df, re, df