SFTPサーバーからファイルを取り込む

備考

プレビュー

SFTP コネクタはパブリック プレビュー段階です。

LakeFlow Connectを使用して SFTP サーバーからファイルを取り込む方法を学習します。 SFTP コネクタは、Auto Loader 機能を拡張し、Unity Catalog ガバナンスを使用して SFTP サーバーからの安全な増分取り込みを提供します。

主な特徴

SFTP コネクタは以下を提供します。

• 秘密鍵とパスワードベースの認証。
• 1 回限りの保証による増分ファイルの取り込みと処理。
• 自動スキーマ推論、進化、およびデータ救出。
• 安全な取り込みと資格情報のための Unity Catalog ガバナンス。
• ワイド ファイル形式のサポート: JSON 、 CSV 、 XML 、 PARQUET 、 AVRO 、 TEXT 、 BINARYFILE 、およびORC 。
• パターンとワイルドカードのマッチングの組み込みサポートにより、データのサブセットを簡単にターゲットにできます。
• LakeFlow Spark宣言型パイプライン、 Databricks SQL 、サーバーレス、およびDatabricks Runtime 17.3 以降のクラシックを含むすべてのコンピュート タイプで利用可能。

始める前に

接続と取り込みパイプラインを作成するには、次のものが必要です。

• Unity Catalog が有効になっているワークスペース。

• CREATE CONNECTION SFTP接続を作成する権限、またはアクセスモードに基づいて既存の接続を使用するための適切な権限： アクセスモード：

  • 専用アクセスMode ： MANAGE CONNECTION 。
  • 標準アクセスMode ： USE CONNECTION 。

• Databricks Runtimeバージョン 17.3 以降を使用するコンピュート。

SFTPの設定

まず、ソース SFTP サーバーが Databricks クラスター環境にアクセスできることを確認します。

• ワークスペースで設定された VPC でリモート サーバーが使用可能であることを確認します。
• SSH ルールでDatabricks VPCの IP 範囲 (クラシック コンピュートを使用している場合) または安定した IP (サーバーレス コンピュートを使用している場合) が許可されていることを確認します。
• クラシック コンピュート プレーンから、ロード バランサー、NAT ゲートウェイ、インターネット ゲートウェイ、または同等のものを使用して安定した IP アドレスを設定し、 Databricksコンピュートがデプロイされているサブネットに接続します。 これにより、コンピュート リソースは、SFTP サーバー側で許可リストに登録できる安定したパブリック IP アドレスを共有できるようになります。 ネットワーク設定を構成する手順については、 VPC ピアリングを参照してください。
• サーバレス コンピュート プレーンから、ステップ 1: ネットワーク接続構成を作成し、安定した IP をコピーして、安定した出力 IP を取得します。

接続を作成する

SFTP 資格情報を保存するためのUnity Catalog接続を作成します。 CREATE CONNECTION権限が必要です。

コネクタは次の認証方法をサポートしています。

• PEM秘密鍵
• パスワードベースの認証

Databricks では、PEM 秘密キー認証の使用を推奨しています。Databricks では、ソース SFTP サーバー上で最小限の権限を持つ資格情報 (読み取り専用アクセスに制限された非ルート ユーザーなど) を使用することも推奨しています。

パイプラインを作成すると、コネクタは使用可能なホストに一致する接続を自動的に見つけようとします。一致する接続が複数ある場合、コネクタはホストに正常に接続された最初の接続を選択します。ただし、Databricks ではユーザーを明示的に指定することをお勧めします。これにより、コネクタがホスト アクセス権を持つ別のユーザーの接続を選択しないことが保証されます。

PEM秘密鍵（推奨）

Catalog Explorer

1. Databricksワークスペースで、 をクリックします。 カタログ

2. クリック 接続し てから、 「接続」 をクリックします。

3. 「接続を作成」 ボタンをクリックしてください。

4. 接続のセットアップ ウィザードの 「接続の基本」 ページで、一意の 接続名 を入力します。

5. 接続タイプ には SFTP を選択します。

6. 認証タイプ には、 PEM 秘密キー を選択します。

7. 「次へ」 を選択します。

8. [認証] ページの [ホスト] に、外部サーバーのホスト名を入力します。

9. 「ユーザー」 には、外部インスタンスにアクセスするために使用するユーザー ID を入力します。

