class_docstr: InputLoader that reads data in as a Spark DataFrame.
class_docstr_tok: ["InputLoader","that","reads","data","in","as","a","Spark","DataFrame","."]
docstr: Load the contents from an S3 based data source as a DataFrame.
  
  Args:
      s3_data_source (Union[CSVDataSource, ParquetDataSource]):
          A data source that is based in S3.
  
  Raises:
      ValueError: If an invalid DataSource is provided.
  
  Returns:
      DataFrame: Contents of the data loaded from S3.
docstr_tok: ["Load","the","contents","from","an","S3","based","data","source","as","a","DataFrame",".","Args",":","s3_data_source","(","Union","[","CSVDataSource",",","ParquetDataSource","]",")",":","A","data","source","that","is","based","in","S3",".","Raises",":","ValueError",":","If","an","invalid","DataSource","is","provided",".","Returns",":","DataFrame",":","Contents","of","the","data","loaded","from","S3","."]
code: def load_from_s3(
      self, s3_data_source: Union[CSVDataSource, ParquetDataSource]
  ) -> DataFrame:
      """Load the contents from an S3 based data source as a DataFrame.
  
      Args:
          s3_data_source (Union[CSVDataSource, ParquetDataSource]):
              A data source that is based in S3.
  
      Raises:
          ValueError: If an invalid DataSource is provided.
  
      Returns:
          DataFrame: Contents of the data loaded from S3.
      """
      spark_session = self.spark_session_factory.spark_session
      s3a_uri = s3_data_source.s3_uri.replace("s3://", "s3a://")
  
      if isinstance(s3_data_source, CSVDataSource):
          # TODO: Accept `schema` parameter. (Inferring schema requires a pass through every row)
          logger.info("Loading data from %s.", s3a_uri)
          return spark_session.read.csv(
              s3a_uri,
              header=s3_data_source.csv_header,
              inferSchema=s3_data_source.csv_infer_schema,
          )
  
      if isinstance(s3_data_source, ParquetDataSource):
          logger.info("Loading data from %s.", s3a_uri)
          return spark_session.read.parquet(s3a_uri)
  
      raise ValueError("An invalid data source was provided.")
  
imports: from __future__ import absolute_import
  import logging
  import re
  from abc import ABC, abstractmethod
  from typing import Generic, Optional, TypeVar, Union
  import attr
  from pyspark.sql import DataFrame
  from sagemaker import Session
  from sagemaker.feature_store.feature_processor._constants import FEATURE_GROUP_ARN_REGEX_PATTERN
  from sagemaker.feature_store.feature_processor._data_source import CSVDataSource, FeatureGroupDataSource, ParquetDataSource, IcebergTableDataSource
  from sagemaker.feature_store.feature_processor._spark_factory import SparkSessionFactory
  from sagemaker.feature_store.feature_processor._input_offset_parser import InputOffsetParser
  from sagemaker.feature_store.feature_processor._env import EnvironmentHelper
label_desc: Use image node_id 3 for calling the SparkDataFrameInputLoader obj's underlying member method code with example usage: obj.load_from_s3(s3_data_source) and returns: spark_session, spark_session