Import from Databricks¶

Elevate Data empowers organizations to effectively manage their data lifecycle by enabling seamless integration of diverse data sources on a large scale, while also providing tools for enhanced governance and efficiency.

Elevate Data manages connections to various data sources (S3, Snowflake, and Databricks at this stage), allowing for efficient data extraction.

It also provides a central repository of data elements, imported from different sources, for greater transparency and efficient collaboration.

This notebook outlines how to:

1. Create a Connection: Connections allow for a secure setup to external data sources, such as a Snowflake host, a Databricks Delta Sharing, or an Amazon S3 bucket.
2. Retrieve and list connections to manage your data sources.
3. Create a Data Source: A Data Source defines the specific elements within the connection from which data is extracted.
4. Create a Data Table: A Data Table references a Data Source and triggers the import of its first version. With blocking=True, the call waits for that import to finish before returning.
5. Retrieve and list data tables to manage your data assets.
6. Preview 100 rows of the version.
7. Create additional data versions to refresh your data.

In this example, we are going to create a Connection to a Databricks Delta Sharing and import data from it using a Data Table and create a new version of the data table.

Install the SDK¶

In [ ]:

skip-execution

%pip install --upgrade earnix-elevate

%pip install --upgrade earnix-elevate

Set your Elevate server and credentials¶

In this example we use environment variables to authenticate to Elevate, but you can also inject your credentials using Python arguments to each Service's client, like in the commented example.

In [ ]:

import os

os.environ.setdefault("E2_SERVER", "YOUR_SERVER")
os.environ.setdefault("E2_CLIENT_ID", "YOUR_CLIENT_ID")
os.environ.setdefault("E2_SECRET_KEY", "YOUR_SECRET_KEY")

# Alternatively:
# ConnectionService(
#     server="YOUR_SERVER",
#     client_id="YOUR_CLIENT_ID",
#     secret_key="YOUR_SECRET_KEY"
# )

import os os.environ.setdefault("E2_SERVER", "YOUR_SERVER") os.environ.setdefault("E2_CLIENT_ID", "YOUR_CLIENT_ID") os.environ.setdefault("E2_SECRET_KEY", "YOUR_SECRET_KEY") # Alternatively: # ConnectionService( # server="YOUR_SERVER", # client_id="YOUR_CLIENT_ID", # secret_key="YOUR_SECRET_KEY" # )

Imports and demo preparation¶

In [ ]:

import json
from datetime import datetime

from earnix_elevate import (
    ConnectionService,
    CreateDataTableRequest,
    CreateDeltaSharingConnectionRequest,
    CreateDeltaSharingDataSourceRequest,
    DataSourceService,
    DataTableService,
    DeltaSharingJsonProfileAuthRequest,
)

DEMO_SUFFIX = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

connection_service = ConnectionService()
data_source_service = DataSourceService()
data_table_service = DataTableService()

import json from datetime import datetime from earnix_elevate import ( ConnectionService, CreateDataTableRequest, CreateDeltaSharingConnectionRequest, CreateDeltaSharingDataSourceRequest, DataSourceService, DataTableService, DeltaSharingJsonProfileAuthRequest, ) DEMO_SUFFIX = datetime.now().strftime("%Y-%m-%d %H:%M:%S") connection_service = ConnectionService() data_source_service = DataSourceService() data_table_service = DataTableService()

Create a Connection¶

There are several steps to follow in order to create a Delta Sharing connection.

In [ ]:

import os

# Using OAUTH2
OAUTH2ProfileContent = json.dumps(
    {
        "shareCredentialsVersion": 2,
        "endpoint": os.environ.get(
            "DATABRICKS_OAUTH2_ENDPOINT",
            "https://your-region.cloud.databricks.com/api/2.0/delta-sharing/metastores/00000000-0000-0000-0000-000000000000/recipients/00000000-0000-0000-0000-000000000000",
        ),
        "tokenEndpoint": os.environ.get(
            "DATABRICKS_OAUTH2_TOKEN_ENDPOINT",
            "https://login.microsoftonline.com/00000000-0000-0000-0000-000000000000/oauth2/v2.0/token",
        ),
        "type": "oauth_client_credentials",
        "clientId": os.environ.get(
            "DATABRICKS_OAUTH2_CLIENT_ID", "00000000-0000-0000-0000-000000000000"
        ),
        "clientSecret": os.environ.get(
            "DATABRICKS_OAUTH2_CLIENT_SECRET",
            "Example~ClientSecretValueGoesHere0000000",
        ),
        "scope": os.environ.get(
            "DATABRICKS_OAUTH2_SCOPE",
            "api://00000000-0000-0000-0000-000000000000/databricks/.default",
        ),
    }
)

