# SFTP Parquet インポート Integration

`SFTP Parquet` 用データコネクターを使用すると、SFTP サーバーに保存されている Parquet ファイルを Treasure AI にインポートできます。

## 前提条件

- Treasure AI の基礎知識。
- この Integration を使用する前に、お使いの環境で有効なプロトコルを決定してください。
- ファイアウォールを使用している場合は、許可された IP 範囲とポートを確認してください。サーバー管理者は、セキュリティ上の理由から、デフォルトポート番号を TCP 22 から変更することがあります。
- "PuTTY" およびその他の形式はサポートされていません。


## 制限事項とサポート

- **Parquet ファイルのみ** - この Integration は Parquet ファイル専用です。その他のファイル形式（CSV、JSON、TSV など）はサポートされていません。
- ファイルは有効な Parquet 形式である必要があります。
- Out of Memory を回避するため、parquet ファイルの行グループサイズは約 128 MB にしてください


## Treasure Data Integration の静的 IP アドレス

セキュリティポリシーで IP ホワイトリストが必要な場合は、接続を成功させるために Treasure Data の IP アドレスを許可リストに追加する必要があります。

リージョンごとに整理された静的 IP アドレスの完全なリストは、次のリンクにあります:
[IP Addresses for Integrations](/apis/endpoints/ip-addresses-integrations-result-workers)

## Treasure Console を使用して接続を作成する

### 新しい接続を作成する

Treasure AI では、クエリを実行する前にデータ接続を作成および設定する必要があります。データ接続の一部として、Integration にアクセスするための認証を提供します。

1. **Treasure Console** を開きます。
2. **Integration Hub** > **Catalog** に移動します。
3. SFTP Parquet を検索して選択します。


![新しい認証](/assets/sftp_parquet_create_authentication.c0f74bbdbd0859b4db4b777d64451d640d0237cc62be5dd85d8c55ac1a40858f.a126b8b7.webp)
4. **認証を作成** を選択します。

![認証詳細](/assets/sftp_parquet_create_authentication_detail.edece0f1d93926a18f8a85860c58b5ebbea9f3a236abb8ed6031206451c3edf3.a126b8b7.webp)
5. 次のダイアログが開きます。パラメータを編集します。**続行** を選択します。

| パラメータ | 説明 |
|  --- | --- |
| **Host** | リモート SFTP インスタンスのホスト情報（例：IP アドレス）。 |
| **Port** | リモート SFTP インスタンスの接続ポート、デフォルトは 22。 |
| **User** | リモート SFTP インスタンスへの接続に使用するユーザー名。 |
| **認証モード** | SFTP サーバーで認証を行う方法を選択します。 |
| **秘密鍵ファイル** | `認証モード` から「公開鍵/秘密鍵ペア」が選択されている場合に必要です。（RSA、DSS、ECDSA、および ED25519 がサポートされています。） |
| **秘密鍵ファイルのパスフレーズ** | （オプション）必要に応じて、提供された秘密ファイルのパスフレーズを入力します。 |
| **再試行制限** | 失敗した接続を再試行する回数（デフォルト 10）。 |
| **タイムアウト** | 接続タイムアウト（秒）（デフォルト 600）。 |
| **最大 SFTP 接続数** | 同時 SFTP 接続の最大数（整数、デフォルト 1、最小 1、最大 10）。並列ファイル転送の接続プールサイズを制御します。 |


1. 接続の名前を入力します。
2. 認証を他のユーザーと共有するかどうかを選択します。
3. **続行** を選択します。


## データを Treasure AI に転送する

認証済み接続を作成すると、自動的に認証ページに移動します。

1. 作成した接続を検索します。
2. **新しいソース** を選択します。
3. データ転送名フィールドに **ソース** の名前を入力します。
4. **次へ** を選択します。
5. 以下のパラメータを編集します。


