fix/Databricks example. (#193)

Author: hubert-rutkowski85

CHANGELOG.md

@@ -1,4 +1,4 @@
-## 0.1.1-dev4
+## 0.1.1-dev5
 
 ### Enhancements
 
@@ -11,7 +11,9 @@
 * **Add Delta Table destination to v2**
 * **Migrate Slack Source Connector to V2**
 
->>>>>>> 9214214 (bump changelog)
+### Fixes
+
+* **Fix Databricks Volumes destination** Fix for filenames to not be hashes.
 
 ## 0.1.0
 

test/integration/connectors/databricks_tests/test_volumes_native.py

@@ -16,7 +16,7 @@
     source_connector_validation,
 )
 from test.integration.utils import requires_env
-from unstructured_ingest.v2.interfaces import FileData
+from unstructured_ingest.v2.interfaces import FileData, SourceIdentifiers
 from unstructured_ingest.v2.processes.connectors.databricks.volumes_native import (
     CONNECTOR_TYPE,
     DatabricksNativeVolumesAccessConfig,
@@ -139,7 +139,11 @@ def validate_upload(client: WorkspaceClient, catalog: str, volume: str, volume_p
 async def test_volumes_native_destination(upload_file: Path):
     env_data = get_env_data()
     volume_path = f"databricks-volumes-test-output-{uuid.uuid4()}"
-    mock_file_data = FileData(identifier="mock file data", connector_type=CONNECTOR_TYPE)
+    file_data = FileData(
+        source_identifiers=SourceIdentifiers(fullpath=upload_file.name, filename=upload_file.name),
+        connector_type=CONNECTOR_TYPE,
+        identifier="mock file data",
+    )
     with databricks_destination_context(
         volume="test-platform", volume_path=volume_path, env_data=env_data
     ) as workspace_client:
@@ -153,9 +157,9 @@ async def test_volumes_native_destination(upload_file: Path):
             ),
         )
         if uploader.is_async():
-            await uploader.run_async(path=upload_file, file_data=mock_file_data)
+            await uploader.run_async(path=upload_file, file_data=file_data)
         else:
