Merge pull request #25 from databrickslabs/feature/dlt-meta-uc-cli

Added
- dtlmeta labs cli feature
- unit test coverage for uc_enabled feature in onboarding and dataflowpipeline

Author: ravi-databricks

.coveragerc

@@ -5,6 +5,9 @@ include = src/*.py
 omit =
     */site-packages/*
     tests/*
+    src/install.py
+    src/config.py
+    src/cli.py
 
 [report]
 exclude_lines =

.gitignore

@@ -151,4 +151,7 @@ deployment-merged.yaml
 .vscode/
 
 # ignore integration test onboarding file.
-integration-tests/conf/dlt-meta/onboarding.json
+integration-tests/conf/dlt-meta/onboarding.json
+
+.databricks
+.databricks-login.json

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 [Please read through the Keep a Changelog (~5min)](https://keepachangelog.com/en/1.0.0/).
 
+## [v.0.0.5] 
+- enabled UC (link to PR)
+- databricks labs cli integration (link to PR)
+
 ## [v0.0.4] - 2023-10-09
 ### Added
 - Functionality to introduce an new option for event hub configuration. Namely a source_details option 'eventhub.accessKeySecretName' to properly construct the eh_shared_key_value properly. Without this option, there were errors while connecting to the event hub service (linked to [issue-13 - java.lang.RuntimeException: non-nullable field authBytes was serialized as null #13](https://github.com/databrickslabs/dlt-meta/issues/13))

Makefile

@@ -0,0 +1,6 @@
+clean:
+	rm -fr build .databricks dlt_meta.egg-info
+
+dev:
+	python3 -m venv .databricks
+	.databricks/bin/python -m pip install -e .

README.md

@@ -68,6 +68,9 @@ With this framework you need to record the source and target metadata in an onbo
 
 ## Getting Started
 Refer to the [Getting Started](https://databrickslabs.github.io/dlt-meta/getting_started)
+### Databricks Labs DLT-META CLI lets you run onboard and deploy in interactive python terminal
+- ```databricks labs dlt-meta onboard``` 
+- ```databricks labs dlt-meta deploy```
 
 ## More questions
 Refer to the [FAQ](https://databrickslabs.github.io/dlt-meta/faq)

demo/launch_dais_demo.py

@@ -18,11 +18,11 @@ class DLTMETADAISDemo(DLTMETARunner):
     - workspace_client: Databricks workspace client
     - base_dir: base directory
     """
-    def __init__(self, args, workspace_client, base_dir):
+    def __init__(self, args, ws, base_dir):
         self.args = args
-        self.workspace_client = workspace_client
+        self.ws = ws
         self.base_dir = base_dir
-    
+
     def init_runner_conf(self) -> DLTMetaRunnerConf:
         """
         Initialize the runner configuration.
@@ -33,13 +33,13 @@ def init_runner_conf(self) -> DLTMetaRunnerConf:
         run_id = uuid.uuid4().hex
         runner_conf = DLTMetaRunnerConf(
             run_id=run_id,
-            username=self.args.__dict__['username'],
+            username=self._my_username(self.ws),
             dbfs_tmp_path=f"{self.args.__dict__['dbfs_path']}/{run_id}",
             int_tests_dir="file:./demo",
             dlt_meta_schema=f"dlt_meta_dataflowspecs_demo_{run_id}",
             bronze_schema=f"dlt_meta_bronze_dais_demo_{run_id}",
             silver_schema=f"dlt_meta_silver_dais_demo_{run_id}",
-            runners_nb_path=f"/Users/{self.args.__dict__['username']}/dlt_meta_dais_demo/{run_id}",
+            runners_nb_path=f"/Users/{self._my_username(self.ws)}/dlt_meta_dais_demo/{run_id}",
             node_type_id=cloud_node_type_id_dict[self.args.__dict__['cloud_provider_name']],
             dbr_version=self.args.__dict__['dbr_version'],
             cloudfiles_template="demo/conf/onboarding.template",
@@ -70,7 +70,7 @@ def run(self, runner_conf: DLTMetaRunnerConf):
             print(e)
         # finally:
         #     self.clean_up(runner_conf)
-            
+
     def launch_workflow(self, runner_conf: DLTMetaRunnerConf):
         """
         Launch the workflow for DLT-META DAIS DEMO.
@@ -82,9 +82,9 @@ def launch_workflow(self, runner_conf: DLTMetaRunnerConf):
         runner_conf.job_id = created_job.job_id
         print(f"Job created successfully. job_id={created_job.job_id}, started run...")
         print(f"Waiting for job to complete. run_id={created_job.job_id}")
-        run_by_id = self.workspace_client.jobs.run_now(job_id=created_job.job_id).result()
+        run_by_id = self.ws.jobs.run_now(job_id=created_job.job_id).result()
         print(f"Job run finished. run_id={run_by_id}")
-            
+
     def create_daisdemo_workflow(self, runner_conf: DLTMetaRunnerConf):
         """
         Create the workflow for DLT-META DAIS DEMO.
@@ -96,7 +96,7 @@ def create_daisdemo_workflow(self, runner_conf: DLTMetaRunnerConf):
         - created_job: created job object
         """
         database, dlt_lib = self.init_db_dltlib(runner_conf)
-        return self.workspace_client.jobs.create(
+        return self.ws.jobs.create(
             name=f"dltmeta_dais_demo-{runner_conf.run_id}",
             tasks=[
                 jobs.Task(
@@ -174,7 +174,6 @@ def create_daisdemo_workflow(self, runner_conf: DLTMetaRunnerConf):
 
 
 dais_args_map = {"--profile": "provide databricks cli profile name, if not provide databricks_host and token",
-                 "--username": "provide databricks username, this is required to upload runners notebook",
                  "--source": "provide source. Supported values are cloudfiles, eventhub, kafka",
                  "--uc_catalog_name": "provide databricks uc_catalog name, \
                      this is required to create volume, schema, table",
@@ -183,7 +182,7 @@ def create_daisdemo_workflow(self, runner_conf: DLTMetaRunnerConf):
                  "--dbfs_path": "Provide databricks workspace dbfs path where you want run integration tests \
                         e.g --dbfs_path=dbfs:/tmp/DLT-META/"}
 
-dais_mandatory_args = ["username", "source", "cloud_provider_name",
+dais_mandatory_args = ["source", "cloud_provider_name",
                        "dbr_version", "dbfs_path"]
 
 

demo/launch_techsummit_demo.py

@@ -62,11 +62,11 @@ class DLTMETATechSummitDemo(DLTMETARunner):
     - workspace_client: Databricks workspace client.
     - base_dir: Base directory.
     """
-    def __init__(self, args, workspace_client, base_dir):
+    def __init__(self, args, ws, base_dir):
         self.args = args
-        self.workspace_client = workspace_client
+        self.ws = ws
         self.base_dir = base_dir
-    
+
     def init_runner_conf(self) -> TechsummitRunnerConf:
         """
         Initializes the TechsummitRunnerConf object with the provided configuration parameters.
@@ -78,19 +78,20 @@ def init_runner_conf(self) -> TechsummitRunnerConf:
         print(f"run_id={run_id}")
         runner_conf = TechsummitRunnerConf(
             run_id=run_id,
+            username=self._my_username(self.ws),
             dbfs_tmp_path=f"{self.args.__dict__['dbfs_path']}/{run_id}",
             dlt_meta_schema=f"dlt_meta_dataflowspecs_demo_{run_id}",
             bronze_schema=f"dlt_meta_bronze_demo_{run_id}",
             silver_schema=f"dlt_meta_silver_demo_{run_id}",
             runners_full_local_path='./demo/dbc/tech_summit_dlt_meta_runners.dbc',
-            runners_nb_path=f"/Users/{self.args.__dict__['username']}/dlt_meta_techsummit_demo/{run_id}",
+            runners_nb_path=f"/Users/{self._my_username(self.ws)}/dlt_meta_techsummit_demo/{run_id}",
             node_type_id=cloud_node_type_id_dict[self.args.__dict__['cloud_provider_name']],
             dbr_version=self.args.__dict__['dbr_version'],
             env="prod",
             table_count=self.args.__dict__['table_count'] if self.args.__dict__['table_count'] else "100",
             table_column_count=(self.args.__dict__['table_column_count'] if self.args.__dict__['table_column_count']
                                 else "5"),
-            table_data_rows_count=(self.args.__dict__['table_data_rows_count'] 
+            table_data_rows_count=(self.args.__dict__['table_data_rows_count']
                                    if self.args.__dict__['table_data_rows_count'] else "10"),
             worker_nodes=self.args.__dict__['worker_nodes'] if self.args.__dict__['worker_nodes'] else "4",
             source=self.args.__dict__['source'],
@@ -110,24 +111,24 @@ def init_dltmeta_runner_conf(self, runner_conf: DLTMetaRunnerConf):
         - runner_conf: The DLTMetaRunnerConf object containing the runner configuration parameters.
         """
         fp = open(runner_conf.runners_full_local_path, "rb")
-        self.workspace_client.workspace.mkdirs(runner_conf.runners_nb_path)
-        self.workspace_client.workspace.upload(path=f"{runner_conf.runners_nb_path}/runners",
-                                               format=ImportFormat.DBC, content=fp.read())
+        self.ws.workspace.mkdirs(runner_conf.runners_nb_path)
+        self.ws.workspace.upload(path=f"{runner_conf.runners_nb_path}/runners",
+                                 format=ImportFormat.DBC, content=fp.read())
         if runner_conf.uc_catalog_name:
-            SchemasAPI(self.workspace_client.api_client).create(catalog_name=runner_conf.uc_catalog_name,
-                                                                name=runner_conf.dlt_meta_schema,
-                                                                comment="dlt_meta framework schema")
-            volume_info = self.workspace_client.volumes.create(catalog_name=runner_conf.uc_catalog_name,
-                                                               schema_name=runner_conf.dlt_meta_schema,
-                                                               name=runner_conf.uc_volume_name,
-                                                               volume_type=VolumeType.MANAGED)
+            SchemasAPI(self.ws.api_client).create(catalog_name=runner_conf.uc_catalog_name,
+                                                  name=runner_conf.dlt_meta_schema,
+                                                  comment="dlt_meta framework schema")
+            volume_info = self.ws.volumes.create(catalog_name=runner_conf.uc_catalog_name,
+                                                 schema_name=runner_conf.dlt_meta_schema,
+                                                 name=runner_conf.uc_volume_name,
+                                                 volume_type=VolumeType.MANAGED)
             runner_conf.volume_info = volume_info
-            SchemasAPI(self.workspace_client.api_client).create(catalog_name=runner_conf.uc_catalog_name,
-                                                                name=runner_conf.bronze_schema,
-                                                                comment="bronze_schema")
-            SchemasAPI(self.workspace_client.api_client).create(catalog_name=runner_conf.uc_catalog_name,
-                                                                name=runner_conf.silver_schema,
-                                                                comment="silver_schema")
+            SchemasAPI(self.ws.api_client).create(catalog_name=runner_conf.uc_catalog_name,
+                                                  name=runner_conf.bronze_schema,
+                                                  comment="bronze_schema")
+            SchemasAPI(self.ws.api_client).create(catalog_name=runner_conf.uc_catalog_name,
+                                                  name=runner_conf.silver_schema,
+                                                  comment="silver_schema")
 
         self.build_and_upload_package(runner_conf)  # comment this line before merging to master
 
@@ -160,9 +161,9 @@ def launch_workflow(self, runner_conf: DLTMetaRunnerConf):
         runner_conf.job_id = created_job.job_id
         print(f"Job created successfully. job_id={created_job.job_id}, started run...")
         print(f"Waiting for job to complete. run_id={created_job.job_id}")
-        run_by_id = self.workspace_client.jobs.run_now(job_id=created_job.job_id).result()
+        run_by_id = self.ws.jobs.run_now(job_id=created_job.job_id).result()
         print(f"Job run finished. run_id={run_by_id}")
-        
+
     def create_techsummit_demo_workflow(self, runner_conf: TechsummitRunnerConf):
         """
         Creates the workflow for the Techsummit Demo by defining the tasks and their dependencies.
@@ -174,7 +175,7 @@ def create_techsummit_demo_workflow(self, runner_conf: TechsummitRunnerConf):
         - created_job: The created job object.
         """
         database, dlt_lib = self.init_db_dltlib(runner_conf)
-        return self.workspace_client.jobs.create(
+        return self.ws.jobs.create(
             name=f"dlt-meta-dais-demo-{runner_conf.run_id}",
             tasks=[
                 jobs.Task(
@@ -240,7 +241,6 @@ def create_techsummit_demo_workflow(self, runner_conf: TechsummitRunnerConf):
 
 
 techsummit_args_map = {"--profile": "provide databricks cli profile name, if not provide databricks_host and token",
-                       "--username": "provide databricks username, this is required to upload runners notebook",
                        "--source": "provide --source=cloudfiles",
                        "--uc_catalog_name": "provide databricks uc_catalog name, \
                                             this is required to create volume, schema, table",
@@ -253,7 +253,7 @@ def create_techsummit_demo_workflow(self, runner_conf: TechsummitRunnerConf):
                        "--table_data_rows_count": "table_data_rows_count"
                        }
 
-techsummit_mandatory_args = ["username", "source", "cloud_provider_name", "dbr_version", "dbfs_path"]
+techsummit_mandatory_args = ["source", "cloud_provider_name", "dbr_version", "dbfs_path"]
 
 
 def main():

integration_tests/cleanup_script.py

@@ -19,7 +19,7 @@ def process_arguments():
     parser.add_argument("--profile",
                         help="provide databricks cli profile name, if not provide databricks_host and token")
     parser.add_argument("--uc_catalog_name",
-                        help="provide uc_catalog_name")    
+                        help="provide uc_catalog_name")
     args = parser.parse_args()
     return args
 
@@ -32,7 +32,7 @@ def main():
     # job_list = workspace_client.jobs.list()
     # for job in job_list:
     #     print(f"Deleting job:{job.creator_user_name}")
-        #workspace_client.jobs.delete(job.job_id)
+    # workspace_client.jobs.delete(job.job_id)
     # list = workspace_client.pipelines.list_pipelines(filter="name like 'dlt-meta-integration-test-silver-%'")
     # print("List of pipelines:")
     # for pipeline in list:
@@ -42,7 +42,7 @@ def main():
     # print("List of pipelines:")
     # for pipeline in list:
     #     print(f"id = {pipeline.pipeline_id} , name = {pipeline.name}")
-    #     workspace_client.pipelines.delete(pipeline.pipeline_id)        
+    #     workspace_client.pipelines.delete(pipeline.pipeline_id)
     uc_catalog_name = args.uc_catalog_name
     schema_list = workspace_client.schemas.list(catalog_name=uc_catalog_name)
     for schema in schema_list:
@@ -60,4 +60,4 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    main()