Python でLakebase のデータを変換・出力するETL 処理を作る方法

Pythonエコシステムには多くのモジュールがあり、システム構築を素早く効率的に行うことができます。本記事では、CData Python Connector for Lakebase とpetl フレームワークを使って、Lakebase のデータにPython から接続してデータを変換、CSV に出力するETL 変換を実装してみます。

CData Python Connector は効率的なデータ処理によりLakebase のデータ にPython から接続し、高いパフォーマンスを発揮します。Lakebase にデータをクエリする際、ドライバーはフィルタリング、集計などがサポートされている場合SQL 処理を直接Lakebase 側に行わせ、サポートされていないSQL 処理については、組み込みのSQL エンジンによりクライアント側で処理を行います（JOIN やSQL 関数など）。

必要なモジュールのインストール

pip で必要なモジュールおよびフレームワークをインストールします：

pip install petl
pip install pandas

Python でLakebase のデータをETL 処理するアプリを構築

モジュールとフレームワークをインストールしたら、ETL アプリケーションを組んでいきます。コードのスニペットは以下の通りです。フルコードは記事の末尾に付いています。

CData Connector を含むモジュールをインポートします。

import petl as etl
import pandas as pd
import cdata.lakebase as mod

接続文字列で接続を確立します。connect 関数を使って、CData Lakebase Connector からLakebase への接続を行います

cnxn = mod.connect("DatabricksInstance=lakebase;Server=127.0.0.1;Port=5432;Database=my_database;")

Databricks Lakebase に接続するには、以下のプロパティを設定します。

• DatabricksInstance: Databricks インスタンスまたはサーバーホスト名を指定します。形式は instance-abcdef12-3456-7890-abcd-abcdef123456.database.cloud.databricks.com です。
• Server: Lakebase データベースをホストするサーバーのホスト名または IP アドレスを指定します。
• Port（オプション）: Lakebase データベースをホストするサーバーのポート番号を指定します。デフォルトは 5432 です。
• Database（オプション）: Lakebase サーバーへの認証後に接続するデータベースを指定します。デフォルトでは認証ユーザーのデフォルトデータベースに接続します。

OAuth クライアント認証

OAuth クライアント資格情報を使用して認証するには、サービスプリンシパルで OAuth クライアントを構成します。手順の概要は以下のとおりです。

1. 新しいサービスプリンシパルを作成・構成する
2. サービスプリンシパルに権限を割り当てる
3. サービスプリンシパル用の OAuth シークレットを作成する

詳細については、ヘルプドキュメントの「Setting Up OAuthClient Authentication」セクションをご参照ください。

OAuth PKCE 認証

PKCE（Proof Key for Code Exchange）を使用した OAuth code タイプで認証するには、以下のプロパティを設定します。

• AuthScheme: OAuthPKCE を指定します。
• User: 認証ユーザーのユーザー ID を指定します。

詳細については、ヘルプドキュメントをご参照ください。

Lakebase をクエリするSQL 文の作成

Lakebase にはSQL でデータアクセスが可能です。Orders エンティティからのデータを読み出します。

sql = "SELECT ShipName, ShipCity FROM Orders WHERE ShipCountry = 'USA'"

Lakebase データのETL 処理

DataFrame に格納されたクエリ結果を使って、petl でETL（抽出・変換・ロード）パイプラインを組みます。この例では、Lakebase のデータ を取得して、ShipCity カラムでデータをソートして、CSV ファイルにデータをロードします。

table1 = etl.fromdb(cnxn,sql)

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

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

CData Python Connector for Lakebase を使えば、データベースを扱う場合と同感覚で、Lakebase のデータ を扱うことができ、petl のようなETL パッケージから直接データにアクセスが可能になります。

おわりに

Lakebase Python Connector の30日の無償トライアル をぜひダウンロードして、Lakebase のデータ への接続をPython アプリやスクリプトから簡単に作成しましょう。

フルソースコード

import petl as etl
import pandas as pd
import cdata.lakebase as mod

cnxn = mod.connect("DatabricksInstance=lakebase;Server=127.0.0.1;Port=5432;Database=my_database;")

sql = "SELECT ShipName, ShipCity FROM Orders WHERE ShipCountry = 'USA'"

table1 = etl.fromdb(cnxn,sql)

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

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