Apache Airflow で DB2 データを連携

Jerod Johnson
Jerod Johnson
Senior Technology Evangelist
CData JDBC Driver を使用して、Apache Airflow で DB2 のデータ にアクセスして処理。

Apache Airflow は、データエンジニアリングワークフローの作成、スケジューリング、モニタリングをサポートするツールです。 CData JDBC Driver for IBM DB2 と組み合わせることで、Airflow からリアルタイムの DB2 のデータ を扱うことができます。 この記事では、Apache Airflow インスタンスから DB2 のデータ に接続してクエリを実行し、結果を CSV ファイルに保存する方法を説明します。

CData JDBC ドライバーは、最適化されたデータ処理機能を組み込んでおり、 リアルタイムの DB2 のデータ を扱う際に比類のないパフォーマンスを発揮します。複雑な SQL クエリを DB2 に発行すると、 ドライバーはフィルタや集計などのサポートされている SQL 操作を直接 DB2 にプッシュし、 サポートされていない操作（主に SQL 関数や JOIN 操作）は組み込みの SQL エンジンを使用してクライアント側で処理します。 また、組み込みの動的メタデータクエリ機能により、ネイティブのデータ型を使用して DB2 のデータ の操作・分析が可能です。

DB2 への接続を設定

組み込みの接続文字列デザイナー

JDBC URL の構築には、DB2 JDBC Driver に組み込まれている接続文字列デザイナーを使用できます。JAR ファイルをダブルクリックするか、コマンドラインから JAR ファイルを実行してください。

java -jar cdata.jdbc.db2.jar

接続プロパティを入力し、接続文字列をクリップボードにコピーします。

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

  • Server: DB2 を実行するサーバー名。
  • Port: DB2 サーバーのポート。
  • Database: DB2 データベース名。

接続の準備ができたら、認証スキームを選択し、以下で説明するように適切なプロパティを設定します。

CData 製品 は、DB2 への認証に4つの異なるスキームをサポートします。DB2 ユーザー資格情報（デフォルト）、暗号化されたユーザー資格情報、IBM Identity and Access Management（IAM）認証、および Kerberos です。

DB2 ユーザー資格情報

ユーザー資格情報を使用して認証するには、次のプロパティを設定します。
  • AuthScheme：USRIDPWD。
  • User：データベースへのアクセス権を持つユーザーのユーザー名。
  • Password：データベースへのアクセス権を持つユーザーのパスワード。

暗号化されたユーザー資格情報

サーバーがセキュア認証に対応しており、暗号化されたユーザー資格情報を使用して認証を行いたい場合は、このプロパティを設定します。
  • AuthScheme：EUSRIDPWD

IAM、Kerberos で認証したい場合は、ヘルプドキュメントの「はじめに」セクションを参照してください。

パスワード方式によるSSH 接続

パスワード方式によるSSH接続時に必要なプロパティ一覧を以下に示します。

  • User: DB2 のユーザ
  • Password: DB2 のパスワード
  • Database: DB2 の接続先データベース
  • Server: DB2 のサーバー
  • Port: DB2 のポート
  • UserSSH: "true"
  • SSHAuthMode: "Password"
  • SSHPort: SSH のポート
  • SSHServer: SSH サーバー
  • SSHUser: SSH ユーザー
  • SSHPassword: SSH パスワード

接続文字列形式では以下のようになります。

Server=10.0.1.2;Port=50000;User=admin;Password=admin;Database=testUseSSH=true;SSHAuthMode=Password;SSHPort=22;SSHServer=ssh-server;SSHUser=root;SSHPassword=sshpasswd;

公開鍵認証方式方式によるSSH 接続

公開鍵認証によるSSH接続時に必要なプロパティ一覧を以下に示します。

  • User: DB2 のユーザ
  • Password: DB2 のパスワード
  • Database: DB2 の接続先データベース
  • Server: DB2 のサーバー
  • Port: DB2 のポート
  • UserSSH: "true"
  • SSHAuthMode: "Public_Key"
  • SSHClientCertType: キーストアの種類
  • SSHPort: SSH のポート
  • SSHServer: SSH サーバー
  • SSHUser: SSH ユーザー
  • SSHClientCert: 秘密鍵ファイルのパス

接続文字列形式では以下のようになります。

Server=10.0.1.2;Port=50000;User=admin;Password=admin;Database=test;UseSSH=true;SSHAuthMode=Public_Key;SSHClientCertType=PUBLIC_KEY_FILE;SSHPort=22;SSHServer=ssh-server;SSHUser=root;SSHClientCert=C:\Keys\key.pem;

