SparkAutoMapper

Fluent API to map data from one view to another in Spark.

Uses native Spark functions underneath so it is just as fast as hand writing the transformations.

Since this is just Python, you can use any Python editor. Since everything is typed using Python typings, most editors will auto-complete and warn you when you do something wrong

Usage

pip install sparkautomapper

Documentation

https://icanbwell.github.io/SparkAutoMapper/

SparkAutoMapper input and output

You can pass either a dataframe to SparkAutoMapper or specify the name of a Spark view to read from.

You can receive the result as a dataframe or (optionally) pass in the name of a view where you want the result.

Dynamic Typing Examples

Set a column in destination to a text value (read from pass in data frame and return the result in a new dataframe)

Set a column in destination to a text value

fromspark_auto_mapper.automappers.automapperimportAutoMappermapper=AutoMapper(
keys=["member_id"]
).columns(
dst1="hello"
)

Set a column in destination to a text value (read from a Spark view and put result in another Spark view)

Set a column in destination to a text value

fromspark_auto_mapper.automappers.automapperimportAutoMappermapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst1="hello"
)

Set a column in destination to an int value

Set a column in destination to a text value

fromspark_auto_mapper.automappers.automapperimportAutoMappermapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst1=1050
)

Copy a column (src1) from source_view to destination view column (dst1)

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst1=A.column("src1")
)

Or you can use the shortcut for specifying a column (wrap column name in [])

fromspark_auto_mapper.automappers.automapperimportAutoMappermapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst1="[src1]"
)

Convert data type for a column (or string literal)

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
birthDate=A.date(A.column("date_of_birth"))
)

Use a Spark SQL Expression (Any valid Spark SQL expression can be used)

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
gender=A.expression(
""" CASE WHEN `Member Sex` = 'F' THEN 'female' WHEN `Member Sex` = 'M' THEN 'male' ELSE 'other' END """
)
)

Specify multiple transformations

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst1="[src1]",
birthDate=A.date("[date_of_birth]"),
gender=A.expression(
""" CASE WHEN `Member Sex` = 'F' THEN 'female' WHEN `Member Sex` = 'M' THEN 'male' ELSE 'other' END """
)
)

Use variables or parameters

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAdefmapping(parameters: dict):
mapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst1=A.column(parameters["my_column_name"])
)

Use conditional logic

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAdefmapping(parameters: dict):
mapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst1=A.column(parameters["my_column_name"])
)
ifparameters["customer"] =="Microsoft":
mapper=mapper.columns(
important_customer=1,
customer_name=parameters["customer"]
)
returnmapper

Using nested array columns

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).withColumn(
dst2=A.list(
[
"address1",
"address2"
]
)
)

Using nested struct columns

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst2=A.complex(
use="usual",
family="imran"
)
)

Using lists of structs

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapper(
view="members",
source_view="patients",
keys=["member_id"]
).columns(
dst2=A.list(
[
A.complex(
use="usual",
family="imran"
),
A.complex(
use="usual",
family="[last_name]"
)
]
)
)

Executing the AutoMapper

spark.createDataFrame(
[
(1, 'Qureshi', 'Imran'),
(2, 'Vidal', 'Michael'),
],
['member_id', 'last_name', 'first_name']
).createOrReplaceTempView("patients")
source_df: DataFrame=spark.table("patients")
df=source_df.select("member_id")
df.createOrReplaceTempView("members")
result_df: DataFrame=mapper.transform(df=df)

Statically Typed Examples

To improve the auto-complete and syntax checking even more, you can define Complex types:

Define a custom data type:

fromspark_auto_mapper.type_definitions.automapper_defined_typesimportAutoMapperTextInputTypefromspark_auto_mapper.helpers.automapper_value_parserimportAutoMapperValueParserfromspark_auto_mapper.data_types.dateimportAutoMapperDateDataTypefromspark_auto_mapper.data_types.listimportAutoMapperListfromspark_auto_mapper_fhir.fhir_types.automapper_fhir_data_type_complex_baseimportAutoMapperFhirDataTypeComplexBaseclassAutoMapperFhirDataTypePatient(AutoMapperFhirDataTypeComplexBase):
# noinspection PyPep8Namingdef__init__(self,
id_: AutoMapperTextInputType,
birthDate: AutoMapperDateDataType,
name: AutoMapperList,
gender: AutoMapperTextInputType
) ->None:
super().__init__()
self.value=dict(
id=AutoMapperValueParser.parse_value(id_),
birthDate=AutoMapperValueParser.parse_value(birthDate),
name=AutoMapperValueParser.parse_value(name),
gender=AutoMapperValueParser.parse_value(gender)
)

Now you get auto-complete and syntax checking:

fromspark_auto_mapper.automappers.automapperimportAutoMapperfromspark_auto_mapper.helpers.automapper_helpersimportAutoMapperHelpersasAmapper=AutoMapperFhir(
view="members",
source_view="patients",
keys=["member_id"]
).withResource(
resource=F.patient(
id_=A.column("a.member_id"),
birthDate=A.date(
A.column("date_of_birth")
),
name=A.list(
F.human_name(
use="usual",
family=A.column("last_name")
)
),
gender="female"
)
)

Publishing a new package

1. Edit VERSION to increment the version
2. Create a new release
3. The GitHub Action should automatically kick in and publish the package
4. You can see the status in the Actions tab