Directory

Directory コネクターを使用すると、Data Connection の Agent が稼働しているホスト上にあるファイルを直接取り込めます。このコネクターは、Agent マシン上でローカルに生成または保存されたファイルを Foundry に同期する必要がある場合に役立ちます。

サポートされている機能

データモデル

このコネクターは、任意の形式のファイルを Foundry データセットに転送できます。ファイル形式は保持され、転送中も転送後もスキーマは適用されません。データにアクセスするには、必要なスキーマを出力データセットに適用するか、下流のデータ加工を作成します。

セットアップ

1. Data Connection アプリケーションを開き、画面の右上にある + 新規ソースを選択します。
2. 利用可能なコネクターの種類から Directory を選択します。
3. ソースは Agent ワーカーで実行されるように設定されます。
4. 追加の設定プロンプトに従って、コネクターのセットアップを続けます。

Foundry でのコネクターのセットアップについて詳しくはこちら。

設定オプション

Directory からデータを同期する

Directory コネクターはファイルベースの同期インターフェースを使用します。

外部トランスフォームを使用して Agent ホストからファイルを取り込む

柔軟性と制御性を高めるには、外部トランスフォームを使用して Agent ホストからファイルを取り込めます。この方法では、リモートの Agent ホスト上のファイルにアクセスしながら、Foundry ワーカーで同期ロジックを実行できます。

前提条件

1. REST API ソースを作成する：Data Connection アプリケーションに移動し、新しい REST API ソースを作成します。
2. 接続情報を設定する：
   • ドメインを Agent ホストのアドレスに設定します（プライベート IP アドレスの場合は、プレースホルダーのドメイン名を使用します。下記を参照してください）。
   • ポートを 22（SSH）に設定します。
   • ホストに SSH 接続できるユーザーの SSH ユーザー名とパスワードをシークレットとして追加します。
3. Agent プロキシのエグレスポリシーを追加する： Agent ホストのアドレス（プライベート IP アドレスの場合は、プレースホルダーのドメイン名を使用します。下記を参照してください）に対して、Agent 自体を使用する Agent プロキシのエグレスポリシーを作成します。これにより、Foundry ワーカーは Agent を経由してトラフィックをルーティングし、Agent ホストに到達できるようになります。

4. ソースをコードリポジトリにインポートする：外部トランスフォームのセットアップガイドに従って、ソースを Python トランスフォームのリポジトリにインポートします。

Python トランスフォームのリポジトリに paramiko ライブラリがインストールされていることを確認してください。コードリポジトリの左側のパネルにあるライブラリタブから追加できます。

例：SSH 経由で Agent ホストからファイルを読み取る

次の例では、Paramiko ↗ Python ライブラリを使用して、SSH 経由で Agent ホストに接続し、ファイルを Foundry データセットに読み込む方法を示します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from transforms.api import transform, Output, Input, LightweightOutput, LightweightInput, lightweight
from transforms.external.systems import external_systems, Source, ResolvedSource
import paramiko


@lightweight
@external_systems(
    agent_source=Source("<source_rid>")  # REST API ソースの RID に置き換えます
)
@transform(
    output_dataset=Output("<output_dataset_rid>"),  # 出力データセットの RID に置き換えます
    files_to_read=Input("<input_dataset_rid>"),  # 読み取るファイルのパスを含むデータセット
)
def compute(
    agent_source: ResolvedSource,
    output_dataset: LightweightOutput,
    files_to_read: LightweightInput,
):
    """
    SSH 経由でリモートの Agent ホストからファイルを読み取り、Foundry データセットに書き込みます。
    """
    # 1. SSH 接続のセットアップ
    hostname = "<agent_hostname>"  # Agent のホスト名に置き換えます
    username = "<ssh_username>"  # SSH ユーザー名に置き換えます
    password = agent_source.get_secret("<password_secret_name>")  # シークレット名に置き換えます

    # 2. SSH 接続を確立する
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(hostname, username=username, password=password)

    # 3. 入力データセットからファイルパスを読み取る
    remote_file_paths = files_to_read.pandas()["remote_file_path"].tolist()

    # 4. SFTP 接続を開く
    sftp = client.open_sftp()

    # 5. 各ファイルを読み取り、出力データセットに書き込む
    for remote_path in remote_file_paths:
        with sftp.open(remote_path, "rb") as remote_file:
            file_binary_data = remote_file.read()

        # パスからファイル名を抽出し、出力に書き込みます
        filename = remote_path.split("/")[-1]
        with output_dataset.filesystem().open(filename, "wb") as f:
            f.write(file_binary_data)

    # 6. 接続を閉じる
    sftp.close()
    client.close()

例：SSH 経由で Agent ホストからファイルを削除する

取り込み後に Directory ソースからファイルが確実に削除される必要がある場合は、完了ストラテジーに頼るのではなく、外部トランスフォームを使用できます。完了ストラテジーによる削除はベストエフォートでのみ実行されます。

次の例では、処理後に Agent ホストからファイルを削除する方法を示します。この方法を使用するには、次の手順を実行します。

1. Directory ソースからファイルを取り込み、正常に取り込まれたファイルのパスのリストをデータセットに出力する上流の同期を作成します。
2. 取り込まれたファイルのパスのデータセットを入力として使用し、同期の完了後にこの削除トランスフォームが実行されるようにスケジュールします。
3. 出力データセットには各ファイルの削除ステータスが含まれるため、どのファイルが正常に削除されたかを監査できます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from transforms.api import transform, Output, Input, LightweightOutput, LightweightInput, lightweight
from transforms.external.systems import external_systems, Source, ResolvedSource
import paramiko


@lightweight
@external_systems(
    agent_source=Source("<source_rid>")  # REST API ソースの RID に置き換えます
)
@transform(
    output_dataset=Output("<output_dataset_rid>"),  # 出力データセットの RID に置き換えます
    files_to_delete=Input("<input_dataset_rid>"),  # 削除するファイルのパスを含むデータセット
)
def compute(
    agent_source: ResolvedSource,
    output_dataset: LightweightOutput,
    files_to_delete: LightweightInput,
):
    """
    SSH 経由でリモートの Agent ホストからファイルを削除します。
    入力データセットには、削除するファイルの絶対パスを含む列 'remote_file_path' が必要です。
    """
    hostname = "<agent_hostname>"  # Agent のホスト名に置き換えます
    username = "<ssh_username>"  # SSH ユーザー名に置き換えます
    password = agent_source.get_secret("<password_secret_name>")  # シークレット名に置き換えます

    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(hostname, username=username, password=password)

    remote_file_paths = files_to_delete.pandas()["remote_file_path"].tolist()

    sftp = client.open_sftp()

    deletion_results = []
    for remote_path in remote_file_paths:
        try:
            sftp.remove(remote_path)
            deletion_results.append({"file": remote_path, "status": "deleted"})
        except FileNotFoundError:
            deletion_results.append({"file": remote_path, "status": "not_found"})
        except PermissionError:
            deletion_results.append({"file": remote_path, "status": "permission_denied"})

    sftp.close()
    client.close()

    # 削除結果を出力データセットに書き込みます
    import pandas as pd
    results_df = pd.DataFrame(deletion_results)
    output_dataset.write_pandas(results_df)