1.Added delta reader as per PR comment

ravi-db · ravi-db · commit b1f53045b933 · 2023-05-16T16:04:18.000-07:00
2.Corrected docs formatting for code blocks
3.Added unit tests for Delta reader
diff --git a/docs/content/faq/general.md b/docs/content/faq/general.md
@@ -17,7 +17,7 @@ DLT-META is a solution/framework using Databricks Delta Live Tables aka DLT whic
 
 **Q. What different types of reader are supported using DLT-META ?**
 
-DLT-META uses Databricks [Auto Loader](https://docs.databricks.com/ingestion/auto-loader/index.html) to read from s3/adls/blog stroage.
+DLT-META uses Databricks [Auto Loader](https://docs.databricks.com/ingestion/auto-loader/index.html), DELTA, KAFKA, EVENTHUB to read from s3/adls/blog stroage.
 
 **Q. Can DLT-META support any other readers?**
 
diff --git a/docs/content/getting_started/additionals.md b/docs/content/getting_started/additionals.md
@@ -25,28 +25,28 @@ export DATABRICKS_TOKEN=<DATABRICKS TOKEN> # Account needs permission to create
 
     5b. Run the command for eventhub ```python integration-tests/run-integration-test.py --cloud_provider_name=azure --dbr_version=11.3.x-scala2.12 --source=eventhub --dbfs_path=dbfs:/tmp/DLT-META/ --eventhub_name=iot --eventhub_secrets_scope_name=eventhubs_creds --eventhub_namespace=int_test-standard --eventhub_port=9093 --eventhub_producer_accesskey_name=producer ----eventhub_consumer_accesskey_name=consumer```
 
-        For eventhub integration tests, the following are the prerequisites:
-        1. Needs eventhub instance running
-        2. Using Databricks CLI, Create databricks secrets scope for eventhub keys
-        3. Using Databricks CLI, Create databricks secrets to store producer and consumer keys using the scope created in step 2 
+    For eventhub integration tests, the following are the prerequisites:
+    1. Needs eventhub instance running
+    2. Using Databricks CLI, Create databricks secrets scope for eventhub keys
+    3. Using Databricks CLI, Create databricks secrets to store producer and consumer keys using the scope created in step 2 
 
-        Following are the mandatory arguments for running EventHubs integration test
-        1. Provide your eventhub topic : --eventhub_name
-        2. Provide eventhub namespace : --eventhub_namespace
-        3. Provide eventhub port : --eventhub_port
-        4. Provide databricks secret scope name : ----eventhub_secrets_scope_name
-        5. Provide eventhub producer access key name : --eventhub_producer_accesskey_name
-        6. Provide eventhub access key name : --eventhub_consumer_accesskey_name
+    Following are the mandatory arguments for running EventHubs integration test
+    1. Provide your eventhub topic : --eventhub_name
+    2. Provide eventhub namespace : --eventhub_namespace
+    3. Provide eventhub port : --eventhub_port
+    4. Provide databricks secret scope name : ----eventhub_secrets_scope_name
+    5. Provide eventhub producer access key name : --eventhub_producer_accesskey_name
+    6. Provide eventhub access key name : --eventhub_consumer_accesskey_name
 
 
     5c. Run the command for kafka ```python3 integration-tests/run-integration-test.py --cloud_provider_name=aws --dbr_version=11.3.x-scala2.12 --source=kafka --dbfs_path=dbfs:/tmp/DLT-META/ --kafka_topic_name=dlt-meta-integration-test --kafka_broker=host:9092```
 
-        For kafka integration tests, the following are the prerequisites:
-        1. Needs kafka instance running
+    For kafka integration tests, the following are the prerequisites:
+    1. Needs kafka instance running
 
-        Following are the mandatory arguments for running EventHubs integration test
-        1. Provide your kafka topic name : --kafka_topic_name
-        2. Provide kafka_broker : --kafka_broker
+    Following are the mandatory arguments for running EventHubs integration test
+    1. Provide your kafka topic name : --kafka_topic_name
+    2. Provide kafka_broker : --kafka_broker
 
 6. Once finished integration output file will be copied locally to 
 ```integration-test-output_<run_id>.txt```
diff --git a/docs/content/getting_started/runoboardingopt2.md b/docs/content/getting_started/runoboardingopt2.md
@@ -12,18 +12,18 @@ draft: false
 ```
 ```
 onboarding_params_map = {
-	                        "database": "dlt_demo",
-	                        "onboarding_file_path": "dbfs:/onboarding_files/users_onboarding.json",
-	 						"bronze_dataflowspec_table": "bronze_dataflowspec_table", 
-	                        "bronze_dataflowspec_path": "dbfs:/onboarding_tables_cdc/bronze",                       
-	                        "silver_dataflowspec_table": "silver_dataflowspec_table",
-	                        "silver_dataflowspec_path": "dbfs:/onboarding_tables_cdc/silver",
-	                        "overwrite": "True",
-	                        "onboard_layer": "bronze_silver",
-	                        "env": "dev",
-	                        "version": "v1",
-	                        "import_author": "Ravi"
-    					} 
+		"database": "dlt_demo",
+		"onboarding_file_path": "dbfs:/onboarding_files/users_onboarding.json",
+		"bronze_dataflowspec_table": "bronze_dataflowspec_table", 
+		"bronze_dataflowspec_path": "dbfs:/onboarding_tables_cdc/bronze",                       
+		"silver_dataflowspec_table": "silver_dataflowspec_table",
+		"silver_dataflowspec_path": "dbfs:/onboarding_tables_cdc/silver",
+		"overwrite": "True",
+		"onboard_layer": "bronze_silver",
+		"env": "dev",
+		"version": "v1",
+		"import_author": "Ravi"
+} 
 
 from src.onboard_dataflowspec import OnboardDataflowspec
 OnboardDataflowspec(spark, onboarding_params_map).onboard_dataflow_specs()
diff --git a/src/dataflow_pipeline.py b/src/dataflow_pipeline.py
@@ -98,6 +98,8 @@ def read_bronze(self) -> DataFrame:
         bronze_dataflow_spec: BronzeDataflowSpec = self.dataflowSpec
         if bronze_dataflow_spec.sourceFormat == "cloudFiles" or bronze_dataflow_spec.sourceFormat == "delta":
             return PipelineReaders.read_dlt_cloud_files(self.spark, bronze_dataflow_spec, self.schema_json)
+        if bronze_dataflow_spec.sourceFormat == "delta":
+            return PipelineReaders.read_dlt_delta(self.spark, bronze_dataflow_spec)
         elif bronze_dataflow_spec.sourceFormat == "eventhub" or bronze_dataflow_spec.sourceFormat == "kafka":
             return PipelineReaders.read_kafka(self.spark, bronze_dataflow_spec, self.schema_json)
         else:
diff --git a/src/pipeline_readers.py b/src/pipeline_readers.py
@@ -46,6 +46,34 @@ def read_dlt_cloud_files(spark, bronze_dataflow_spec, schema_json) -> DataFrame:
                 .load(source_path)
             )
 