PATProfileContent = json.dumps(
    {
        "shareCredentialsVersion": 1,
        "bearerToken": os.environ.get("DATABRICKS_PAT_TOKEN", "XXX"),
        "endpoint": os.environ.get(
            "DATABRICKS_PAT_ENDPOINT",
            "https://your-region.cloud.databricks.com/api/2.0/delta-sharing/metastores/XXX",
        ),
        "expirationTime": "2027-01-10T14:06:53.036Z",
    }
)

# use OAUTH2ProfileContent for OAUTH2 or PATProfileContent for PAT:
auth_details = DeltaSharingJsonProfileAuthRequest(
    authType="deltasharing_json_profile",
    deltaSharingProfileName="config.share",
    deltaSharingProfileContent=PATProfileContent,
)

new_conn = CreateDeltaSharingConnectionRequest(
    type="delta_sharing",
    name="Databricks Demo Conn " + DEMO_SUFFIX,
    description="Delta Sharing connection example",
    auth=auth_details,
)

create_conn = connection_service.create_connection(new_conn)
print(f"\n{create_conn=}")

import os # Using OAUTH2 OAUTH2ProfileContent = json.dumps( { "shareCredentialsVersion": 2, "endpoint": os.environ.get( "DATABRICKS_OAUTH2_ENDPOINT", "https://your-region.cloud.databricks.com/api/2.0/delta-sharing/metastores/00000000-0000-0000-0000-000000000000/recipients/00000000-0000-0000-0000-000000000000", ), "tokenEndpoint": os.environ.get( "DATABRICKS_OAUTH2_TOKEN_ENDPOINT", "https://login.microsoftonline.com/00000000-0000-0000-0000-000000000000/oauth2/v2.0/token", ), "type": "oauth_client_credentials", "clientId": os.environ.get( "DATABRICKS_OAUTH2_CLIENT_ID", "00000000-0000-0000-0000-000000000000" ), "clientSecret": os.environ.get( "DATABRICKS_OAUTH2_CLIENT_SECRET", "Example~ClientSecretValueGoesHere0000000", ), "scope": os.environ.get( "DATABRICKS_OAUTH2_SCOPE", "api://00000000-0000-0000-0000-000000000000/databricks/.default", ), } ) PATProfileContent = json.dumps( { "shareCredentialsVersion": 1, "bearerToken": os.environ.get("DATABRICKS_PAT_TOKEN", "XXX"), "endpoint": os.environ.get( "DATABRICKS_PAT_ENDPOINT", "https://your-region.cloud.databricks.com/api/2.0/delta-sharing/metastores/XXX", ), "expirationTime": "2027-01-10T14:06:53.036Z", } ) # use OAUTH2ProfileContent for OAUTH2 or PATProfileContent for PAT: auth_details = DeltaSharingJsonProfileAuthRequest( authType="deltasharing_json_profile", deltaSharingProfileName="config.share", deltaSharingProfileContent=PATProfileContent, ) new_conn = CreateDeltaSharingConnectionRequest( type="delta_sharing", name="Databricks Demo Conn " + DEMO_SUFFIX, description="Delta Sharing connection example", auth=auth_details, ) create_conn = connection_service.create_connection(new_conn) print(f"\n{create_conn=}")

Retrieve Connection Details¶

After creating a connection, you can retrieve its details using the connection ID.

In [ ]:

retrieved_conn = connection_service.get_connection(create_conn.id)
print(f"\n{retrieved_conn=}")

retrieved_conn = connection_service.get_connection(create_conn.id) print(f"\n{retrieved_conn=}")

List All Connections¶

You can also list all available connections in your environment.

In [ ]:

all_connections = connection_service.list_connections().items
print(f"\nFound {len(all_connections)} connections:")
for conn in all_connections:
    print(f"  - {conn.name} (ID: {conn.id})")

all_connections = connection_service.list_connections().items print(f"\nFound {len(all_connections)} connections:") for conn in all_connections: print(f" - {conn.name} (ID: {conn.id})")

Create a Data Table¶

There are several steps to follow in order to create a Data Table that references specific Delta Sharing elements through the established connection. Once the Data Table is created, the import of the first version will begin automatically.

Create the Data Source¶

When creating a Data Source for the Data Table, provide the ID of the connection you wish to use, and define the table name, schema, and share to retrieve data from the source.