| パラメータ | 説明 |
|  --- | --- |
| ユーザーディレクトリルート | パスプレフィックスがユーザーディレクトリ下にあるかどうかを確認します。例：/home/test_user |
| パスプレフィックス
 | 対象ファイルのプレフィックス。フォルダを指す必要があります（文字列、必須）。パスプレフィックスはフォルダパスである必要があります。ファイルパスに部分的なファイル名が含まれている場合、`無効な path_prefix:xxx` エラーメッセージが表示されます。
path_prefix は Parquet ファイルを含むフォルダパスである必要があります。
 |
| パスマッチパターン | ファイルパスをクエリする正規表現を入力します。ファイルパスが指定されたパターンと一致しない場合、そのファイルはスキップされます。例：パターン `.parquet$` を指定した場合、パスがパターンと一致しないファイルはスキップされます。 |
| インクリメンタル | インクリメンタル読み込みを有効にします（ブール値、オプション。デフォルト：true）。インクリメンタル読み込みが有効になっている場合、次回実行の設定差分に last_path パラメータが含まれ、次回実行時にそのパス以前のファイルがスキップされます。そうでない場合、last_path は含まれません。 |
| 開始パス以降 | これより辞書式順序で大きいパスのみがインポートされます。 |


1. **次へ** を選択します。


データ設定ページは必要に応じて変更できますが、ページをスキップすることも可能です。

![新しいデータソース](/assets/sftp_parquet_create_source.e339833b61b0a85a88c3b49aa72f05f511b5245b140695ad9effa97302187a0b.a126b8b7.webp)

![データソース詳細](/assets/sftp_parquet_create_source_detail.c3a0d77cc09da3090b4ee900298e1796ec79b5039afca21bfb2ac8770e34bd88.a126b8b7.webp)

### Data Preview

インポートを実行する前に、Generate Preview を選択してデータの[プレビュー](/products/customer-data-platform/integration-hub/batch/import/previewing-your-source-data)を表示できます。Data preview はオプションであり、選択した場合はダイアログの次のページに安全にスキップできます。

1. **Next** を選択します。Data Preview ページが開きます。
2. データをプレビューする場合は、**Generate Preview** を選択します。
3. データを確認します。


### 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** で転送の結果を確認できます。

# Treasure Workflow を使用した SFTP Parquet インポート

ワークフローを作成して実行します

```yaml
_export:
  td:
    database: workflow_sftp_parquet
    table: workflow_sftp_parquet
+import_from_sftp_parquet:
  td_load>: imports/seed.yml
  database: ${td.database}
  table: ${td.table}
```

インポート用に *seed.yml* ファイルを SFTP 接続詳細で変更します。

```yaml
in:
  type: sftp_parquet
  host: HOST
  port: <PORT, default is 22>
  auth_method: key_pair
  user: USER
  secret_key_file:
    content: |
      -----BEGIN RSA PRIVATE KEY-----
      Proc-Type: 4,ENCRYPTED
      DEK-Info: AES-128-CBC...
      ...
      -----END RSA PRIVATE KEY-----
  secret_key_passphrase: PASSPHRASE
  user_directory_is_root: true
  timeout: 600
  path_prefix: /path/to/parquet/files
out:
  mode: append
```

| 設定パラメータ | 値 |
|  --- | --- |
| host: | （文字列、必須） |
| port: | （文字列、デフォルト：22） |
| auth_method: | （文字列 ['password', 'key_pair']、必須） |
| user: | （文字列、必須） |
| password: | （文字列、デフォルト：null） |
| secret_key_file: | （文字列、デフォルト：null）。OpenSSH 形式が必要です。 |
| secret_key_passphrase: | （文字列、デフォルト：""） |
| user_directory_is_root: | （ブール値、デフォルト：true） |
| timeout: SFTP 接続タイムアウト（秒） | （整数、デフォルト：600） |
| maximum_sftp_connections: 最大同時 SFTP 接続数 | （整数、デフォルト：1、最小：1、最大：10） |
| path_prefix: 出力パスのプレフィックス | （文字列、必須） |
| incremental: インクリメンタル読み込みを有効にします | （ブール値、オプション。デフォルト：true）。インクリメンタル読み込みが有効になっている場合、次回実行の設定差分に last_path パラメータが含まれ、次回実行時にそのパス以前のファイルがスキップされます。そうでない場合、last_path は含まれません。 |
| path_match_pattern: | ファイルパスにマッチする正規表現。ファイルパスがこのパターンと一致しない場合、そのファイルはスキップされます（正規表現文字列、オプション） |
| total_file_count_limit: | 読み込むファイルの最大数（整数、オプション） |
| min_task_size（実験的）: | タスクの最小サイズ。これが 0 より大きい場合、1つのタスクに複数の入力ファイルが含まれます。タスク数が多すぎると出力またはエグゼキュータープラグインのパフォーマンスに悪影響を与える場合に便利です。（整数、オプション） |


