Databricks の High Concurrency クラスターから Unity Catalog のテーブルをDatabricks SQL 経由で参照する方法

## 概要

Databricks の High Concurrency クラスターから Unity Catalog のテーブルを参照することが実施できないのですが、Databricks SQL 経由であれば Unity Catalog のテーブルを参照できたため、実行したコードとその結果を共有します。

High Concurrency クラスターから Unity Catalog のテーブルを参照すると次のようなエラーが発生します。

```text
Unity Catalog is not enabled on this cluster.
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d9288a15-5936-5292-95fc-7ddda14ab158.png)

そこで、JDBC 経由で Databricks SQL に接続して、Unity Catalog のテーブルを参照するようにしました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ffeea8dc-41e5-b34e-7203-3ba697908ece.png)


## コード

### Unity Catalog のテーブルの準備

この手順は、Databricks SQL をアタッチして実行する想定です。

```python
%sql
CREATE CATALOG IF NOT EXISTS qiita;
CREATE SCHEMA IF NOT EXISTS qiita.schema_01;
```

```python
%sql
CREATE OR REPLACE TABLE qiita.schema_01.nation_delta
(
  N_NATIONKEY integer COMMENT 'This is a N_NATIONKEY'
  ,N_NAME string COMMENT 'This is a N_NAME'
  ,N_REGIONKEY integer COMMENT 'This is a N_REGIONKEY'
  ,N_COMMENT string COMMENT 'This is a N_COMMENT'
  ,update_date TIMESTAMP COMMENT 'This is a update_date'
  ,getnerated_col string GENERATED ALWAYS AS ('default_value') COMMENT 'This is a getnerated_col'
  ,identity_col BIGINT GENERATED ALWAYS AS IDENTITY COMMENT 'This is a identity_col'
)
PARTITIONED BY (
  N_REGIONKEY,
  update_date
)
COMMENT 'This is a nation_delta'
TBLPROPERTIES (
  delta.dataSkippingNumIndexedCols = 1
  ,delta.logRetentionDuration = 'interval 30 days'
  ,delta.deletedFileRetentionDuration= 'interval 7 days'
  ,delta.enableChangeDataFeed = true
  ,delta.autoOptimize.optimizeWrite = true
  ,delta.autoOptimize.autoCompact = true
  ,delta.tuneFileSizesForRewrites = true
)
;
 
-- Insert into table
INSERT INTO qiita.schema_01.nation_delta
(
  N_NATIONKEY
  ,N_NAME
  ,N_REGIONKEY
  ,N_COMMENT
  ,update_date
)
SELECT
  N_NATIONKEY
  ,N_NAME
  ,N_REGIONKEY 
  ,N_COMMENT
  ,CAST(current_timestamp AS date) AS update_date
  FROM
    read_files(
      'dbfs:/databricks-datasets/tpch/data-001/nation/nation.tbl',
      format => 'csv',
      sep => '|',
      schema => '
        N_NATIONKEY  integer
        ,N_NAME       string
        ,N_REGIONKEY  integer 
        ,N_COMMENT    string
      ')
;
```

```python
databricks-logocreate_table(Python)
 Import Notebook
%sql
CREATE CATALOG IF NOT EXISTS qiita;
CREATE SCHEMA IF NOT EXISTS qiita.schema_01;
OK
%sql
CREATE OR REPLACE TABLE qiita.schema_01.nation_delta
(
  N_NATIONKEY integer COMMENT 'This is a N_NATIONKEY'
  ,N_NAME string COMMENT 'This is a N_NAME'
  ,N_REGIONKEY integer COMMENT 'This is a N_REGIONKEY'
  ,N_COMMENT string COMMENT 'This is a N_COMMENT'
  ,update_date TIMESTAMP COMMENT 'This is a update_date'
  ,getnerated_col string GENERATED ALWAYS AS ('default_value') COMMENT 'This is a getnerated_col'
  ,identity_col BIGINT GENERATED ALWAYS AS IDENTITY COMMENT 'This is a identity_col'
)
PARTITIONED BY (
  N_REGIONKEY,
  update_date
)
COMMENT 'This is a nation_delta'
TBLPROPERTIES (
  delta.dataSkippingNumIndexedCols = 1
  ,delta.logRetentionDuration = 'interval 30 days'
  ,delta.deletedFileRetentionDuration= 'interval 7 days'
  ,delta.enableChangeDataFeed = true
  ,delta.autoOptimize.optimizeWrite = true
  ,delta.autoOptimize.autoCompact = true
  ,delta.tuneFileSizesForRewrites = true
)
;
 
-- Insert into table
INSERT INTO qiita.schema_01.nation_delta
(
  N_NATIONKEY
  ,N_NAME
  ,N_REGIONKEY
  ,N_COMMENT
  ,update_date
)
SELECT
  N_NATIONKEY
  ,N_NAME
  ,N_REGIONKEY 
  ,N_COMMENT
  ,CAST(current_timestamp AS date) AS update_date
  FROM
    read_files(
      'dbfs:/databricks-datasets/tpch/data-001/nation/nation.tbl',
      format => 'csv',
      sep => '|',
      schema => '
        N_NATIONKEY  integer
        ,N_NAME       string
        ,N_REGIONKEY  integer 
        ,N_COMMENT    string
      ')
;
Table
 
num_affected_rows
num_inserted_rows
1
25
25
1 row
%sql
SELECT 
  *
  FROM
    qiita.schema_01.nation_delta;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/08698dff-bbba-cea8-e3cd-f6c855f6a704.png)

### Databricks トークンを取得

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f987a460-c840-ed7b-12f3-b3d13c20b83e.png)

### トークンを Databricks シークレットに登録

```python
%pip install databricks-sdk --upgrade -q
```

```python
dbutils.library.restartPython()
```

```python
from databricks.sdk import WorkspaceClient
from databricks.sdk.core import DatabricksError

w = WorkspaceClient()
```

```python
# scope 名をセット
scope_name = 'qiita'
scope_key_and_values = {
    "db_token": "dapiaff1ebb04975f4efd08eaaaaaa", # Databricks トークンを設定
}
```

```python
# scope を作成
try:
    w.secrets.create_scope(scope=scope_name)
except DatabricksError:
    print(f"Scope `{scope_name}` already exists!")
 
# シークレットを登録
for key,value in scope_key_and_values.items():
    w.secrets.put_secret(
        scope=scope_name,
        key=key,
        string_value=value,
    )
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c5116847-c57f-20d6-d98e-458c6863ab2b.png)

### JDBC 経由で Databricks SQL に接続してデータフレームを作成

```python
host = "adb-554481832333334.4.azuredatabricks.net" # Sever hostname を設定
httpPath = "/sql/1.0/warehouses/95cac05d1333333" # HTTP path を設定
```

```python
df = (
    spark.read.format("databricks")
    .option("host", host)
    .option("httpPath", httpPath)
    .option("personalAccessToken", dbutils.secrets.get(scope_name,"db_token"))
    .option("dbtable", "qiita.schema_01.nation_delta")
    .load()
)
```

```python
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/316e36f7-976d-ea4b-1a77-b10784f871b2.png)