flowchart LR
name1 & name2 --> name3
name4 --> name5
INPUTS: - name1 - name2
certainly does something
@transform(
someclass,
Input('name1','dbfs://location1')
+ Input('name2','dbfs://location2')
>> Output('name3','dbfss://location3')
)
def do_something(self,**context):
""" certainly does something """
print(context['name1'].location)INPUTS: - name4
takes in some adls stuff and transforms it
@transform(
someclass,
Input('name4','dbfss://location_nowhere')
>> Output('name5','dbfss://location_somewhere')
)
def do_something_else(self,**context):
""" takes in some adls stuff and transforms it """
print(context)None
flowchart LR
output_name --> other_output_name
Enrichment_data_source & other_output_name --> final_gold_table
input_name --> output_name
INPUTS: - output_name
This method is to transform the Bronze Alchemy layer into the Silver Alchemy layer
Args:
None
Returns:
None
@transform(
Input('output_name','<adls path>')
>> Output('other_output_name','<adls path>')
)
def bronze_to_silver(**context) -> None:
"""
This method is to transform the Bronze Alchemy layer into the Silver Alchemy layer
Args:
None
Returns:
None
"""
self.logger.info(f"Bronze data transformed and written to {self.silver_path}")INPUTS: - Enrichment_data_source - other_output_name
@transform(
Input('Enrichment_data_source','<adls path>')
+ Input('other_output_name','<adls_path>')
>> Output('final_gold_table','<adls_path>')
)
def silver_to_gold(**context) -> None:
""" """
self.logger.info(f"{context}")INPUTS: - input_name
This method is designed to bring in data from an external source to the Bronze Alchemy layer
access input and output objects through context i.e.
>>> context["<input name>"].location
--> "<adls path>"
Args:
None
Returns:
None
@transform(
Input('input_name', '<adls path>')
# + <other input objects>
>> Output('output_name', '<adls path>')
)
def source_to_bronze(**context) -> None:
"""
This method is designed to bring in data from an external source to the Bronze Alchemy layer
access input and output objects through context i.e.
>>> context["<input name>"].location
--> "<adls path>"
Args:
None
Returns:
None
"""
self.logger.info(f"Source data written to {self.bronze_path}")