Databricks Volumes v2 source connector (#288)

Author: Paul-Cornell

api-reference/ingest/source-connectors/databricks-volumes.mdx

@@ -0,0 +1,23 @@
+---
+title: Databricks Volumes
+---
+
+import NewDocument from '/snippets/general-shared-text/new-document.mdx';
+
+<NewDocument />
+
+import SharedContentDatabricksVolumes from '/snippets/sc-shared-text/databricks-volumes-cli-api.mdx';
+import SharedAPIKeyURL from '/snippets/general-shared-text/api-key-url.mdx';
+
+<SharedContentDatabricksVolumes/>
+<SharedAPIKeyURL/>
+
+Now call the Unstructured Ingest CLI or the Unstructured Ingest Python library. The destination connector can be any of the ones supported. This example uses the local destination connector:
+
+import DatabricksVolumesAPISh from '/snippets/source_connectors/databricks-volumes.sh.mdx';
+import DatabricksVolumesAPIPyV2 from '/snippets/source_connectors/databricks-volumes.v2.py.mdx';
+
+<CodeGroup>
+  <DatabricksVolumesAPISh />
+  <DatabricksVolumesAPIPyV2 />
+</CodeGroup>

mint.json

@@ -150,6 +150,7 @@
               "open-source/ingest/source-connectors/box",
               "open-source/ingest/source-connectors/confluence",
               "open-source/ingest/source-connectors/couchbase",
+              "open-source/ingest/source-connectors/databricks-volumes",
               "open-source/ingest/source-connectors/delta-table",
               "open-source/ingest/source-connectors/discord",
               "open-source/ingest/source-connectors/dropbox",
@@ -300,6 +301,7 @@
               "api-reference/ingest/source-connectors/box",
               "api-reference/ingest/source-connectors/confluence",
               "api-reference/ingest/source-connectors/couchbase",
+              "api-reference/ingest/source-connectors/databricks-volumes",
               "api-reference/ingest/source-connectors/delta-table",
               "api-reference/ingest/source-connectors/discord",
               "api-reference/ingest/source-connectors/dropbox",

open-source/ingest/source-connectors/databricks-volumes.mdx

@@ -0,0 +1,27 @@
+---
+title: Databricks Volumes
+---
+
+import NewDocument from '/snippets/general-shared-text/new-document.mdx';
+
+<NewDocument />
+
+import SharedContentDatabricksVolumes from '/snippets/sc-shared-text/databricks-volumes-cli-api.mdx';
+
+<SharedContentDatabricksVolumes/>
+
+Now call the Unstructured Ingest CLI or the Unstructured Ingest Python library. The destination connector can be any of the ones supported. This example uses the local destination connector.
+
+This example sends data to Unstructured API services for processing by default. To process data locally instead, see the instructions at the end of this page.
+
+import DatabricksVolumesSh from '/snippets/source_connectors/databricks-volumes.sh.mdx';
+import DatabricksVolumesPyV2 from '/snippets/source_connectors/databricks-volumes.v2.py.mdx';
+
+<CodeGroup>
+  <DatabricksVolumesSh />
+  <DatabricksVolumesPyV2 />
+</CodeGroup>
+
+import SharedPartitionByAPIOSS from '/snippets/ingest-configuration-shared/partition-by-api-oss.mdx';
+
+<SharedPartitionByAPIOSS/>

snippets/destination_connectors/databricks_volumes.sh.mdx

@@ -11,16 +11,11 @@ unstructured-ingest \
     --partition-endpoint $UNSTRUCTURED_API_URL \
     --strategy hi_res \
     --additional-partition-args="{\"split_pdf_page\":\"true\", \"split_pdf_allow_failed\":\"true\", \"split_pdf_concurrency_level\": 15}" \
-    --chunk-by-api \
     --chunking-strategy by_title \
-    --chunk-api-key $UNSTRUCTURED_API_KEY \
-    --chunking-endpoint $UNSTRUCTURED_API_URL \ 
     --embedding-provider huggingface \
-    --embedding-model-name sentence-transformers/all-mpnet-base-v2 \
   databricks-volumes \
+    --profile $DATABRICKS_PROFILE \
     --host $DATABRICKS_HOST \
-    --token $DATABRICKS_TOKEN \
-    --cluster-id $DATABRICKS_CLUSTER_ID \
     --catalog $DATABRICKS_CATALOG \
     --schema $DATABRICKS_SCHEMA \
     --volume $DATABRICKS_VOLUME \

snippets/destination_connectors/databricks_volumes.v1.py.mdx

@@ -26,8 +26,7 @@ def get_writer() -> Writer:
         connector_config=SimpleDatabricksVolumesConfig(
             host=os.getenv("DATABRICKS_HOST"),
             access_config=DatabricksVolumesAccessConfig(
-                token=os.getenv("DATABRICKS_TOKEN"),
-                cluster_id=os.getenv("DATABRICKS_CLUSTER_ID")
+                token=os.getenv("DATABRICKS_TOKEN")
             ),
         ),
         write_config=DatabricksVolumesWriteConfig(
@@ -63,8 +62,7 @@ if __name__ == "__main__":
             chunking_strategy="by_title",
         ),
         embedding_config=EmbeddingConfig(
-            provider="huggingface",
-            model_name="sentence-transformers/all-mpnet-base-v2",
+            provider="huggingface"
         ),
         writer=writer,
         writer_kwargs={},

snippets/destination_connectors/databricks_volumes.v2.py.mdx

@@ -4,11 +4,34 @@ import os
 from unstructured_ingest.v2.pipeline.pipeline import Pipeline
 from unstructured_ingest.v2.interfaces import ProcessorConfig
 
+# For all supported Databricks authentication types, you can import this:
 from unstructured_ingest.v2.processes.connectors.databricks_volumes import (
     DatabricksVolumesConnectionConfig,
     DatabricksVolumesAccessConfig,
     DatabricksVolumesUploaderConfig
 )
+
+# Alternatively, for supported Databricks on AWS authentication types only, you can import this:
+# from unstructured_ingest.v2.processes.connectors.databricks.volumes_aws import (
+#     DatabricksAWSVolumesConnectionConfig,
+#     DatabricksAWSVolumesAccessConfig,
+#     DatabricksAWSVolumesUploaderConfig
+# )
+
+# Alternatively, for supported Azure Databricks authentication types only, you can import this:
+# from unstructured_ingest.v2.processes.connectors.databricks.volumes_azure import (
+#     DatabricksAzureVolumesConnectionConfig,
+#     DatabricksAzureVolumesAccessConfig,
+#     DatabricksAzureVolumesUploaderConfig
+# )
+
+# Alternatively, for supported Databricks on Google Cloud authentication types only, you can import this:
+# from unstructured_ingest.v2.processes.connectors.databricks.volumes_gcp import (
+#     DatabricksGoogleVolumesConnectionConfig,
+#     DatabricksGoogleVolumesAccessConfig,
+#     DatabricksGoogleVolumesUploaderConfig
+# )
+
 from unstructured_ingest.v2.processes.connectors.local import (
     LocalIndexerConfig,
     LocalDownloaderConfig,
@@ -37,28 +60,63 @@ if __name__ == "__main__":
                 "split_pdf_concurrency_level": 15
             }
         ),