10. 「次へ」 を選択します。

11. 「接続の詳細」 ページで、PEM 形式で秘密キーを入力します。該当する場合は、キーのパスフレーズも入力します。

12. ホスト キーのフィンガープリントのチェックをスキップする場合は、 「ホスト キーのフィンガープリントを強制する」を オフにします。

    これを選択すると、サーバーの公開キーが予想される SHA-256 フィンガープリントと一致する場合にのみ接続が続行されます。無効にすると、一致に関係なく接続が続行されます。これを無効にする前に、ネットワーク管理者に確認してください。

13. 「ホスト キー フィンガープリントを強制する」を オンにした場合は、SFTP サーバーのフィンガープリントを入力します。

    フィンガープリントは、サーバー管理者から取得するか、CLI コマンドを使用して取得できます。また、 [テスト]を押して接続を作成 > [テスト] することもできます。結果のエラー メッセージには、フィンガープリントが提供されます。例えば：

    ECDSA key fingerprint is SHA256:XXX/YYY

14. [テストと接続の作成] を選択します。

15. 接続が成功したら、 「作成」 をクリックします。

SQL

SQL

-- Create a connection using a username and SSH private key.
CREATE CONNECTION my_sftp_connection
TYPE sftp
OPTIONS (
  host 'my.sftpserver.com',
  -- The following credentials can also be used in-line, but Databricks recommends
  -- accessing them using the secrets scope.
  user secret('my_secret_scope','my_sftp_username'),
  pem_private_key secret('my_secret_scope','my_sftp_private_key'),
  -- Port for the host
  port '22',
  -- Passphrase for the private key (optional).
  pem_key_passphrase secret('my_secret_scope','my_sftp_private_key_passphrase'),
  -- SFTP server fingerprint. You can retrieve this from your server administrator or using CLI commands.
  key_fingerprint 'SHA256:ASampleFingerprintValueZy...',
);

パスワードベースの認証

Catalog Explorer

1. Databricksワークスペースで、 をクリックします。 カタログ

2. クリック 接続し てから、 「接続」 をクリックします。

3. 「接続を作成」 ボタンをクリックしてください。

4. 接続のセットアップ ウィザードの 「接続の基本」 ページで、一意の 接続名 を入力します。

5. 接続タイプ には SFTP を選択します。

6. 認証タイプ には、 ユーザー名とパスワード を選択します。

7. 「次へ」 を選択します。

8. [認証] ページの [ホスト] に、外部サーバーのホスト名を入力します。

9. 「ユーザー」 には、外部インスタンスにアクセスするために使用するユーザー ID を入力します。

10. 「パスワード」 には、外部インスタンスのパスワードを入力します。

11. 「次へ」 を選択します。

12. ホスト キーのフィンガープリントのチェックをスキップする場合は、 「ホスト キーのフィンガープリントを強制する」を オフにします。

    これを選択すると、サーバーの公開キーが予想される SHA-256 フィンガープリントと一致する場合にのみ接続が続行されます。無効にすると、一致に関係なく接続が続行されます。これを無効にする前に、ネットワーク管理者に確認してください。

13. 「ホスト キー フィンガープリントを強制する」を オンにした場合は、SFTP サーバーのフィンガープリントを入力します。

    フィンガープリントは、サーバー管理者から取得するか、CLI コマンドを使用して取得できます。また、 [テスト]を押して接続を作成 > [テスト] することもできます。結果のエラー メッセージには、フィンガープリントが提供されます。例えば：

    ECDSA key fingerprint is SHA256:XXX/YYY

14. [テストと接続の作成] を選択します。

15. 接続が成功したら、 「作成」 をクリックします。

SQL

SQL

-- Create a connection using a username and password.
CREATE CONNECTION my_sftp_connection
TYPE sftp
OPTIONS (
  host 'my.sftpserver.com',
  user secret('my_secret_scope','my_sftp_username'),
  password secret('my_secret_scope','my_sftp_password'),
  -- Port for the host.
  port '22',
  -- SFTP server fingerprint. You can retrieve this from your server administrator or using CLI commands.
  key_fingerprint 'SHA256:ASampleFingerprintValueZy...',
);

SFTPサーバーからファイルを読み取る

次の例は、Auto Loader のストリーミング機能を使用して SFTP サーバーからファイルを読み取る方法を示しています。Auto Loader使用法の詳細については、 「一般的なデータ読み込みパターン」を参照してください。

