Support writing to Databricks delta table using a module #1483
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
drobison00
requested changes
Jan 30, 2024
| delta_table_write_mode = module_config.get("DELTA_WRITE_MODE", "append") | ||
| databricks_host=module_config.get("DATABRICKS_HOST", None) | ||
| databricks_token=module_config.get("DATABRICKS_TOKEN", None) | ||
| databricks_cluster_id=module_config.get("DATABRICKS_CLUSTER_ID", None) |
There was a problem hiding this comment.
Please add error checking here for any config parameter that might cause the builder.remote call to fail later, this gives us more immediate feedback during module loading instead of runtime.
| @@ -0,0 +1,113 @@ | |||
| # Copyright (c) 2022-2023, NVIDIA CORPORATION. | |||
There was a problem hiding this comment.
Suggested change
| # Copyright (c) 2022-2023, NVIDIA CORPORATION. | |
| # Copyright (c) 2024, NVIDIA CORPORATION. |
| except ImportError as import_exc: | ||
| IMPORT_EXCEPTION = import_exc | ||
|
|
||
| def _extract_schema_from_pandas_dataframe(df: pd.DataFrame) -> "sql_types.StructType": |
There was a problem hiding this comment.
Lets add at least one unit test for this helper function; mocking out everything to deal with the spark internals in the module's work unit probably isn't worth it right now, but having a test file gives us a spot for sprouting new tests later.
| from databricks.connect import DatabricksSession | ||
| from pyspark.sql import types as sql_types | ||
| except ImportError as import_exc: | ||
| IMPORT_EXCEPTION = import_exc |
There was a problem hiding this comment.
Is 'IMPORT_EXCEPTION' used anywhere? If we're not handling the failure to import via some kind of work around, then lets log an error here and re-raise the ImportErrror
Comment on lines
+69
to
+85
| module_config = builder.get_current_module_config() | ||
| """module_config contains all the required configuration parameters, that would otherwise be passed to the stage. | ||
| Parameters | ||
| ---------- | ||
| config : morpheus.config.Config | ||
| Pipeline configuration instance. | ||
| delta_path : str, default None | ||
| Path of the delta table where the data need to be written or updated. | ||
| databricks_host : str, default None | ||
| URL of Databricks host to connect to. | ||
| databricks_token : str, default None | ||
| Access token for Databricks cluster. | ||
| databricks_cluster_id : str, default None | ||
| Databricks cluster to be used to query the data as per SQL provided. | ||
| delta_table_write_mode: str, default "append" | ||
| Delta table write mode for storing data. | ||
| """ |
There was a problem hiding this comment.
Suggested change
| module_config = builder.get_current_module_config() | |
| """module_config contains all the required configuration parameters, that would otherwise be passed to the stage. | |
| Parameters | |
| ---------- | |
| config : morpheus.config.Config | |
| Pipeline configuration instance. | |
| delta_path : str, default None | |
| Path of the delta table where the data need to be written or updated. | |
| databricks_host : str, default None | |
| URL of Databricks host to connect to. | |
| databricks_token : str, default None | |
| Access token for Databricks cluster. | |
| databricks_cluster_id : str, default None | |
| Databricks cluster to be used to query the data as per SQL provided. | |
| delta_table_write_mode: str, default "append" | |
| Delta table write mode for storing data. | |
| """ | |
| """ | |
| Parameters | |
| ---------- | |
| config : morpheus.config.Config | |
| Pipeline configuration instance. | |
| delta_path : str, default None | |
| Path of the delta table where the data need to be written or updated. | |
| databricks_host : str, default None | |
| URL of Databricks host to connect to. | |
| databricks_token : str, default None | |
| Access token for Databricks cluster. | |
| databricks_cluster_id : str, default None | |
| Databricks cluster to be used to query the data as per SQL provided. | |
| delta_table_write_mode: str, default "append" | |
| Delta table write mode for storing data. | |
| """ | |
| module_config = builder.get_current_module_config() |
drobison00
added
non-breaking
drobison00
reviewed
Jan 30, 2024
mdemoret-nv
changed the title
|
/ok to test |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Adds support to write to delta lake table using module, this will close #1482