# CLI（TD Toolbelt）を使用した SFTP インポート

## TD Toolbelt をインストールする

最新の [Treasure Data Toolbelt](https://toolbelt.treasuredata.com/) をインストールします。

```
$ td --version
```

## シード設定ファイル（seed.yml）を作成する

以下の例に示すように、SFTP Parquet の詳細を含む *seed.yml* を準備します。2つの認証方法をサポートしています：公開/秘密鍵ペアとパスワード。

### 公開鍵と秘密鍵ペア認証

以下の内容で *seed.yml* を作成します。

```yaml
in:
  type: sftp_parquet
  host: HOST
  port: <PORT, default is 22>
  auth_method: key_pair
  user: USER
  secret_key_file:
    content: |
      -----BEGIN RSA PRIVATE KEY-----
      Proc-Type: 4,ENCRYPTED
      DEK-Info: AES-128-CBC...
      ...
      -----END RSA PRIVATE KEY-----
  secret_key_passphrase: PASSPHRASE
  user_directory_is_root: true
  timeout: 600
  path_prefix: /path/to/parquet/files
out:
  mode: append
  exec: {}
```

`secret\_key\_file` には OpenSSH 形式が必要です。

### パスワード認証

以下の内容で *seed.yml* を作成します。

```yaml
in:
  type: sftp_parquet
  host: HOST
  port: <PORT, default is 22>
  auth_method: password
  user: USER
  password: PASSWORD
  user_directory_is_root: true
  timeout: 600
  path_prefix: /path/to/parquet/files
out:
  mode: append
  exec: {}
```

パスワードには次の特殊文字を使用できます："#$!*@"

SFTP Parquet Integration は、指定されたプレフィックスに一致するすべての Parquet ファイルをインポートします。path_prefix は Parquet ファイルを含むフォルダを指す必要があります（例：path_prefix: `path/to/parquet/files` –> `path/to/parquet/files/data_201501.parquet`、`path/to/parquet/files/data_201502.parquet`、…、`path/to/parquet/files/data_201505.parquet`）。

## フィールドを推測する（load.yml を生成）

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

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

*load.yml* を開くと、推測されたファイル形式定義が表示されます。Parquet ファイルは自己記述的なので、列スキーマは Parquet メタデータから自動的に検出されます。

```yaml
in:
  type: sftp_parquet
  host: HOST
  port: <PORT, default is 22>
  auth_method: key_pair
  user: USER
  secret_key_file:
    content: |
      -----BEGIN RSA PRIVATE KEY-----
      Proc-Type: 4,ENCRYPTED
      DEK-Info: AES-128-CBC...
      ...
      -----END RSA PRIVATE KEY-----
  secret_key_passphrase: PASSPHRASE
  user_directory_is_root: true
  timeout: 600
  path_prefix: /path/to/parquet/files
out:
  mode: append
  exec: {}
```

次に、*preview* コマンドを使用して、システムがファイルをどのように解析するかをプレビューできます。

```
td connector:preview load.yml
```

guess コマンドは Parquet ファイルのメタデータを読み取り、列名と型を自動的に決定します。

システムが予期しない列名や列型を検出した場合、スキーマを変更できますが、Parquet ファイルは通常、明確に定義されたスキーマを持っています。

また、データロードジョブを実行する前に、データベースとテーブルを作成する必要があります。以下の手順に従ってください：

```bash
td database:create td_sample_db
td table:create td_sample_db td_sample_table
```

## ロードジョブを実行する

ロードジョブを送信します。データのサイズによっては、数時間かかる場合があります。データを保存する Treasure AI データベースとテーブルを指定します。

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

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

```bash
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 に存在しない場合、connector:issue コマンドは失敗します。この場合、[データベースを作成](https://docs.treasure.ai/smart/project-product-documentation/creating-or-viewing-a-database) し、[テーブルを作成](https://docs.treasure.ai/smart/project-product-documentation/creating-or-viewing-tables) を手動で行うか、*td connector:issue* コマンドで *--auto-create-table* オプションを使用してデータベースとテーブルを自動作成します：

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

Integration はサーバー側でレコードをソートしません。時間ベースのパーティション分割を効果的に使用するには、事前にレコードをソートしてください。

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

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

## スケジュール実行

インクリメンタル SFTP Parquet ファイルインポート用に定期的な Integration 実行をスケジュールできます。高可用性を確保するために、スケジューラーを慎重に設定しています。この機能を使用することで、ローカルデータセンターに *cron* デーモンが不要になります。

スケジュールインポートの場合、SFTP Parquet 用 Integration は、指定されたプレフィックスに一致するすべてのファイル（例：path_prefix: `path/to/parquet/files` –> `path/to/parquet/files/data_201501.parquet`、`path/to/parquet/files/data_201502.parquet`、…、`path/to/parquet/files/data_201505.parquet`）を最初にインポートし、次回実行のために最後のパス（`path/to/parquet/files/data_201505.parquet`）を記憶します。

2回目以降の実行では、アルファベット順（辞書順）で最後のパス以降のファイルのみをインポートします。（`path/to/parquet/files/data_201506.parquet`、…）

### スケジュールを作成する

*td connector:create* コマンドを使用して新しいスケジュールを作成できます。以下が必要です：スケジュールの名前、cron スタイルのスケジュール、データを保存するデータベースとテーブル、および Integration 設定ファイル。

```bash
td connector:create \
daily_parquet_import \
"10 0 * * *" \
td_sample_db \
td_sample_table \
load.yml
```

Treasure AI のストレージは時間でパーティション分割されているため、*--time-column* オプションを指定することをお勧めします。

```bash
td connector:create \
daily_parquet_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 などのタイムゾーン略語は *サポートされておらず*、予期しないスケジュールにつながる可能性があります。

### スケジュールをリスト表示する

*td connector:list* コマンドを実行すると、現在スケジュールされているエントリのリストが表示されます。

```
td connector:list
```

### 設定とスケジュール履歴を表示する

*td connector:show* は、スケジュールエントリの実行設定を表示します。

```
td connector:show daily_parquet_import
```

*td connector:history* は、スケジュールエントリの実行履歴を表示します。個々の実行結果を調査するには、*td job jobid* を使用します。

```bash
td connector:history daily_parquet_import
```

### スケジュールを削除する

*td connector:delete* はスケジュールを削除します。

```bash
td connector:delete daily_parquet_import
```

## out プラグインのモード

seed.yml の *out* セクションでファイルインポートモードを指定できます。

### append（デフォルト）

これはデフォルトモードで、レコードが対象テーブルに追加されます。

```yaml
in:
  ...
out:
  mode: append
```

### replace

このモードは、対象テーブル内のデータを置き換えます。対象テーブルに加えられた手動スキーマ変更は、このモードでもそのまま保持されることに注意してください。

```
in:
  ...
out:
  mode: replace
```

# Workflow を使用した SFTP Parquet サーバーからのインポート

SFTP サーバーから Parquet ファイルをインポートするサンプルワークフローについては、[Treasure Boxes](https://github.com/treasure-data/treasure-boxes/tree/master/td_load/sftp) を参照してください。