# Databricks Import Integration

このデータコネクタを使用すると、Databricksから Treasure Data にデータをインポートできます。

## 前提条件

- Treasure Data の基本的な知識とTreasure Data アカウントへのアクセス権
- Databricks の基本的な知識とDatabricks アカウントへのアクセス権
- Databricks サーバーへのアクセス権


## 要件と制限事項

- Databricks サーバーが休止モードの場合、起動に数分かかることがあり、コネクタがタイムアウトする可能性があります。


## TD Console 経由での Databricks からのインポート

### Databricks コネクタのホスト名とHTTP パスの取得

1. Databricks の Web サイトにログインします。
2. **SQL** タブに移動し、**SQL Warehouses** を選択します。


![](/assets/screen_shot_2023-07-14_at_10_20_06.8cae14d9ca0b17c9b0ef9c81f79f8c4c39844c94ca7028bd287bd35311973334.f2aea2dd.png)
3. SQL Warehouse のインスタンスを選択します。

![](/assets/screen-shot-2023-07-14-at-10.13.49.246eeb2686f256c460dbca5eb43707bc4ec7ed6278582fc41e464221902dfd4e.f2aea2dd.png)
4. **Connection details** タブを選択します。
5. SQL Warehouse のServer hostname とHTTP path をメモします。

![](/assets/screen_shot_2023-06-30_at_08_50_56.25a28c85382e26612b6c1365cfae218cfcd6e808eb067d78d372bb8845d5610e.f2aea2dd.png)

### 認証の作成

最初のステップは、認証情報のセットを使用して新しい認証を作成することです。

1. **Integrations Hub > Catalog** を選択します。


![](/assets/image2021-9-30_14-38-2.ca92fa4ab9277dca95973c6bd413fc662a3f0d04b57d58f7a8c952a29f28bbec.f2aea2dd.png)

1. カタログで **databricks** を検索します。
2. アイコンの上にマウスを移動し、**Create Authentication** を選択します。


![](/assets/databricks.52d00c6e6d332e238921ad7df0439e7030961c141d547600e3cefbfafc5bbd81.f2aea2dd.png)

1. **Credentials** タブが選択されていることを確認し、統合の認証情報を入力します。


![](/assets/databricksnewauth.d784c65fa203e45bf52acb6befd56b42e1a950c01a7ccad059e21d877c9824b4.f2aea2dd.png)

| パラメータ | 説明 |
|  --- | --- |
| Hostname | Databricks インスタンスのホスト名です。 |
| HTTP Path | Databricks インスタンスのHTTP パスです。 |
| Authentication Method | 認証方法です。Basic または Personal Access Token を選択できます。 |
| Username | ログインに使用するユーザー名です。Authentication Method が Basic の場合にのみ有効です。 |
| Password | ログインに使用するパスワードです。Authentication Method が Basic の場合にのみ有効です。 |
| Token | Databricks パーソナルアクセストークンです。Authentication Method が Personal Access Token の場合にのみ有効です。 |


1. **Continue** を選択します。
2. 認証の名前を入力し、**Done** を選択します。


### ソースの作成

1. TD Console を開きます。
2. **Integrations Hub** > **Authentications** に移動します。
3. 新しい認証を見つけて、**New Source** を選択します。


![](/assets/databrickssource.7bef3fa03f1d1c30947fdabbf45767ca90b78d82974f4aafd3f6a77ae42e41d3.f2aea2dd.png)

#### 接続の作成

1. Data Transfer Name フィールドにソース名を入力します。
2. **Next** を選択します。


| パラメータ | 説明 |
|  --- | --- |
| Data Transfer Name | データ転送ソースの名前です。 |
| Authentication | 使用する認証の名前です。 |


Source Table タブが選択された状態で Create Source ページが表示されます。

#### ソーステーブルの識別

![](/assets/databricks-import-integration-2025-03-31.0baaea50153a553c622eb203d9b157ef317eff8c5b290c9536799b27c126ea39.f2aea2dd.png)