+    @staticmethod
+    def read_dlt_delta(spark, bronze_dataflow_spec) -> DataFrame:
+        """Read dlt delta.
+
+        Args:
+            spark (_type_): _description_
+            bronze_dataflow_spec (_type_): _description_
+            schema_json (_type_): _description_
+
+        Returns:
+            DataFrame: _description_
+        """
+        logger.info("In read_dlt_cloud_files func")
+        source_path = bronze_dataflow_spec.sourceDetails["path"]
+        reader_config_options = bronze_dataflow_spec.readerConfigOptions
+
+        if reader_config_options and len(reader_config_options) > 0:
+            return (
+                spark.readStream.format(bronze_dataflow_spec.sourceFormat)
+                .options(**reader_config_options)
+                .load(source_path)
+            )
+        else:
+            return (
+                spark.readStream.format(bronze_dataflow_spec.sourceFormat)
+                .load(source_path)
+            )
+
     @staticmethod
     def get_db_utils(spark):
         """Get databricks utils using DBUtils package."""
diff --git a/tests/test_pipeline_readers.py b/tests/test_pipeline_readers.py
@@ -141,6 +141,30 @@ def test_read_cloud_files_positive(self):
         customer_df = PipelineReaders.read_dlt_cloud_files(self.spark, bronze_dataflow_spec, schema)
         self.assertIsNotNone(customer_df)
 
+    def test_read_delta_positive(self):
+        """Test read_cloud_files positive."""
+        bronze_map = PipelineReadersTests.bronze_dataflow_spec_map
+        source_format_map = {"sourceFormat": "delta"}
+        bronze_map.update(source_format_map)
+        source_details_map = {"sourceDetails": {"path": "tests/resources/delta/customers"}}
+        bronze_map.update(source_details_map)
+        bronze_dataflow_spec = BronzeDataflowSpec(**bronze_map)
+        customer_df = PipelineReaders.read_dlt_delta(self.spark, bronze_dataflow_spec)
+        self.assertIsNotNone(customer_df)
+
+    def test_read_delta_with_read_config_positive(self):
+        """Test read_cloud_files positive."""
+        bronze_map = PipelineReadersTests.bronze_dataflow_spec_map
+        source_format_map = {"sourceFormat": "delta"}
+        bronze_map.update(source_format_map)
+        source_details_map = {"sourceDetails": {"path": "tests/resources/delta/customers"}}
+        bronze_map.update(source_details_map)
+        reader_config = {"readerConfigOptions": {"maxFilesPerTrigger": "1"}}
+        bronze_map.update(reader_config)
+        bronze_dataflow_spec = BronzeDataflowSpec(**bronze_map)
+        customer_df = PipelineReaders.read_dlt_delta(self.spark, bronze_dataflow_spec)
+        self.assertIsNotNone(customer_df)        
+
     @patch.object(PipelineReaders, "get_db_utils", return_value=dbutils)
     @patch.object(dbutils, "secrets.get", return_value={"called"})
     def test_get_eventhub_kafka_options(self, get_db_utils, dbutils):