In [ ]:

new_ds = CreateDeltaSharingDataSourceRequest(
    type="delta_sharing",
    name="Databricks Demo DS " + DEMO_SUFFIX,
    connectionId=create_conn.id,
    share="elevate_share",
    schema="default",
    table="life_insurance_100",
)

create_ds = data_source_service.create_data_source(new_ds)
print(f"\n{create_ds=}")

new_ds = CreateDeltaSharingDataSourceRequest( type="delta_sharing", name="Databricks Demo DS " + DEMO_SUFFIX, connectionId=create_conn.id, share="elevate_share", schema="default", table="life_insurance_100", ) create_ds = data_source_service.create_data_source(new_ds) print(f"\n{create_ds=}")

Create the Data Table¶

In [ ]:

new_dt = CreateDataTableRequest(
    name="Bee-Insurance Demo DT " + DEMO_SUFFIX,
    dataSourceId=create_ds.id,
)

# blocking=True creates the table and waits for its import to finish,
# returning the imported table (raising on failure/timeout).
create_dt = data_table_service.create_data_table(new_dt, blocking=True)
print(f"\n{create_dt=}")
data_version_number = create_dt.latest_data_version.version_number

new_dt = CreateDataTableRequest( name="Bee-Insurance Demo DT " + DEMO_SUFFIX, dataSourceId=create_ds.id, ) # blocking=True creates the table and waits for its import to finish, # returning the imported table (raising on failure/timeout). create_dt = data_table_service.create_data_table(new_dt, blocking=True) print(f"\n{create_dt=}") data_version_number = create_dt.latest_data_version.version_number

Retrieve Data Table Details¶

After creating a data table, you can retrieve its details using the data table ID.

In [ ]:

retrieved_dt = data_table_service.get_data_table(create_dt.id)
print(f"\n{retrieved_dt=}")

retrieved_dt = data_table_service.get_data_table(create_dt.id) print(f"\n{retrieved_dt=}")

List All Data Tables¶

You can also list all available data tables in your environment.

In [ ]:

all_data_tables = data_table_service.list_data_tables().items
print(f"\nFound {len(all_data_tables)} data tables:")
for dt in all_data_tables:
    print(f"  - {dt.name} (ID: {dt.id})")

all_data_tables = data_table_service.list_data_tables().items print(f"\nFound {len(all_data_tables)} data tables:") for dt in all_data_tables: print(f" - {dt.name} (ID: {dt.id})")

Preview¶

Once the version has been successfully imported, you can preview the first 100 rows of the imported data.

In [ ]:

dvt = data_table_service.get_data_version_preview(create_dt.id, data_version_number)
print(f"\n{dvt=}")

dvt = data_table_service.get_data_version_preview(create_dt.id, data_version_number) print(f"\n{dvt=}")

Create Additional Data Versions¶

You can create new versions of your data table to retrieve updated data from the source. This is useful when the underlying data in Databricks has changed and you want to refresh your data table.

In [ ]:

new_version = data_table_service.create_data_version(create_dt.id)
print(f"\n{new_version=}")

new_version = data_table_service.create_data_version(create_dt.id) print(f"\n{new_version=}")

Cleanup¶

Delete all resources created in this notebook.

In [ ]:

# Cleanup — delete all resources created in this notebook
try:
    data_table_service.delete_data_table(data_table_id=create_dt.id)
    print(f"Deleted data table {create_dt.id}")
except Exception as e:
    print(f"Failed to delete data table: {e}")

try:
    data_source_service.delete_data_source(data_source_id=create_ds.id)
    print(f"Deleted data source {create_ds.id}")
except Exception as e:
    print(f"Failed to delete data source: {e}")

try:
    connection_service.delete_connection(connection_id=create_conn.id)
    print(f"Deleted connection {create_conn.id}")
except Exception as e:
    print(f"Failed to delete connection: {e}")

# Cleanup — delete all resources created in this notebook try: data_table_service.delete_data_table(data_table_id=create_dt.id) print(f"Deleted data table {create_dt.id}") except Exception as e: print(f"Failed to delete data table: {e}") try: data_source_service.delete_data_source(data_source_id=create_ds.id) print(f"Deleted data source {create_ds.id}") except Exception as e: print(f"Failed to delete data source: {e}") try: connection_service.delete_connection(connection_id=create_conn.id) print(f"Deleted connection {create_conn.id}") except Exception as e: print(f"Failed to delete connection: {e}")