| パラメータ | 説明 |
|  --- | --- |
| Catalog | データセットカタログです。 |
| Schema | データセットスキーマです。 |
| Use custom SELECT query? | カスタムクエリを構築する場合はチェックボックスを選択し、クエリを作成する場合はフィールドに入力します。 |
| Skip validate query? | クエリの構文検証をスキップします |
| Select Columns | データセットから選択する列を定義します。screen-shotUse custom SELECT queryscreen-shot が選択されていない場合にのみ必要です。 |
| Table | 選択するテーブルです。screen-shotUse custom SELECT queryscreen-shot が選択されていない場合にのみ必要です。 |
| Filter Conditions | テーブルから取得するデータに追加の特定性が必要な場合は、WHERE 句の一部として指定できます。 |
| Order by | 特定のフィールドでレコードを並べ替える必要がある場合に指定します。増分列は order by 条件に含める必要があります。 |
| Query | SQL クエリを含むフィールドです。screen-shotUse custom SELECT queryscreen-shot が選択されている場合にのみ必要です。 |
| Incremental Loading | 増分読み込みを有効にする場合はチェックボックスを選択します |
| Incremental Columns | 増分列を定義します。これらの列は `order by` フィルタに含める必要があります。 |
| Start after value | 増分読み込みに使用される列の値です。 |
| Invalid Value Handling Mode | 無効な値の処理方法を指定します。次のいずれかを選択できます。  - Fail Job - Ignore Invalid Value - Ignore Row |


**Next** を選択します。

#### カスタムクエリの作成

- シンプルなクエリ:



```
query: SELECT * FROM simple_data WHERE cut = 'Very Good' ORDER BY `c_custkey`
```

- 増分とlast_records を使用したクエリ


`incremental\_columns` の値は、パラメータの位置に従い、列と同じ名前にする必要があります。


```
query: SELECT * FROM simple_data WHERE myid > :myid AND count > :count ORDER BY `myid`
incremental: true
incremental_columns: ['myid', 'count']
last_record: ['4', '5']
```

クエリには、`last\_record` に値が入力されたWHERE 条件が必要です。

Order by は必須ではありませんが、より正確な結果を得るために指定する必要があります。

screen-shotmyidscreen-shot が4より大きく、screen-shotcountscreen-shot が5より大きい実際のクエリは、次のようになります。


```
SELECT * FROM simple_data WHERE myid > 4 AND count > 5 ORDER BY `myid`
```

#### データ設定の定義

![](/assets/databricksdatasettings.d78ee51b26cf55c0ad0bf28e7f5fb90eaea92cde597ebc86a75d90c1891ca09d.f2aea2dd.png)

| パラメータ | 説明 |
|  --- | --- |
| COLUMN OPTIONS | 特定のカラムのデータ型を定義します。 |
| DEFAULT COLUMN_OPTIONS | 特定のデータ型の変換データ型を定義します。例: longを常にstringに変換する |


**Next** を選択します。

#### データのプレビュー

データプレビューはオプションであり、次のダイアログページに進む場合は **Next** をクリックしても問題ありません。

1. インポートを実行する前にデータのプレビューを表示するには、**Generate Preview** を選択します。


データプレビューに表示されるデータは、ソースから概算されたものです。これはインポートされる実際のデータではありません。

1. データが期待通りに表示されていることを確認します。


![](/assets/snippet-data-preview-2024-02-09.27dc5fd8772fca4f7f44ab28c00476ae1894744fe1e75d06932628929cc7bff1.4e139be3.png)
3. **Next** を選択します。

### Data Placement

データの配置について、データを配置したいターゲット database と table を選択し、インポートを実行する頻度を指定します。