-            uploader.run(path=upload_file, file_data=mock_file_data)
+            uploader.run(path=upload_file, file_data=file_data)
 
         validate_upload(
             client=workspace_client,

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "0.1.1-dev4"  # pragma: no cover
+__version__ = "0.1.1-dev5"  # pragma: no cover

unstructured_ingest/v2/examples/databricks_volumes.py renamed to unstructured_ingest/v2/examples/databricks_volumes_dest.py

@@ -5,18 +5,17 @@
 from unstructured_ingest.v2.logger import logger
 from unstructured_ingest.v2.pipeline.pipeline import Pipeline
 from unstructured_ingest.v2.processes.chunker import ChunkerConfig
-from unstructured_ingest.v2.processes.connectors.databricks_volumes import (
+from unstructured_ingest.v2.processes.connectors.databricks.volumes_native import (
     CONNECTOR_TYPE,
-    DatabricksVolumesAccessConfig,
-    DatabricksVolumesConnectionConfig,
-    DatabricksVolumesUploaderConfig,
+    DatabricksNativeVolumesAccessConfig,
+    DatabricksNativeVolumesConnectionConfig,
+    DatabricksNativeVolumesUploaderConfig,
 )
 from unstructured_ingest.v2.processes.connectors.local import (
     LocalConnectionConfig,
     LocalDownloaderConfig,
     LocalIndexerConfig,
 )
-from unstructured_ingest.v2.processes.embedder import EmbedderConfig
 from unstructured_ingest.v2.processes.partitioner import PartitionerConfig
 
 base_path = Path(__file__).parent.parent.parent.parent
@@ -29,25 +28,26 @@
     logger.info(f"writing all content in: {work_dir.resolve()}")
     Pipeline.from_configs(
         context=ProcessorConfig(work_dir=str(work_dir.resolve())),
-        indexer_config=LocalIndexerConfig(input_path=str(docs_path.resolve()) + "/multisimple/"),
+        indexer_config=LocalIndexerConfig(input_path=str(docs_path.resolve()) + "/fake-text.txt"),
         downloader_config=LocalDownloaderConfig(download_dir=download_path),
         source_connection_config=LocalConnectionConfig(),
         partitioner_config=PartitionerConfig(strategy="fast"),
         chunker_config=ChunkerConfig(
-            chunking_strategy="by_title",
-            chunk_include_orig_elements=False,
-            chunk_max_characters=1500,
-            chunk_multipage_sections=True,
+            chunking_strategy="basic",
         ),
-        embedder_config=EmbedderConfig(embedding_provider="huggingface"),
-        destination_connection_config=DatabricksVolumesConnectionConfig(
-            access_config=DatabricksVolumesAccessConfig(
-                username=os.environ["DATABRICKS_USERNAME"],
-                password=os.environ["DATABRICKS_PASSWORD"],
+        embedder_config=None,
+        destination_connection_config=DatabricksNativeVolumesConnectionConfig(
+            access_config=DatabricksNativeVolumesAccessConfig(
+                client_id=os.environ["DATABRICKS_CLIENT_ID"],
+                client_secret=os.environ["DATABRICKS_CLIENT_SECRET"],
             ),
             host=os.environ["DATABRICKS_HOST"],
+            catalog=os.environ["DATABRICKS_CATALOG"],
+            volume=os.environ["DATABRICKS_VOLUME"],
+            volume_path=os.environ["DATABRICKS_VOLUME_PATH"],
         ),
-        uploader_config=DatabricksVolumesUploaderConfig(
+        uploader_config=DatabricksNativeVolumesUploaderConfig(
+            overwrite=True,
             catalog=os.environ["DATABRICKS_CATALOG"],
             volume=os.environ["DATABRICKS_VOLUME"],
             volume_path=os.environ["DATABRICKS_VOLUME_PATH"],

unstructured_ingest/v2/examples/databricks_volumes_source.py

@@ -0,0 +1,53 @@
+import os
+from pathlib import Path
+
+from unstructured_ingest.v2.interfaces import ProcessorConfig
+from unstructured_ingest.v2.logger import logger
+from unstructured_ingest.v2.pipeline.pipeline import Pipeline
+from unstructured_ingest.v2.processes.chunker import ChunkerConfig
+from unstructured_ingest.v2.processes.connectors.databricks.volumes_native import (
+    CONNECTOR_TYPE,
+    DatabricksNativeVolumesAccessConfig,
+    DatabricksNativeVolumesConnectionConfig,
+    DatabricksNativeVolumesDownloaderConfig,
+    DatabricksNativeVolumesIndexerConfig,
+)
+from unstructured_ingest.v2.processes.connectors.local import (
+    LocalUploaderConfig,
+)
+from unstructured_ingest.v2.processes.partitioner import PartitionerConfig
+
+base_path = Path(__file__).parent.parent.parent.parent
+docs_path = base_path / "example-docs"
+work_dir = base_path / "tmp_ingest" / CONNECTOR_TYPE
+output_path = work_dir / "output"
+download_path = work_dir / "download"
+
+if __name__ == "__main__":
+    logger.info(f"writing all content in: {work_dir.resolve()}")
+    Pipeline.from_configs(
+        context=ProcessorConfig(work_dir=str(work_dir.resolve())),
+        indexer_config=DatabricksNativeVolumesIndexerConfig(
+            host=os.environ["DATABRICKS_HOST"],
+            catalog=os.environ["DATABRICKS_CATALOG"],
+            volume=os.environ["DATABRICKS_VOLUME"],
+            volume_path=os.environ["DATABRICKS_VOLUME_PATH"],
+        ),
+        downloader_config=DatabricksNativeVolumesDownloaderConfig(download_dir=download_path),
+        source_connection_config=DatabricksNativeVolumesConnectionConfig(
+            access_config=DatabricksNativeVolumesAccessConfig(
+                client_id=os.environ["DATABRICKS_CLIENT_ID"],
+                client_secret=os.environ["DATABRICKS_CLIENT_SECRET"],
+            ),
+            host=os.environ["DATABRICKS_HOST"],
+            catalog=os.environ["DATABRICKS_CATALOG"],
+            volume=os.environ["DATABRICKS_VOLUME"],
+            volume_path=os.environ["DATABRICKS_VOLUME_PATH"],
+        ),
+        partitioner_config=PartitionerConfig(strategy="fast"),
+        chunker_config=ChunkerConfig(
+            chunking_strategy="basic",
+        ),
+        embedder_config=None,
+        uploader_config=LocalUploaderConfig(output_dir=str(output_path.resolve())),
+    ).run()

unstructured_ingest/v2/processes/connectors/databricks/volumes.py

@@ -166,7 +166,7 @@ def precheck(self) -> None:
             raise DestinationConnectionError(f"failed to validate connection: {e}")
 
     def run(self, path: Path, file_data: FileData, **kwargs: Any) -> None:
-        output_path = os.path.join(self.upload_config.path, path.name)
+        output_path = os.path.join(self.upload_config.path, file_data.source_identifiers.filename)
         with open(path, "rb") as elements_file:
             self.connection_config.get_client().files.upload(
                 file_path=output_path,