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!