1. **Next** を選択します。Storage の下で、インポートされたデータを配置する新しい database を作成するか、既存の database を選択し、新しい table を作成するか、既存の table を選択します。
2. **Database** を選択 > **Select an existing** または **Create New Database** を選択します。
3. オプションで、database 名を入力します。
4. **Table** を選択 > **Select an existing** または **Create New Table** を選択します。
5. オプションで、table 名を入力します。
6. データをインポートする方法を選択します。
  - **Append** (デフォルト) - データインポートの結果は table に追加されます。
table が存在しない場合は作成されます。
  - **Always Replace** - 既存の table の全体の内容をクエリの結果出力で置き換えます。table が存在しない場合は、新しい table が作成されます。
  - **Replace on New Data** - 新しいデータがある場合のみ、既存の table の全体の内容をクエリの結果出力で置き換えます。
7. **Timestamp-based Partition Key** 列を選択します。
デフォルトキーとは異なるパーティションキーシードを設定したい場合は、long または timestamp 列をパーティショニング時刻として指定できます。デフォルトの時刻列として、add_time フィルターで upload_time を使用します。
8. データストレージの **Timezone** を選択します。
9. **Schedule** の下で、このクエリを実行するタイミングと頻度を選択できます。


#### 一度だけ実行

1. **Off** を選択します。
2. **Scheduling Timezone** を選択します。
3. **Create & Run Now** を選択します。


#### 定期的に繰り返す

1. **On** を選択します。
2. **Schedule** を選択します。UI では、*@hourly*、*@daily*、*@monthly*、またはカスタム *cron* の 4 つのオプションが提供されます。
3. **Delay Transfer** を選択して、実行時間の遅延を追加することもできます。
4. **Scheduling Timezone** を選択します。
5. **Create & Run Now** を選択します。


転送が実行された後、**Data Workbench** > **Databases** で転送の結果を確認できます。

## Workflowを使用したDatabricksからのインポート

Workflowの screen-shottd_load>:screen-shot オペレーターを使用して、Databricksからデータをインポートできます。既にSOURCEを作成している場合は実行できます。SOURCEを作成したくない場合は、screen-shot.ymlscreen-shot ファイルを使用してインポートできます。

### Sourceの使用

1. TD Consoleを開きます。
2. Integrations Hub > Sourcesに移動します。
3. 対象のSourceを特定します。
4. Sourceの右端にあるmore icon (...)を選択し、Copy Unique IDを選択します。


![](/assets/image2021-10-12_12-26-58.09d9b84b0f1f752c7c95b0bc1c2d8e8b7302e5b91c6a3cb5f01309dadf53a604.f2aea2dd.png)

Unique IDは次のようになります: screen-shotdatabricks_import_1234••••••screen-shot

1. screen-shottd_load>:screen-shot オペレーターを使用してWorkflowタスクを定義します。



```yaml
+load:
  td_load>: databricks_import_1234••••••
  database: ${td.dest_db}
  table: ${td.dest_table}
```

1. Workflowを実行します。


### ymlファイルの使用

1. 対象の screen-shot.ymlscreen-shot ファイルを特定します。