SFTP URI から読み取る場合、コネクタは URI 内の<host>と<username>を利用可能な接続と照合して、以前に作成した Unity Catalog 接続を自動的に解決します。SFTP URI の<username>と<host>は、接続に保存されている資格情報と一致する必要があります。一致する接続が複数ある場合、コネクタはホストに正常に接続された最初の接続を使用します。

Python

# Run the Auto Loader job to ingest all existing data in the SFTP server.
# The <username> and <host> in the URI must match the connection created in the previous step.
# The connector automatically resolves the matching Unity Catalog connection for authentication.
df = (spark.readStream.format("cloudFiles")
  .option("cloudFiles.schemaLocation", "<path to store schema information>") # This is a cloud storage path
  .option("cloudFiles.format", "csv") # Or other format supported by Auto Loader
  # Specify the absolute path on the SFTP server starting from the root /.
  # Example: /home/<username>/data/files or /uploads/csv_files
  .load("sftp://<username>@<host>:<port>/<absolute_path_to_files>")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "<path to store checkpoint information>") # This is a cloud storage path.
  .trigger(availableNow = True)
  .table("<table name>"))
df.awaitTermination()

次の例は、 LakeFlow Spark宣言型パイプラインのAuto Loaderを使用して SFTP サーバーからファイルを読み取る方法を示しています。

Python

Python

from pyspark import pipelines as dp

# The <username> and <host> in the URI must match the connection created in the previous step.
# The connector automatically resolves the matching Unity Catalog connection for authentication.
@dp.table
def sftp_bronze_table():
  return (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "csv") # Or other format supported by Auto Loader
    # Specify the absolute path on the SFTP server starting from the root /.
    # Example: /home/username/data/files or /uploads/csv_files
    .load("sftp://<username>@<host>:<port>/<absolute_path_to_files>")))

SQL

SQL

-- The <username> and <host> in the URI must match the connection created in the previous step.
-- The connector automatically resolves the matching Unity Catalog connection for authentication.
CREATE OR REFRESH STREAMING TABLE sftp_bronze_table
AS SELECT * FROM STREAM read_files(
  "sftp://<username>@<host>:<port>/<absolute_path_to_files>",
  format => "csv"
)

Auto Loaderオプションを構成します。 次のオプションを除くすべてのオプションがサポートされています:

• cloudFiles.useNotifications
• cloudFiles.useManagedFileEvents
• cloudFiles.cleanSource
• クラウド固有のオプション

制限事項

• SFTP は、 COPY INTO 、 spark.read 、 dbutils.lsなどの他の取り込みサーフェスではサポートされていません。
• SFTP サーバーへの書き戻しはサポートされていません。
• Auto Loader cleanSource (取り込み後にソースでファイルを削除またはアーカイブする) はサポートされていません。
• FTP プロトコルはサポートされていません。

よくある質問

SFTP コネクタに関するよくある質問への回答を見つけます。

ワイルドカードまたはファイル名パターンを使用して、取り込むファイルを選択するにはどうすればよいですか?

SFTP コネクタは、SFTP サーバーから読み取るための標準の Auto Loader フレームワーク上に構築されます。これは、すべての Auto Loader オプションがサポートされていることを意味します。ファイル名のパターンとワイルドカードの場合は、 pathGlobFilterまたはfileNamePatternオプションを使用します。Auto Loaderオプションを参照してください。

SFTP コネクタは暗号化されたファイルを取り込むことができますか?(PGP はサポートされていますか?)

コネクタは送信中に暗号化を解除しませんが、暗号化されたファイルをバイナリ ファイルとして取り込み、取り込み後に暗号化を解除することができます。

互換性のない秘密鍵形式をどのように処理すればよいですか?

PEM 形式のみがサポートされます。次のいずれかの方法で、PEM 形式の秘密鍵を生成できます。

• (オプション 1) 標準の PEM 形式で新しい RSA キーを作成します。

  Shell

  % ssh-keygen -t rsa -m pem

• (オプション 2) 既存の OpenSSH 形式のキーを PEM 形式に変換します。

  Shell

  % ssh-keygen -p -m pem -f /path/to/key  # This updates the key file.

その他のリソース

• Auto Loaderとは
• パイプラインでデータをロードする