Issues/49

[ravi-databricks]

Need to provide ability so that file metadata can be added to dataframe

e.g import dlt @dlt.table def bronze(): return (spark.readStream.format("cloudFiles") # define the schema for the ~6 common columns across files. All other input fields will be "rescued" into a JSON string column that can be queried via dot notation. .schema("Common1 string, Common2 string, _file_path string") # _file_path is a hidden auto-field but shows up in rescueData column JSON with this method. Spoofing that I have the same column in my input file so i can drop this spoofed column later .option("cloudFiles.format", "csv") .option("cloudFiles.schemaEvolutionMode", "rescue") .option("cloudFiles.rescuedDataColumn","extraFields") # override default _rescuedData column name with whatever you want to call this column .option("header","true") .load("/Volumes/vol/data/*.txt") .select("*","_metadata") # add file metadata information to output .drop("_file_path") # discard dummy input column to keep _file_path out of extraFields rescue data column ) 

Introduced select_metadata_cols inside source_details in onboarding file

 "source_metadata": { "include_autoloader_metadata_column": "True", "autoloader_metadata_col_name": "source_metadata", "select_metadata_cols": { "input_file_name": "_metadata.file_name", "input_file_path": "_metadata.file_path" } 

This will be utilized to add metadata columns to target tables.

Introducing custom_transform_func so that customers can bring their own transformations:
e.g.
you have custom transformations defined as python code as show below which takes dataframe as input and returns transformed dataframe

from pyspark.sql import DataFrame from pyspark.sql.functions import lit def custom_tranform_func_test(input_df) -> DataFrame: return input_df.withColumn('custom_col', lit('test')) 

you should able to pass this function to dlt-meta invoke function as below

layer = spark.conf.get("layer", None) from src.dataflow_pipeline import DataflowPipeline DataflowPipeline.invoke_dlt_pipeline(spark, layer, custom_tranform_func=custom_tranform_func_test) 

[codecov]

Codecov Report

Attention: Patch coverage is 56.25000% with 14 lines in your changes missing coverage. Please review.

Please upload report for BASE (feature/v0.0.8@3555aaa). Learn more about missing BASE report.

Files	Patch %	Lines
src/pipeline_readers.py	33.33%	13 Missing and 1 partial ⚠️

Additional details and impacted files

@@ Coverage Diff @@ ## feature/v0.0.8 #56 +/- ## ================================================= Coverage ? 88.57% ================================================= Files ? 8 Lines ? 858 Branches ? 168 ================================================= Hits ? 760 Misses ? 46 Partials ? 52 

Flag	Coverage Δ
unittests	88.57% <56.25%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[howardwu-db]

Need a period before "For eventhub="

[ravi-databricks]

done!