Mapper

Class

class MapperMode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Possible values: [‘replace’, ‘append’, ‘prepend’, ‘rename_and_validate’]

class MissingColumnHandling(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Possible values: [‘raise_error’, ‘skip’, ‘nullify’]

exception MapperParameterNotSupported

exception DataTypeValidationFailed

exception ColumnMappingNotSupported

class Mapper(mapping: List[Tuple], ignore_ambiguous_columns: bool = False, missing_column_handling: MissingColumnHandling = MissingColumnHandling.raise_error, mode: MapperMode = MapperMode.replace, annotator_options: dict | None = None, **kwargs)

Selects, transforms, comments and casts or validates a DataFrame based on the provided mapping.

Examples

>>> from pyspark.sql import functions as F, types as T
>>> from spooq.transformer import Mapper
>>> from spooq.transformer import mapper_transformations as spq
>>>
>>> cmt_id = "Identifier for entity (UUID4)"
>>> cmt_type = "Type of entity ('type_a', 'type_b' or 'type_c')"
>>>
>>> mapping = [
>>>     ("id",            "data.relationships.food.data.id",  spq.to_str,                     cmt_id),
>>>     ("version",       "data.version",                     spq.to_int),
>>>     ("type",          "elem.attributes.type",             "string",                       cmt_type),
>>>     ("created_at",    "elem.attributes.created_at",       spq.to_timestamp),
>>>     ("created_on",    "elem.attributes.created_at",       spq.to_timestamp(cast="date")),
>>>     ("processed_at",  F.current_timestamp(),              spq.as_is,
>>> ]
>>> mapper = Mapper(mapping=mapping)
>>> mapper.transform(input_df).printSchema()
root
 |-- id: string (nullable = true)
 |-- version: integer (nullable = true)
 |-- type: string (nullable = true)
 |-- created_at: timestamp (nullable = true)
 |-- created_on: date (nullable = true)
 |-- processed_at: timestamp (nullable = false)

Parameters

• mapping (list of tuple containing three or four elements, respectively) – This is the main parameter for this transformation. It gives information about the column names for the output DataFrame, the column names (paths) from the input DataFrame, their data types and optionally a column comment. Custom data types are also supported, which can clean, pivot, anonymize, … the data itself. Please have a look at the spooq.transformer.mapper_transformations module for more information.

• missing_column_handling (MissingColumnHandling, Defaults to MissingColumnHandling.raise_error) –

  Specifies how to proceed in case a source column does not exist in the source DataFrame:

  • raise_error (default)

    Raise an exception

  • nullify

    Create source column filled with null

  • skip

    Skip the mapping transformation

• ignore_ambiguous_columns (bool, Defaults to False) – It can happen that the input DataFrame has ambiguous column names (like “Key” vs “key”) which will raise an exception with Spark when reading. This flag surpresses this exception and skips those affected columns.

• mode (MapperMode, Defaults to MapperMode.replace) –

  Defines whether the mapping should fully replace the schema of the input DataFrame or just add to it. Following modes are supported:

  • replace

    The output schema is the same as the provided mapping. => output schema: columns from mapping

  • append

    The columns provided in the mapping are added at the end of the input schema. If a column already exists in the input DataFrame, its position is kept. => output schema: input columns + columns from mapping

  • prepend

    The columns provided in the mapping are added at the beginning of the input schema. If a column already exists in the input DataFrame, its position is kept. => output schema: columns from mapping + input columns

  • rename_and_validate

    All built-in, custom transformations (except renaming) and casts are disabled. The Mapper only renames the columns and validates that the output data type is the same as the input data type. The transformation will fail if any spooq / custom transformations (except as_is) are defined! => output schema: columns from mapping

• annotator_option (dict, Defaults to {}) – Options that are passed as parameters to the Annotator instance used by the Mapper Transformer if comments are provided.

Note

Let’s talk about Mappings:

The mapping should be a list of tuples that contain all necessary information per column.

• Column Name: str

  Sets the name of the column in the resulting output DataFrame.

• Source Path / Name / Column / Function: str, Column or functions

  Points to the name of the column in the input DataFrame. If the input is a flat DataFrame, it will essentially be the column name. If it is of complex type, it will point to the path of the actual value. For example: data.relationships.sample.data.id, where id is the value we want. It is also possible to directly pass a PySpark Column which will get evaluated. This can contain arbitrary logic supported by Spark. For example: F.current_date() or F.when(F.col("size") == 180, F.lit("tall")).otherwise(F.lit("tiny")).

• DataType: str, DataType or mapper_transformations

  DataTypes can be types from pyspark.sql.types (like T.StringType()), simple strings supported by PySpark (like “string”) and custom transformations provided by spooq (like spq.to_timestamp). You can find more information about the transformations at https://spooq.rtfd.io/en/latest/transformer/mapper.html#module-spooq.transformer.mapper_transformations.

• Comment: str (optional)

  Applies a comment to the respective column, if provided.

Note

The available input columns can vary from batch to batch if you use schema inference (f.e. on json data) for the extraction. Via the parameter missing_column_handling you can specify a strategy on how to handle missing columns on the input DataFrame. It is advised to use the ‘raise_error’ option as it can uncover typos and bugs.

transform(input_df: DataFrame) → DataFrame

Performs a transformation on a DataFrame.

Parameters

input_df (DataFrame) – Input DataFrame

Returns

Transformed DataFrame.

Return type

DataFrame

Note

This method does only take the Input DataFrame as a parameters. Any other needed parameters are defined in the initialization of the Transformator Object.

Custom Transformations

This is a collection of module level functions to be applied to a DataFrame. These methods can be used with the Mapper transformer or directly within a select or withColumn statement.

All functions support following generic functionalities:

• alt_src_cols: Alternative source columns that will be used within a coalesce function if provided

• cast: Explicit casting after the transformation (sane defaults are set for each function)

to_str, to_int, to_long, to_float, to_double are convenience methods with a hardcoded cast that cannot be changed.

All examples assume following code has been executed before:

>>> from pyspark.sql import Row
>>> from pyspark.sql import functions as F, types as T
>>> from spooq.transformer import Mapper
>>> from spooq.transformer import mapper_transformations as spq

spooq.transformer.mapper_transformations.as_is([...])	Returns a renamed column without any casting.
spooq.transformer.mapper_transformations.to_num([...])	More robust conversion to number data types (Default: LongType).
spooq.transformer.mapper_transformations.to_bool([...])	More robust conversion to BooleanType.
spooq.transformer.mapper_transformations.to_timestamp([...])	More robust conversion to TimestampType (or as a formatted string).
spooq.transformer.mapper_transformations.str_to_array([...])	Splits a string into a list (ArrayType).
spooq.transformer.mapper_transformations.map_values([...])	Maps input values to specified output values.
spooq.transformer.mapper_transformations.meters_to_cm([...])	Converts meters to cm and casts the result to an IntegerType.
spooq.transformer.mapper_transformations.has_value([...])	Returns True if the source_column is
spooq.transformer.mapper_transformations.apply([...])	Applies a function / partial
spooq.transformer.mapper_transformations.to_json_string([...])	Returns a column as json compatible string.
spooq.transformer.mapper_transformations.to_str([...])	Convenience transformation that only casts to string.
spooq.transformer.mapper_transformations.to_int([...])	Syntactic sugar for calling to_num(cast="int")
spooq.transformer.mapper_transformations.to_long([...])	Syntactic sugar for calling to_num(cast="long")
spooq.transformer.mapper_transformations.to_float([...])	Syntactic sugar for calling to_num(cast="float")
spooq.transformer.mapper_transformations.to_double([...])	Syntactic sugar for calling to_num(cast="double")