Example of Mapper Transformer =============================

.. code-block:: python

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

input_df = spark.createDataFrame( [ Row( structa=Row(idx="000123_456", sts="enabled", ts="1597069446000"), struct_b=Row(itms="1,2,4", sts="whitelisted", ts="2020-08-12T12:43:14+0000"), struct_c=Row(email="abc@def.com", gndr="F", dt="2020-08-05", cmt="fine"), ), Row( structa=Row(idx="000654_321", sts="off", ts="1597069500784"), struct_b=Row(itms="5", sts="blacklisted", ts="2020-07-01T12:43:14+0000"), struct_c=Row(email="", gndr="m", dt="2020-06-27", cmt="faulty"), ), ], schema=""" a: struct<idx string, sts string, ts string>, b: struct<itms string, sts string, ts string>, c: struct<email string, gndr string, dt string, cmt string> """ ) input_df.printSchema() root |-- a: struct (nullable = true) | |-- idx: string (nullable = true) | |-- sts: string (nullable = true) | |-- ts: string (nullable = true) |-- b: struct (nullable = true) | |-- itms: string (nullable = true) | |-- sts: string (nullable = true) | |-- ts: string (nullable = true) |-- c: struct (nullable = true) | |-- email: string (nullable = true) | |-- gndr: string (nullable = true) | |-- dt: string (nullable = true) | |-- cmt: string (nullable = true)

mapping = [ # output_name # source # transformation ("index", "a.idx", spq.to_int), # removes leading zeros and underline characters ("isenabled", "a.sts", spq.tobool), # recognizes additional words like "on", "off", "disabled", "enabled", ... ("aupdatedat", "a.ts", spq.to_timestamp), # supports unix timestamps in ms or seconds and strings ("items", "b.itms", spq.strtoarray(cast="int")), # splits a comma delimited string into an array and casts its elements ("blockstatus", "b.sts", spq.mapvalues(mapping={"whitelisted": "allowed", "blacklisted": "blocked"})), # applies lookup dictionary ("bupdatedat", "b.ts", spq.to_timestamp), # supports unix timestamps in ms or seconds and strings ("hasemail", "c.email", spq.hasvalue), # interprets also empty strings as no value, although, zeros are values ("gender", "c.gndr", spq.apply(func=F.lower)), # applies provided function to all values ("creationdate", "c.dt", spq.totimestamp(cast="date")), # explicitly casts result after transformation ("processedat", F.currenttimestamp(), spq.as_is), # source column is a function, no transformation to the results ("comment", "c.cmt", "string"), # no transformation, only cast; alternatively: spq.tostr or spq.asis(cast="string") ] outputdf = Mapper(mapping).transform(inputdf)

output_df.show(truncate=False) +------+----------+-----------------------+---------+------------+-------------------+---------+------+-------------+----------------------+-------+ |index |isenabled|aupdatedat |items |blockstatus|bupdatedat |hasemail|gender|creationdate|processed_at |comment| +------+----------+-----------------------+---------+------------+-------------------+---------+------+-------------+----------------------+-------+ |123456|true |2020-08-10 16:24:06 |[1, 2, 4]|allowed |2020-08-12 14:43:14|true |f |2020-08-05 |2022-08-12 09:17:09.83|fine | |654321|false |2020-08-10 16:25:00.784|[5] |blocked |2020-07-01 14:43:14|false |m |2020-06-27 |2022-08-12 09:17:09.83|faulty | +------+----------+-----------------------+---------+------------+-------------------+---------+------+-------------+----------------------+-------+

output_df.printSchema() root |-- index: integer (nullable = true) |-- is_enabled: boolean (nullable = true) |-- aupdatedat: timestamp (nullable = true) |-- items: array (nullable = true) | |-- element: integer (containsNull = true) |-- block_status: string (nullable = true) |-- bupdatedat: timestamp (nullable = true) |-- has_email: boolean (nullable = false) |-- gender: string (nullable = true) |-- creation_date: date (nullable = true) |-- processed_at: timestamp (nullable = false) |-- comment: string (nullable = true)

Features / Components =====================