-        chunker_config=ChunkerConfig(
-            chunk_by_api=True,
-            chunk_api_key=os.getenv("UNSTRUCTURED_API_KEY"),
-            chunking_endpoint=os.getenv("UNSTRUCTURED_API_URL"),
-            chunking_strategy="by_title"
-        ),
-        embedder_config=EmbedderConfig(
-            embedding_provider="huggingface",
-            embedding_model_name="sentence-transformers/all-mpnet-base-v2"
-        ),
+        chunker_config=ChunkerConfig(chunking_strategy="by_title"),
+        embedder_config=EmbedderConfig(embedding_provider="huggingface"),
+        # For specifying a Databricks configuration profile:
         destination_connection_config=DatabricksVolumesConnectionConfig(
-            access_config=DatabricksVolumesAccessConfig(
-                token=os.getenv("DATABRICKS_TOKEN"),
-                cluster_id=os.getenv("DATABRICKS_CLUSTER_ID")
-            ),
-            host=os.getenv("DATABRICKS_HOST")
-        ),
-        uploader_config=DatabricksVolumesUploaderConfig(
+            access_config=DatabricksVolumesAccessConfig(profile=os.getenv("DATABRICKS_PROFILE")),
+            host=os.getenv("DATABRICKS_HOST"),
             catalog=os.getenv("DATABRICKS_CATALOG"),
             schema=os.getenv("DATABRICKS_SCHEMA"),
             volume=os.getenv("DATABRICKS_VOLUME"),
             volume_path=os.getenv("DATABRICKS_VOLUME_PATH")
-        )
+        ),
+        uploader_config=DatabricksVolumesUploaderConfig(overwrite=True)
+        # Other examples:
+        #
+        # For Databricks on AWS, with Databricks personal access token authentication:
+        # destination_connection_config=DatabricksAWSVolumesConnectionConfig(
+        #     access_config=DatabricksAWSVolumesAccessConfig(token=os.getenv("DATABRICKS_TOKEN")),
+        #     host=os.getenv("DATABRICKS_HOST")
+        # ),
+        # uploader_config=DatabricksAWSVolumesUploaderConfig(
+        #     catalog=os.getenv("DATABRICKS_CATALOG"),
+        #     schema=os.getenv("DATABRICKS_SCHEMA"),
+        #     volume=os.getenv("DATABRICKS_VOLUME"),
+        #     volume_path=os.getenv("DATABRICKS_VOLUME_PATH"),
+        #     overwrite=True
+        # )
+        #
+        # For Azure Databricks, with Microsoft Entra ID service principal authentication:
+        # destination_connection_config=DatabricksAzureVolumesConnectionConfig(
+        #     access_config=DatabricksAzureVolumesAccessConfig(
+        #         azure_client_id=os.getenv("ARM_CLIENT_ID"),
+        #         azure_client_secret=os.getenv("ARM_CLIENT_SECRET"),
+        #         azure_tenant_id=os.getenv("ARM_TENANT_ID")   
+        #     ),
+        #     host=os.getenv("DATABRICKS_HOST")
+        # ),
+        # uploader_config=DatabricksAzureVolumesUploaderConfig(
+        #     catalog=os.getenv("DATABRICKS_CATALOG"),
+        #     schema=os.getenv("DATABRICKS_SCHEMA"),
+        #     volume=os.getenv("DATABRICKS_VOLUME"),
+        #     volume_path=os.getenv("DATABRICKS_VOLUME_PATH"),
+        #     overwrite=True
+        # )
+        #
+        # For Databricks on Google Cloud, with Google Cloud Platform credentials authentication:
+        # destination_connection_config=DatabricksGoogleVolumesConnectionConfig(
+        #     access_config=DatabricksGoogleVolumesAccessConfig(
+        #         google_service_account=os.getenv("GOOGLE_CREDENTIALS")
+        #     ),
+        #     host=os.getenv("DATABRICKS_HOST")
+        # ),
+        # uploader_config=DatabricksAWSVolumesUploaderConfig(
+        #     catalog=os.getenv("DATABRICKS_CATALOG"),
+        #     schema=os.getenv("DATABRICKS_SCHEMA"),
+        #     volume=os.getenv("DATABRICKS_VOLUME"),
+        #     volume_path=os.getenv("DATABRICKS_VOLUME_PATH"),
+        #     overwrite=True
+        # )
     ).run()
 ```

snippets/general-shared-text/databricks-volumes-cli-api.mdx

@@ -11,7 +11,6 @@ import AdditionalIngestDependencies from '/snippets/general-shared-text/ingest-d
 The following environment variables:
 
 - `DATABRICKS_HOST` - The Databricks host URL, represented by `--host` (CLI) or `host` (Python).
-- `DATABRICKS_CLUSTER_ID` - The Databricks compute resource ID, represented by `--cluster-id` (CLI) or `cluster_id` (Python).
 - `DATABRICKS_CATALOG` - The Databricks catalog name for the Volume, represented by `--catalog` (CLI) or `catalog` (Python).
 - `DATABRICKS_SCHEMA` - The Databricks schema name for the Volume, represented by `--schema` (CLI) or `schema` (Python). If not specified, `default` is used.
 - `DATABRICKS_VOLUME` - The Databricks Volume name, represented by `--volume` (CLI) or `volume` (Python).

snippets/sc-shared-text/databricks-volumes-cli-api.mdx

@@ -0,0 +1,9 @@
+Connect Databricks Volumes to your preprocessing pipeline, and use the Unstructured Ingest CLI or the Unstructured Ingest Python library to batch process all your documents and store structured outputs locally on your filesystem.
+
+You will need: 
+
+import SharedDatabricksVolumes from '/snippets/general-shared-text/databricks-volumes.mdx';
+import SharedDatabricksVolumesCLIAPI from '/snippets/general-shared-text/databricks-volumes-cli-api.mdx';
+
+<SharedDatabricksVolumes />
+<SharedDatabricksVolumesCLIAPI />

snippets/source_connectors/databricks-volumes.sh.mdx

@@ -0,0 +1,23 @@
+```bash CLI
+#!/usr/bin/env bash
+
+# Chunking and embedding are optional.
+
+unstructured-ingest \
+  databricks-volumes \
+    --profile $DATABRICKS_PROFILE \
+    --host $DATABRICKS_HOST \
+    --catalog $DATABRICKS_CATALOG \
+    --schema $DATABRICKS_SCHEMA \
+    --volume $DATABRICKS_VOLUME \
+    --volume-path $DATABRICKS_VOLUME_PATH \
+    --partition-by-api \
+    --api-key $UNSTRUCTURED_API_KEY \
+    --partition-endpoint $UNSTRUCTURED_API_URL \
+    --strategy hi_res \
+    --additional-partition-args="{\"split_pdf_page\":\"true\", \"split_pdf_allow_failed\":\"true\", \"split_pdf_concurrency_level\": 15}" \
+    --chunking-strategy by_title \
+    --embedding-provider huggingface \
+  local \
+    --output-dir $LOCAL_FILE_OUTPUT_DIR
+```

snippets/source_connectors/databricks-volumes.v2.py.mdx

@@ -0,0 +1,105 @@
+```python Python Ingest v2
+import os
+
+from unstructured_ingest.v2.pipeline.pipeline import Pipeline
+from unstructured_ingest.v2.interfaces import ProcessorConfig
+
+# For all supported Databricks authentication types, you can import this:
+from unstructured_ingest.v2.processes.connectors.databricks_volumes import (
+    DatabricksVolumesIndexerConfig,
+    DatabricksVolumesConnectionConfig,
+    DatabricksVolumesAccessConfig,
+    DatabricksVolumesDownloaderConfig
+)
+
+# Alternatively, for supported Databricks on AWS authentication types only, you can import this:
+# from unstructured_ingest.v2.processes.connectors.databricks.volumes_aws import (
+#     DatabricksAWSVolumesIndexerConfig,
+#     DatabricksAWSVolumesConnectionConfig,
+#     DatabricksAWSVolumesAccessConfig,
+#     DatabricksAWSVolumesDownloaderConfig
+# )
+
+# Alternatively, for supported Azure Databricks authentication types only, you can import this:
+# from unstructured_ingest.v2.processes.connectors.databricks.volumes_azure import (
+#     DatabricksAzureVolumesIndexerConfig,
+#     DatabricksAzureVolumesConnectionConfig,
+#     DatabricksAzureVolumesAccessConfig,
+#     DatabricksAzureVolumesDownloaderConfig
+# )
+
+# Alternatively, for supported Databricks on Google Cloud authentication types only, you can import this:
+# from unstructured_ingest.v2.processes.connectors.databricks.volumes_gcp import (
+#     DatabricksGoogleVolumesIndexerConfig,
+#     DatabricksGoogleVolumesConnectionConfig,
+#     DatabricksGoogleVolumesAccessConfig,
+#     DatabricksGoogleVolumesDownloaderConfig
+# )
+
+from unstructured_ingest.v2.processes.connectors.local import (
+    LocalConnectionConfig,
+    LocalUploaderConfig
+)
+from unstructured_ingest.v2.processes.partitioner import PartitionerConfig
+from unstructured_ingest.v2.processes.chunker import ChunkerConfig
+from unstructured_ingest.v2.processes.embedder import EmbedderConfig
+
+# Chunking and embedding are optional.
+
+if __name__ == "__main__":
+    Pipeline.from_configs(
+        context=ProcessorConfig(reprocess=True),
+        indexer_config=DatabricksVolumesIndexerConfig(recursive=True),
+        # For specifying a Databricks configuration profile:
+        downloader_config=DatabricksVolumesDownloaderConfig(download_dir=os.getenv("LOCAL_FILE_DOWNLOAD_DIR")),
+        source_connection_config=DatabricksVolumesConnectionConfig(
+            access_config=DatabricksVolumesAccessConfig(profile=os.getenv("DATABRICKS_PROFILE")),
+            host=os.getenv("DATABRICKS_HOST"),
+            catalog=os.getenv("DATABRICKS_CATALOG"),
+            schema=os.getenv("DATABRICKS_SCHEMA"),
+            volume=os.getenv("DATABRICKS_VOLUME"),
+            volume_path=os.getenv("DATABRICKS_VOLUME_PATH")
+        ),
+        # Other examples:
+        #
+        # For Databricks on AWS, with Databricks personal access token authentication:
+        # downloader_config=DatabricksAWSVolumesDownloaderConfig(download_dir=os.getenv("LOCAL_FILE_DOWNLOAD_DIR")),
+        # source_connection_config=DatabricksAWSVolumesConnectionConfig(
+        #     access_config=DatabricksAWSVolumesAccessConfig(token=os.getenv("DATABRICKS_TOKEN")),
+        #     host=os.getenv("DATABRICKS_HOST")
+        # ),
+        #
+        # For Azure Databricks, with Microsoft Entra ID service principal authentication:
+        # downloader_config=DatabricksAzureVolumesDownloaderConfig(download_dir=os.getenv("LOCAL_FILE_DOWNLOAD_DIR")),
+        # source_connection_config=DatabricksAzureVolumesConnectionConfig(
+        #     access_config=DatabricksAzureVolumesAccessConfig(
+        #         azure_client_id=os.getenv("ARM_CLIENT_ID"),
+        #         azure_client_secret=os.getenv("ARM_CLIENT_SECRET"),
+        #         azure_tenant_id=os.getenv("ARM_TENANT_ID")   
+        #     ),
+        #     host=os.getenv("DATABRICKS_HOST")
+        # ),
+        #
+        # For Databricks on Google Cloud, with Google Cloud Platform credentials authentication:
+        # downloader_config=DatabricksGoogleVolumesDownloaderConfig(download_dir=os.getenv("LOCAL_FILE_DOWNLOAD_DIR")),
+        # source_connection_config=DatabricksGoogleVolumesConnectionConfig(
+        #     access_config=DatabricksGoogleVolumesAccessConfig(
+        #         google_service_account=os.getenv("GOOGLE_CREDENTIALS")
+        #     ),
+        #     host=os.getenv("DATABRICKS_HOST")
+        # ),
+        partitioner_config=PartitionerConfig(
+            partition_by_api=True,
+            api_key=os.getenv("UNSTRUCTURED_API_KEY"),
+            partition_endpoint=os.getenv("UNSTRUCTURED_API_URL"),
+            additional_partition_args={
+                "split_pdf_page": True,
+                "split_pdf_allow_failed": True,
+                "split_pdf_concurrency_level": 15
+            }
+        ),
+        chunker_config=ChunkerConfig(chunking_strategy="by_title"),
+        embedder_config=EmbedderConfig(embedding_provider="huggingface"),
+        uploader_config=LocalUploaderConfig(output_dir=os.getenv("LOCAL_FILE_OUTPUT_DIR"))
+    ).run()
+```