Apache Airflow でAct-On のデータに連携したワークフローを作る

CData JDBC Driver を使ってApache Airflow からAct-On のデータにアクセスして操作します。

古川えりか
コンテンツスペシャリスト
最終更新日：2022-09-07

この記事で実現できるAct-On 連携のシナリオ

CData

こんにちは！ドライバー周りのヘルプドキュメントを担当している古川です。

Apache Airflow を使うと、データエンジニアリングワークフローの作成、スケジューリング、および監視を行うことができます。CData JDBC Driver for ActOn と組み合わせることで、Airflow からリアルタイムAct-On のデータに連携できます。この記事では、Apache Airflow インスタンスからAct-On のデータに接続してクエリを実行し、結果をCSV ファイルに保存する方法を紹介します。

最適化されたデータ処理が組み込まれたCData JDBC Driver は、リアルタイムAct-On のデータを扱う上で高いパフォーマンスを提供します。 Act-On にSQL クエリを発行すると、CData ドライバーはフィルタや集計などのAct-On 側でサポートしているSQL 操作をAct-On に直接渡し、サポートされていない操作（主にSQL 関数とJOIN 操作）は組み込みSQL エンジンを利用してクライアント側で処理します。組み込みの動的メタデータクエリを使用すると、ネイティブのデータ型を使ってAct-On のデータを操作および分析できます。

Act-On への接続を構成する

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

JDBC URL の作成の補助として、Act-On JDBC Driver に組み込まれている接続文字列デザイナーが使用できます。JAR ファイルをダブルクリックするか、コマンドラインからjar ファイルを実行します。


java -jar cdata.jdbc.acton.jar

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

ActOn はOAuth 認証標準を利用しています。OAuth を使って認証するには、アプリケーションを作成してOAuthClientId、OAuthClientSecret、およびCallbackURL 接続プロパティを取得する必要があります。

認証方法についての詳細は、ヘルプドキュメントの「認証の使用」を参照してください。

組み込みの接続文字列デザイナーを使ってJDBC URL を生成（act-on の場合）

クラスタ環境またはクラウドでJDBC ドライバーをホストするには、ライセンス（フルまたはトライアル）およびランタイムキー（RTK）が必要です。本ライセンス（またはトライアル）の取得については、こちらからお問い合わせください。

以下は、JDBC 接続で要求される必須プロパティです。

プロパティ	値
Database Connection URL	`jdbc:acton:RTK=5246...;InitiateOAuth=GETANDREFRESH`
Database Driver Class Name	cdata.jdbc.acton.ActOnDriver

Airflow でJDBC 接続を確立する

Apache Airflow インスタンスにログインします。
Airflow インスタンスのナビゲーションバーで、「Admin」にカーソルを合わせ、「Connections」をクリックします。
次の画面で「+」マークをクリックして新しい接続を作成します。
Add Connection フォームで、必要な接続プロパティを入力します。
- Connection Id：接続の名前：acton_jdbc
- Connection Type：JDBC Connection
- Connection URL：上記のJDBC 接続URL：jdbc:acton:RTK=5246...;InitiateOAuth=GETANDREFRESH
- Driver Class：cdata.jdbc.acton.ActOnDriver
- Driver Path：PATH/TO/cdata.jdbc.acton.jar
フォームの下にある「Test」ボタンをクリックし、新規の接続をテストします。
新規接続を保存すると、新しく表示される画面に、接続リストに新しい行が追加されたことを示す緑のバナーが表示されます。

DAG を作成する

Airflow におけるDAG は、ワークフローのプロセスを格納するエンティティであり、DAG にトリガーを設定することでワークフローを実行することができます。今回のワークフローでは、シンプルにAct-On のデータに対してSQL クエリを実行し、結果をCSV ファイルに格納します。

はじめに、Home ディレクトリにある「airflow」フォルダに移動します。その中に新しいディレクトリを作成し、タイトルを「dags」とします。ここに、UI に表示されるAirflow のDAG を構築するPython ファイルを格納します。

次に新しいPython ファイルを作成し、タイトルをact-on_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="act-on_hook", schedule_interval="0 10 * * *", start_date=datetime(2022,2,15), catchup=False, tags=['load_csv'])

		# Dag となる関数を定義（取得するテーブルは必要に応じて変更してください）
		def extract_and_load():
		# Define tasks
			@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()

このファイルを保存し、Airflow インスタンスをリフレッシュします。DAG リストの中に、「act-on_hook」というタイトルの新しいDAG が表示されるはずです。
このDAG をクリックし、新しく表示される画面で一時停止解除スイッチをクリックして青色にし、トリガー（＝play）ボタンをクリックしてDAG を実行します。この操作で、act-on_hook.py ファイルのSQL クエリを実行し、結果をCSV としてコード内で指定したファイルパスにエクスポートします。
新規のDAG を実行後、Downloads フォルダ（またはPython スクリプト内で選択したフォルダ）を確認し、CSV ファイルが作成されていることを確認します（本ワークフローの場合はaccount.csv です）。
CSV ファイルを開くと、Apache Airflow によってAct-On のデータがCSV 形式で利用できるようになったことが確認できます。