クラスター環境やクラウドで JDBC ドライバーをホストする場合は、ライセンス（製品版またはトライアル版）とランタイムキー（RTK）が必要です。ライセンス（またはトライアル）の取得については、弊社営業チームにお問い合わせください。

以下は、JDBC 接続に必要な主なプロパティです。

プロパティ値
データベース接続 URLjdbc:db2:RTK=5246...;Server=10.0.1.2;Port=50000;User=admin;Password=admin;Database=test;
データベースドライバークラス名cdata.jdbc.db2.DB2Driver

Airflow で JDBC 接続を設定

  1. Apache Airflow インスタンスにログインします。
  2. Airflow インスタンスのナビゲーションバーで、Admin にカーソルを合わせ、Connections をクリックします。
  3. 次の画面で + ボタンをクリックして、新しい接続を作成します。
  4. Add Connection フォームで、必要な接続プロパティを入力します：
    • Connection Id：接続の名前を入力します（例：db2_jdbc）
    • Connection Type：JDBC Connection
    • Connection URL：上記の JDBC 接続 URL（例：jdbc:db2:RTK=5246...;Server=10.0.1.2;Port=50000;User=admin;Password=admin;Database=test;)
    • Driver Class：cdata.jdbc.db2.DB2Driver
    • Driver Path：PATH/TO/cdata.jdbc.db2.jar
  5. フォーム下部の Test ボタンをクリックして、新しい接続をテストします。
  6. 新しい接続を保存すると、次の画面で接続リストに新しい行が追加されたことを示す緑色のバナーが表示されます。

DAG の作成

Airflow の DAG は、ワークフローのプロセスを保存し、トリガーすることでワークフローを実行できるエンティティです。 ここでのワークフローは、DB2 のデータ に対して SQL クエリを実行し、結果を CSV ファイルに保存するというシンプルなものです。

  1. まず、ホームディレクトリに「airflow」フォルダがあるはずです。その中に「dags」という新しいディレクトリを作成します。 ここに Python ファイルを保存すると、UI 上で Airflow DAG として表示されます。
  2. 次に、新しい Python ファイルを作成し、db2_hook.py という名前を付けます。このファイルに以下のコードを挿入してください： 
    	import time
	from datetime import datetime
	from airflow.decorators import dag, task
	from airflow.providers.jdbc.hooks.jdbc import JdbcHook
	import pandas as pd

	# DAG を宣言
	@dag(dag_id="db2_hook", schedule_interval="0 10 * * *", start_date=datetime(2022,2,15), catchup=False, tags=['load_csv'])

	# DAG 関数を定義
	def extract_and_load():
	# タスクを定義
		@task()
		def jdbc_extract():
			try:
				hook = JdbcHook(jdbc_conn_id="jdbc")
				sql = """ select * from Account """
				df = hook.get_pandas_df(sql)
				df.to_csv("/{some_file_path}/{name_of_csv}.csv",header=False, index=False, quoting=1)
				# print(df.head())
				print(df)
				tbl_dict = df.to_dict('dict')
				return tbl_dict
			except Exception as e:
				print("Data extract error: " + str(e))

		jdbc_extract()

	sf_extract_and_load = extract_and_load()
  3. このファイルを保存し、Airflow インスタンスを更新します。DAG のリストに「db2_hook」という新しい DAG が表示されるはずです。
  4. この DAG をクリックし、次の画面で一時停止スイッチをクリックして青色にオンにします。次に、トリガー（再生）ボタンをクリックして DAG を実行します。これにより、db2_hook.py ファイル内の SQL クエリが実行され、コード内で指定したファイルパスに CSV として結果がエクスポートされます。
  5. 新しい DAG をトリガーした後、Downloads フォルダ（または Python スクリプト内で指定した場所）を確認すると、CSV ファイルが作成されていることがわかります。この例では account.csv です。
  6. CSV ファイルを開くと、Apache Airflow によって DB2 のデータ が CSV 形式で利用可能になっていることを確認できます。

詳細情報と無料トライアル

CData JDBC Driver for IBM DB2 の30日間無料トライアルをダウンロードして、Apache Airflow でリアルタイムの DB2 のデータ を活用してみてください。ご質問があれば、サポートチームまでお気軽にお問い合わせください。

はじめる準備はできましたか？

IBM DB2 Driver の無料トライアルをダウンロードしてお試しください：

 ダウンロード

詳細：

IBM DB2 Icon IBM DB2 JDBC Driver お問い合わせ

IBM DB2 連携のパワフルなJava アプリケーションを素早く作成して配布。

この記事の内容

関連記事