FTP のデータをPythonのpetlでETL処理する方法｜CData Connect AI

Jerod Johnson
Director, Technology Evangelism

CData Connect AI Python SDKとpetlフレームワークで、リアルタイムのFTP のデータを抽出・変換・ロードするETLパイプラインの構築方法を解説。読み取りから書き込みまで対応します。

CData Connect AI の Python SDK と petl フレームワークを使うと、FTP のデータをローカルにドライバーをインストールすることなく、Pythonで直接抽出・変換・ロードするETLパイプラインを構築できます。パーソナルアクセストークンで接続し、数行のコードで設定完了です。

Python は豊富なモジュールのエコシステムを備えているため、素早く作業に取りかかり、システムをより効果的に連携できます。CData Connect AI Python SDK と petl フレームワークを使えば、FTP のデータを抽出・変換して CSV などに出力する FTP 連携のアプリケーションやパイプラインを構築できます。この記事では、Connect AI に接続し、petl を使ってFTP のデータを抽出・変換し、CSV ファイルに書き出す方法をご紹介します。

Connect AI Python SDK（cdata-connect-ai）は DB-API 2.0（PEP 249）に準拠したクライアントなので、petl は etl.fromdb を使って SDK の接続から直接読み取れます。ソースごとにドライバーをインストールする必要はありません。パーソナルアクセストークンで接続して、パイプラインを構築しましょう。

Python + petl でできること

FTP データの定期抽出パイプライン構築

SQL クエリで取得した FTP のレコードを petl の etl.fromdb で抽出し、etl.sort などの変換処理をはさんでから CSV に書き出す、軽量な ETL パイプラインを構築できます。

複数データソースとの統合分析

DB-API 2.0 準拠のクライアントなので、同じコードパターンで FTP 以外のデータソースにも接続でき、petl 上で結合・整形してから任意の出力先にロードできます。

FTP への書き戻し処理の自動化

書き込みに対応した権限を持つ PAT で接続していれば、cursor.executemany を使ったバッチ INSERT により、変換済みデータを FTP に自動で書き戻す処理を組み込めます。

Connect AI で FTP に接続する方法は？

CData Connect AI では、直感的なクリック操作ベースのインターフェースでデータソースに接続できます。

Connect AI にログインし、Sources をクリックして、 Add Connection をクリックします

「Add Connection」パネルから「FTP」を選択します

FTP に接続するために必要な認証プロパティを入力します。
FTP は、プレーンテキスト接続およびSSL/TLS 接続の両方をサポートします。FTP サーバーに接続するには、RemoteHost、User、およびPassword を指定します。SSLMode およびSSLServerCert を設定して、TSL/SSL による接続をセキュアにします。TLS/SSL を介した接続の設定について詳しくは、ヘルプドキュメントの「高度な設定」を参照してください。

次の接続プロパティを設定し、ファイルシステムのリレーショナルビューをコントロールします。
- RemotePath：現在の作業ディレクトリに設定。
- TableDepth：ビューとしてレポートするサブフォルダの深度を制御するために設定。
- FileRetrievalDepth：ファイルを再帰的に取得し、Root テーブルにリストするために設定。
ストアドプロシージャは、ファイル、のダウンロード、アップロード、およびプロトコルコマンドの送信に利用できます。 SQL を使用してサーバーと対話する方法の詳細については、ヘルプドキュメントの「データモデル」を参照してください。
「Save & Test」をクリックします
「Permissions」タブに移動し、ユーザーベースの権限を更新します。

パーソナルアクセストークン（PAT）を生成する

Python SDK は、アカウントのメールアドレスとパーソナルアクセストークン（PAT）を使って Connect AI に認証します。アクセスの粒度を保つために、アプリケーションごとに個別の PAT を作成することをおすすめします。

Connect AI アプリの右上にある歯車アイコン（）をクリックして、設定ページを開きます。
設定ページの Access Tokens セクションに移動し、 Create PAT をクリックします。
PAT に名前を付けて Create をクリックします。
パーソナルアクセストークンは作成時にのみ表示されます。必ずコピーして安全な場所に保管してください。

必要なモジュールをインストールする方法は？

pip ユーティリティを使って、SDK と petl フレームワークをインストールします。

pip install cdata-connect-ai
pip install petl

Python で FTP のデータの ETL アプリを構築する方法は？

必要なモジュールをインストールしたら、ETL アプリを構築する準備は完了です。以下にコードスニペットを紹介しますが、完全なソースコードは記事の末尾に掲載しています。

まずモジュールをインポートし、アカウントのメールアドレスと PAT で Connect AI に接続します。

import petl as etl
import cdata_connect_ai

conn = cdata_connect_ai.connect(
    username="[email protected]",
    password="<your_pat>",
)

FTP をクエリする SQL ステートメントを作成する方法は？

SQL を使って、FTP をクエリするステートメントを作成します。この記事では、MyDirectory エンティティからデータを読み取ります。識別子は <Connection>.<Schema>.<Table> という3つの部分で構成されており、接続名はデフォルトでソース名（例：FTP1）になります。

sql = (
    "SELECT Filesize, Filename "
    "FROM [FTP1].[FTP].[MyDirectory] "
    "WHERE FilePath = '/documents/doc.txt'"
)

FTP のデータの抽出・変換・ロードを実装する方法は？

接続とクエリが準備できたら、petl を使ってFTP のデータを抽出・変換・ロードします。この例では、FTP のデータを抽出し、Filename カラムでデータを並べ替えて、CSV ファイルにロードします。

table1 = etl.fromdb(conn, sql)

table2 = etl.sort(table1, 'Filename')

etl.tocsv(table2, 'mydirectory_data.csv')

新しい行を FTP に書き戻す方法は？

FTP が書き込みに対応している場合は、バッチ INSERT で行を書き戻せます。SDK の executemany は、@name のプレースホルダーと、1行につき1つのパラメータ辞書のリストを受け取ります。

cur = conn.cursor()
cur.executemany(
    "INSERT INTO [FTP1].[FTP].[MyDirectory] (Filesize, Filename) "
    "VALUES (@val1, @val2)",
    [
        {"@val1": "New value 1", "@val2": "New value 1"},
        {"@val1": "New value 2", "@val2": "New value 2"},
    ],
)
print(f"Rows inserted: {cur.rowcount}")

conn.close()

注意：書き込み可能なソースであっても、読み取り専用の PAT や接続権限では書き込み操作は拒否されます。

CData Connect AI Python SDK を使えば、petl のような ETL パッケージでのデータへの直接アクセスも含め、通常のデータベースと同じようにFTP のデータを扱えます。