Apache Airflow で Dynamics 365 データを連携
Apache Airflow は、データエンジニアリングワークフローの作成、スケジューリング、モニタリングをサポートするツールです。 CData JDBC Driver for Dynamics 365 と組み合わせることで、Airflow からリアルタイムの Dynamics 365 のデータ を扱うことができます。 この記事では、Apache Airflow インスタンスから Dynamics 365 のデータ に接続してクエリを実行し、結果を CSV ファイルに保存する方法を説明します。
CData JDBC ドライバーは、最適化されたデータ処理機能を組み込んでおり、 リアルタイムの Dynamics 365 のデータ を扱う際に比類のないパフォーマンスを発揮します。複雑な SQL クエリを Dynamics 365 に発行すると、 ドライバーはフィルタや集計などのサポートされている SQL 操作を直接 Dynamics 365 にプッシュし、 サポートされていない操作(主に SQL 関数や JOIN 操作)は組み込みの SQL エンジンを使用してクライアント側で処理します。 また、組み込みの動的メタデータクエリ機能により、ネイティブのデータ型を使用して Dynamics 365 のデータ の操作・分析が可能です。
Dynamics 365 データ連携について
CData は、Microsoft Dynamics 365 のライブデータへのアクセスと統合を簡素化します。お客様は CData の接続機能を以下の目的で活用しています:
- Sales、Customer Service、Finance & Operations、Marketing など、Dynamics 365 エコシステム全体のデータの読み取りと書き込みができます。
- カスタマイズ可能なキャッシュと、インテリジェントなクエリ集約・分離により、Dynamics CRM のネイティブ機能を拡張できます。
- Azure Active Directory、Azure マネージド サービス ID 認証情報、クライアントシークレットまたは証明書を使用した Azure サービスプリンシパルなど、さまざまな方法で Dynamics 365 に安全に認証できます。
- SQL ストアドプロシージャを使用して、Dynamics 365 エンティティを管理できます。エンティティ間の関連付けの一覧表示、作成、削除などが可能です。
CData のお客様は、データをデータウェアハウスにレプリケートしたい場合(他のデータソースと併せて)や、Microsoft エコシステム内のお気に入りのデータツール(Power BI、Excel など)または外部ツール(Tableau、Looker など)からライブ Dynamics 365 データを分析したい場合など、さまざまな理由で当社の Dynamics 365 接続ソリューションを使用しています。
はじめに
Dynamics 365 への接続を設定
組み込みの接続文字列デザイナー
JDBC URL の構築には、Dynamics 365 JDBC Driver に組み込まれている接続文字列デザイナーを使用できます。JAR ファイルをダブルクリックするか、コマンドラインから JAR ファイルを実行してください。
java -jar cdata.jdbc.dynamics365.jar
接続プロパティを入力し、接続文字列をクリップボードにコピーします。
Dynamics 365 接続プロパティの取得・設定方法
Microsoft Dynamics 365 への接続
Microsoft Dynamics 365 ドライバーは、以下のMicrosoft Dynamics 365 エディションへの接続をサポートしています。
- CustomerService
- FieldService
- FinOpsOnline (デフォルト)
- FinOpsOnPremise
- HumanResources
- Marketing
- ProjectOperations
- Sales
Notes:
- Supply Chain Management はFinance and Operations と同一です。これらのいずれかに接続するには、Edition をFinOpsOnline またはFinOpsOnPremise のいずれかに設定します。
- Microsoft Dynamics 365 Business Central については、個別のMicrosoft Dynamics 365 Business Central ドライバーを使用してください。
- OrganizationURL:お使いのMicrosoft Dynamics 365 組織のURL。例えば、https://orgcb42e1d0.crm.dynamics.com。
- Edition:上記のエディション一覧に示すとおり。
Microsoft Dynamics 365 への認証
Microsoft Dynamics 365 は、Microsoft Entra ID(Azure AD)、Azure サービスプリンシパル、Azure マネージドID(MSI)を経由する認証をサポートします。これらはすべてOAuth 規格に基づきます。 認証方法の詳細は、ヘルプドキュメントを参照してください。
クラスター環境やクラウドで JDBC ドライバーをホストする場合は、ライセンス(製品版またはトライアル版)とランタイムキー(RTK)が必要です。ライセンス(またはトライアル)の取得については、弊社営業チームにお問い合わせください。
以下は、JDBC 接続に必要な主なプロパティです。
| プロパティ | 値 |
|---|---|
| データベース接続 URL | jdbc:dynamics365:RTK=5246...;OrganizationUrl=https://myaccount.operations.dynamics.com/;Edition=Sales; |
| データベースドライバークラス名 | cdata.jdbc.dynamics365.Dynamics365Driver |
Airflow で JDBC 接続を設定
- Apache Airflow インスタンスにログインします。
- Airflow インスタンスのナビゲーションバーで、Admin にカーソルを合わせ、Connections をクリックします。
- 次の画面で + ボタンをクリックして、新しい接続を作成します。
- Add Connection フォームで、必要な接続プロパティを入力します:
- Connection Id:接続の名前を入力します(例:dynamics365_jdbc)
- Connection Type:JDBC Connection
- Connection URL:上記の JDBC 接続 URL(例:jdbc:dynamics365:RTK=5246...;OrganizationUrl=https://myaccount.operations.dynamics.com/;Edition=Sales;)
- Driver Class:cdata.jdbc.dynamics365.Dynamics365Driver
- Driver Path:PATH/TO/cdata.jdbc.dynamics365.jar
- フォーム下部の Test ボタンをクリックして、新しい接続をテストします。
- 新しい接続を保存すると、次の画面で接続リストに新しい行が追加されたことを示す緑色のバナーが表示されます。
DAG の作成
Airflow の DAG は、ワークフローのプロセスを保存し、トリガーすることでワークフローを実行できるエンティティです。 ここでのワークフローは、Dynamics 365 のデータ に対して SQL クエリを実行し、結果を CSV ファイルに保存するというシンプルなものです。
- まず、ホームディレクトリに「airflow」フォルダがあるはずです。その中に「dags」という新しいディレクトリを作成します。 ここに Python ファイルを保存すると、UI 上で Airflow DAG として表示されます。
- 次に、新しい Python ファイルを作成し、dynamics 365_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="dynamics 365_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() - このファイルを保存し、Airflow インスタンスを更新します。DAG のリストに「dynamics 365_hook」という新しい DAG が表示されるはずです。
- この DAG をクリックし、次の画面で一時停止スイッチをクリックして青色にオンにします。次に、トリガー(再生)ボタンをクリックして DAG を実行します。これにより、dynamics 365_hook.py ファイル内の SQL クエリが実行され、コード内で指定したファイルパスに CSV として結果がエクスポートされます。
- 新しい DAG をトリガーした後、Downloads フォルダ(または Python スクリプト内で指定した場所)を確認すると、CSV ファイルが作成されていることがわかります。この例では account.csv です。
- CSV ファイルを開くと、Apache Airflow によって Dynamics 365 のデータ が CSV 形式で利用可能になっていることを確認できます。