screen-shot.ymlscreen-shot ファイルを作成する必要がある場合は、[Amazon S3 Import Integration Using CLI](/ja/int/amazon-s3-import-integration-v2#cli-configurations)を参照してください。

1. screen-shottd_load>:screen-shot オペレーターを使用してWorkflowタスクを定義します。



```yaml
+load:
  td_load>: config/daily_load.yml
  database: ${td.dest_db}
  table: ${td.dest_table}
```

1. Workflowを実行します。


#### パラメーターリファレンス

| 名前 | 説明 | 値 | デフォルト値 | 必須 |
|  --- | --- | --- | --- | --- |
| type | databricks | databricks |  | True |
| auth_method | 認証方法です。 | basic または access_token | basic |  |
| `host_name` | Databricksインスタンスのホスト名です。 |  |  | True |
| `http_path` | DatabricksインスタンスのHTTPパスです。 |  |  | True |
| user_name | ログイン用のユーザー名です。 |  |  | True（auth_methodがbasicの場合） |
| password | ログイン用のパスワードです。 |  |  | True（auth_methodがbasicの場合） |
| access_token | データ取得に使用するaccess_tokenです。 |  |  | True（auth_methodがaccess_tokenの場合） |
| catalog | データセットのカタログです。 |  |  | True |
| schema | データセットのスキーマです。 |  |  | True |
|  |  |  |  |  |
| use_custom_query | カスタムクエリを構築するか、フィールドに入力してカスタムクエリを作成できます。 | true/false | true |  |
| select | データセットから選択するカラムを定義します。これはscreen-shot*Use custom SELECT query*screen-shotがfalseの場合のみ必須です。 |  | * | True（use_custom_queryがfalseの場合） |
| table | 選択するテーブルです。これはscreen-shot*Use custom SELECT query*screen-shot がfalseの場合のみ必須です。 |  |  | True（use_custom_queryがfalseの場合） |
| where | テーブルから取得するデータの詳細な条件が必要な場合は、WHERE句の一部として指定できます。 |  |  |  |
| order_by | 特定のフィールドでレコードを並べ替える必要がある場合は指定します。増分カラムはorder by条件に含める必要があります。 |  |  |  |
| query | 独自のクエリを構築します。 |  |  | True（use_custom_queryがtrueの場合） |
| incremental | 増分読み込みを有効にします。 | true/false | false |  |
| `incremental_columns` | 特定のカラムのデータ型を定義します。 |  |  |  |
| last_record | 増分読み込みの最後のレコード値を定義します。 |  |  |  |
| `default_column_options` | 特定のデータ型の変換データ型を定義します。 |  |  |  |


### サンプルWorkflowコード

サンプルWorkflowコードについては、[Treasure Boxes](https://github.com/treasure-data/treasure-boxes/tree/master/td_load/s3)を参照してください。

## CLI (Toolbelt) を使用したDatabricksからのインポート

コネクタを設定する前に、最新の[TD Toolbelt](https://toolbelt.treasuredata.com/)をインストールしてください。

### Seed設定ファイル (seed.yml) の作成


```yaml
in:
  type: databricks
  auth_method: BASIC
  host_name: "dbc-8297d0a7-2709.cloud.databricks.com"
  http_path: "/sql/1.0/warehouses/1b195ce3a8720eb9"
  username: ***
  password: ***
  access_token: ***
  catalog: hive_metastore
  schema: default
  select: "*"
  table: diamonds
  incremental_columns: [ price, x, y, z ]
  order_by: "`price`, `x`, `y`, `z`"
  incremental: true
  last_record: ['4']
  default_column_options: {"double": {"type": "string"}}
out:
  mode: append
```

#### パラメーターリファレンス

| 名前 | 説明 | 値 | デフォルト値 | 必須 |
|  --- | --- | --- | --- | --- |
| type | databricks | databricks |  | True |
| auth_method | 認証方法です。 | basic/access_token | basic |  |
| `host_name` | Databricksインスタンスのホスト名です。 |  |  | True |
| `http_path` | DatabricksインスタンスのHTTPパスです。 |  |  | True |
| user_name | ログインするためのユーザー名です。 |  |  | True（auth_methodがbasicの場合） |
| password | ログインするためのパスワードです。 |  |  | True（auth_methodがbasicの場合） |
| access_token | データ取得のためのaccess_tokenです。 |  |  | True（auth_methodがaccess_tokenの場合） |
| catalog | データセットのカタログです。 |  |  | True |
| schema | これはデータセットのスキーマです。 |  |  | True |
| use_custom_query | カスタムクエリを作成するか、フィールドを入力してクエリを作成します。 | true/false | true |  |
| skip_validate_query | クエリの検証をスキップします。クエリの検証をスキップすると、プレビューモードは常にダミーデータで実行されます。 | true/false | false |  |
| select | データセットから選択する列を定義します。*カスタムSELECTクエリを使用*がfalseの場合にのみ必須です。 |  | * | use_custom_queryがfalseの場合はTrue |
| table | 選択するテーブルです。*カスタムSELECTクエリを使用*がfalseの場合にのみ必須です。 |  |  | use_custom_queryがfalseの場合はTrue |
| where | テーブルから取得するデータに追加の特定性が必要な場合は、WHERE句の一部として指定できます。 |  |  |  |
| order_by | 特定のフィールドでレコードを並べ替える必要がある場合に指定します。incrementalカラムはorder by条件に含める必要があります。 |  |  |  |
| query | 独自のクエリを作成します。 |  |  | use_custom_queryがtrueの場合はTrue |
| incremental | インクリメンタルローディングを有効にします。 | true/false | false |  |
| `incremental_columns` | 特定の列のデータ型を定義します。 |  |  |  |
| last_record | インクリメンタルローディングの最終レコード値を定義します。 |  |  |  |
| `default_column_options` | 特定のデータ型の変換データ型を定義します。 |  |  |  |


### load.ymlの生成

*connector:guess*コマンドは、ソースファイルを自動的に読み取り、ロジックを使用してファイル形式、フィールド、列を推測します。


```
$ td connector:guess seed.yml -o load.yml
```

*load.yml*を開いて、ファイル形式、エンコーディング、列名、型などの定義を確認できます。

- 例



```yaml
in:
  type: databricks
  auth_method: BASIC
  host_name: "dbc-8297d0a7-2709.cloud.databricks.com"
  http_path: "/sql/1.0/warehouses/1b195ce3a8720eb9"
  username: ***
  password: ***
  access_token: ***
  catalog: hive_metastore
  schema: default
  select: "*"
  table: diamonds
  incremental_columns: [ price, x, y, z ]
  order_by: "`price`, `x`, `y`, `z`"
  query: SELECT * FROM simple_data WHERE myid > :myid
  incremental: true
  last_record: ['4']
  default_column_options: {"double": {"type": "string"}}
out:
  mode: append
```

データをプレビューするには、*td connector:preview*コマンドを使用します。


```
$ td connector:preview load.yml
+-------+---------+----------+---------------------+
| id    | company | customer | created_at          |
+-------+---------+----------+---------------------+
| 11200 | AA Inc. |    David | 2015-03-31 06:12:37 |
| 20313 | BB Imc. |      Tom | 2015-04-01 01:00:07 |
| 32132 | CC Inc. | Fernando | 2015-04-01 10:33:41 |
| 40133 | DD Inc. |    Cesar | 2015-04-02 05:12:32 |
| 93133 | EE Inc. |     Jake | 2015-04-02 14:11:13 |
+-------+---------+----------+---------------------+
```

guessコマンドは、ソースデータファイル内に3行以上、2列以上が必要です。これは、コマンドがソースデータのサンプル行を使用して列定義を評価するためです。

列名または列型が予期せず検出された場合は、*load.yml*ファイルを変更して再度プレビューしてください。

### ロードジョブの実行

ジョブの実行には、データのサイズに応じて数時間かかる場合があります。データを保存するTreasure Dataのデータベースとテーブルを必ず指定してください。

Treasure Dataのストレージは時間によってパーティション分割されているため（[データパーティショニング](https://docs.treasuredata.com/smart/project-product-documentation/data-partitioning-in-treasure-data)を参照）、*--time-column*オプションを指定することをお勧めします。このオプションが指定されていない場合、データコネクタは最初の*long*または*timestamp*列をパーティショニング時間として選択します。*--time-column*で指定する列の型は、*long*または*timestamp*のいずれかである必要があります。

データに時間列がない場合は、*add_time*フィルタオプションを使用して時間列を追加できます。詳細については、[add_timeフィルタプラグイン](https://docs.treasuredata.com/smart/project-product-documentation/add_time-filter-function)を参照してください。


```
$ td connector:issue load.yml --database td_sample_db --table td_sample_table \
  --time-column created_at
```

この*connector:issue*のコマンド例では、すでに*database(td_sample_db)*と*table(td_sample_table)*を作成していることを前提としています。TDにデータベースまたはテーブルが存在しない場合、このコマンドは失敗します。データベースとテーブルを手動で作成するか、*td connector:issue*コマンドで*--auto-create-table*オプションを使用してデータベースとテーブルを自動作成してください。


```
$ td connector:issue load.yml --database td_sample_db --table td_sample_table
 --time-column created_at --auto-create-table
```

データコネクタは、サーバー側でレコードをソートしません。時間ベースのパーティショニングを効果的に使用するには、事前にファイル内のレコードをソートしてください。

*time*というフィールドがある場合は、*--time-column*オプションを指定する必要はありません。


```
$ td connector:issue load.yml --database td_sample_db --table td_sample_table
```

#### インポートモード

*load.yml*ファイルのoutセクションでファイルのインポートモードを指定できます。*out:*セクションは、データがTreasure Dataテーブルにインポートされる方法を制御します。たとえば、データを追加したり、Treasure Dataの既存のテーブル内のデータを置き換えたりすることができます。

| **モード** | **説明** | **例** |
|  --- | --- | --- |
| Append | レコードはターゲットテーブルに追加されます。 | in:    ...  out:    mode: append |
| Always Replace | ターゲットテーブルのデータを置き換えます。ターゲットテーブルに対して行われた手動のスキーマ変更はそのまま保持されます。 | in:    ...  out:    mode: replace |
| Replace on new data | インポートする新しいデータがある場合にのみ、ターゲットテーブルのデータを置き換えます。 | in:    ...  out:    mode: replace_on_new_data |


### 実行のスケジューリング

インクリメンタルファイルインポートのために、定期的なデータコネクタの実行をスケジュールできます。高可用性を確保するために、スケジューラを慎重に設定してください。

スケジュールされたインポートの場合、指定されたプレフィックスに一致し、条件に応じて次のいずれかのフィールドを持つすべてのファイルをインポートできます。

- *use_modified_time*が無効になっている場合、次の実行のために最後のパスが保存されます。2回目以降の実行では、コネクタはアルファベット順で最後のパスの後に来るファイルのみをインポートします。
- それ以外の場合、ジョブが実行された時刻が次の実行のために保存されます。2回目以降の実行では、コネクタはアルファベット順でその実行時刻以降に変更されたファイルのみをインポートします。


タイムゾーンを指定せずに指定されたすべての日付および日付/時刻形式は、常にデフォルトのUTCとして扱われます。

#### TD Toolbeltを使用したスケジュールの作成

新しいスケジュールは、*td connector:create*コマンドを使用して作成できます。


```
$ td connector:create daily_import "10 0 * * *" \
 td_sample_db td_sample_table load.yml
```

Treasure Dataのストレージは時間によってパーティション分割されているため（[データパーティショニング](https://docs.treasuredata.com/smart/project-product-documentation/data-partitioning-in-treasure-data)も参照）、*--time-column*オプションを指定することをお勧めします。


```
$ td connector:create daily_import "10 0 * * *" \
 td_sample_db td_sample_table load.yml \
 --time-column created_at
```

*cron*パラメータは、`@hourly`、`@daily`、`@monthly`の3つの特別なオプションも受け付けます。

デフォルトでは、スケジュールはUTCタイムゾーンで設定されます。*-t*または*--timezone*オプションを使用して、タイムゾーンでスケジュールを設定できます。*--timezone*オプションは、'Asia/Tokyo'、'America/Los_Angeles'などの拡張タイムゾーン形式のみをサポートしています。PSTやCSTなどのタイムゾーンの略語は*サポートされておらず*、予期しないスケジュールになる可能性があります。