この記事では、ASTERIA WarpからDatabricksへデータを書き込む場合の連携パターンを整理し、パターン別におすすめの方法をご紹介いたします。

Databricksと連携する2つのパターン

Databricksへデータを書き込む方法はいくつかの軸で分類することができます。

・実現手段

ETLツールを活用するのか、プログラミングをするのか。

・データ量

一度に書き込むデータは、多いのか少ないのか。

・書き込み経路

サーバからDatabricksへ直接書き込むのか、外部のクラウドストレージを経由して書き込むのか。

それぞれどのパターンを選択すべきかは要件をふまえた検討が必要ですが、ここではASTERIA WarpからDatabricksへのデータ連携（書き込み）をする場合に候補となる、3つのパターンについてご紹介し、サンプルフローの内容を解説いたします。

いずれのパターンも、ASTERIA Warp の Core エディションを含むすべてのエディションでお使いいただけます。

1. スモールスタートパターン

一度に書き込むデータ量が少ない場合におすすめの方法です。メリットは、少量のデータに対して素早く動作する点です。一方、大量のデータを書き込む場合には処理時間が長くなるなどの課題もあります。

2. 大量データパターン

一度に書き込むデータ量が多い場合におすすめの方法です。メリットは、大量のデータを書き込む場合でも性能の劣化が少ない点です。一方、処理開始時に多少のオーバーヘッドがあるため少量のデータではスモールスタートパターンより処理に時間がかかる場合があるという点はデメリットになります。

なお、このパターンはクラウドストレージサービスと組み合わせて利用する必要があるため、設定の難易度が上がる点には注意が必要です。

3. Auto Loaderパターン

継続的にデータを書き込む場合におすすめの方法です。クラウドストレージサービスと組み合わせて利用する必要があるため、設定の難易度が上がる点には注意が必要ですが、すでにクラウドストレージサービスを利用している場合はシステム構成をシンプルにできる可能性があります。

それでは、各パターンについて詳しく見ていきましょう。

1. スモールスタートパターン

Databricksのほかには次のソフトウェアがあれば利用できます。

・Databricksアダプター

例えば、オンプレミスの共有フォルダにあるCSVファイルを、Databricksアダプターを利用してDatabricksへ書き込む場合は、次のようなステップで実現可能です。

1. Databricksアダプターをセットアップする
2. フローを作成し、実行する

Databricksアダプターのセットアップ方法については、Databricksアダプターの紹介をご参照ください。

ここでは、サンプルフローの内容をご紹介します。

フローの概要は次の通りです。シンプルな「ファイル取得→加工・変換→書き込み」というETLを実現するフローとなっています。

• データを取得
  • FileGetコンポーネントを使って、Databricksへ書き込みたいデータを取得します。
• データを加工・変換
  • マッパーを使って、連携データの内容を加工・変換します。Databricksにデータを渡す前に、必要なデータの追加や不要なデータの削除が可能です。
  • ここではマッパーだけを使った例としていますが、RecordFilterコンポーネントなどを使って必要データだけに絞り込むなどの処理を入れることも可能です。
• Databricksへ書き込む
  • DatabricksPutコンポーネントを使って、加工後のデータをDatabricksのテーブルに直接書き込みます。
  • 

2. 大容量データパターン

これはクラウドストレージサービスへファイルをアップロードし、Databricksへ直接SQLを実行して取り込む処理をASTERIA Warpで実行する方法です。

Databricksのほかには次のソフトウェアやサービスがあれば利用できます。

　・クラウドストレージサービス

利用しているDatabricksとASTERIA Warpが対応している、いずれかのクラウドストレージサービス。今回はAWS上のDatabricksを利用しているので、クラウドストレージサービスはAmazon S3を利用しています。利用しているDatabricksが対応しているクラウドストレージサービスの詳細はDatabricks社のサイトでご確認ください。Databricks on AWS、Databricks on Google Cloud、 Azure Databricks

・利用するクラウドストレージサービスと連携するためのアダプター ※今回は Amazon Web Servicesアダプター を使用。

例えば、オンプレミスの共有フォルダにある大きなCSVファイルを、Amazon S3を経由してDatabricksへ書き込む場合は、次のようなステップで実現可能です。

1. Amazon S3の環境を用意し、Databricksと接続するための設定を行う
2. DatabricksにAmazon S3と接続するための設定を行う
3. Databricksアダプターをセットアップする
4. Amazon Web Servicesアダプターをセットアップする
5. フローを作成し、実行する

Amazon S3の環境設定やDatabricksの環境設定の詳細は、ここでは割愛いたします。各社から提供されているドキュメント等をご参照ください。

DatabricksSQLCallコンポーネントでDatabricksへ直接SQLを実行するために、Databricksアダプターの紹介で作成したコネクションの詳細タブで下記のオプションを追加します。

ここでは、サンプルフローの内容をご紹介します。

フローの概要は次の通りです。

このフローをスケジュール実行することで、定期的にフォルダ監視をし、ファイルがあったらデータ登録を実施するという処理を実現できます。

ここでは、連携するファイルを投入するフォルダ（以下、連携用フォルダ）を1つ決めておき、その連携用フォルダに投入されたファイルをS3へアップロード、DatabricksSQLCallコンポーネントを用いてS3からDatabricksへデータ登録する仕組みを想定しています。

• 連携するデータの一覧を取得
  • FileListコンポーネントを使って、連携用フォルダ内のファイル一覧を取得します。複数のファイルを投入しても処理することができます。
  • ここでは、CSVファイルのみ取得する設定としています。
  • 
• アップロードファイル設定
  • FileListコンポーネントで取得した情報やConst関数に設定した値を利用して、AWS S3Uploadコンポーネントのプロパティを設定しています。

  • 

• S3へファイルアップロード
  • 作成済みのAWSコネクションを指定しています。その他必要なプロパティは前述の「アップロードファイル設定」で設定済みです。
  • 
• Databricks用のSQL文を作成（INSERT INTO）
  • S3にあるファイルをテーブルに書き込むためのSQL文（INSERT INTO）を組み立てています。

  • ヘッダーのないファイルを使用しているため、SQL文で項目名とデータ型を設定しています。FROM句で使用しているread_files関数はオプション指定でファイルフォーマット等を設定可能です。SQLや関数の詳細はDatabricks社のサイトでご確認ください。

     

    • 
• S3からDatabricksへデータを書き込むINSERT文実行
  • S3にあるファイルをテーブルに書き込むためのSQL文（INSERT INTO）を実行します。
  • オプション「Query Passthrough」を設定済みのDatabricksコネクションを指定しています。SQL文は前述の「Databricks用のSQL文を作成（INSERT INTO）」で設定済みです。
  • 

3. Auto Loaderパターン

これは、Databricks社が提供している、クラウドストレージ上の新規ファイルを自動的に検出し、増分処理として効率的に取り込んでデータを処理する仕組みである「Auto Loader」を利用するパターンです。

Databricksのほかには次のソフトウェアやサービスがあれば利用できます。

　・クラウドストレージサービス

利用しているDatabricksとASTERIA Warpが対応している、いずれかのクラウドストレージサービス。今回はAWS上のDatabricksを利用しているので、クラウドストレージサービスはAmazon S3を利用しています。利用しているDatabricksが対応しているクラウドストレージサービスの詳細はDatabricks社のサイトでご確認ください。Databricks on AWS、Databricks on Google Cloud、 Azure Databricks

・利用するクラウドストレージサービスと連携するためのアダプター ※今回は Amazon Web Servicesアダプター を使用。

例えば、オンプレミスの共有フォルダにある大きなCSVファイルを、Amazon S3を経由してAuto Loaderを使ってDatabricksへ書き込む場合は、次のようなステップで実現可能です。

1. Amazon S3の環境を用意し、Databricksと接続するための設定を行う
2. DatabricksにAmazon S3と接続するための設定を行う
3. DatabricksにAuto Loaderの設定を行う
4. Amazon Web Servicesアダプターをセットアップする
5. フローを作成し、実行する

Amazon S3の環境設定やDatabricksの環境設定の詳細は、ここでは割愛いたします。各社から提供されているドキュメント等をご参照ください。

フローの概要は次の通りです。Auto Loaderによって監視されているAmazon S3のバケットにファイルをアップロードし、Auto LoaderジョブをAPI実行することで、AutoLoaderによってファイルがテーブルへ書き込まれます。

• 連携するデータの一覧を取得
  • FileListコンポーネントを使って、連携用フォルダ内のファイル一覧を取得します。複数のファイルを投入しても処理することができます。
  • ここでは、CSVファイルのみ取得する設定としています。
  • 
• アップロードファイル設定
  • FileListコンポーネントで取得した情報やConst関数に設定した値を利用して、AWS S3Uploadコンポーネントのプロパティを設定しています。

  • 

• S3へファイルアップロード
  • 作成済みのAWSコネクションを指定しています。その他必要なプロパティは前述の「アップロードファイル設定」で設定済みです。
  • 
• APIパラメータ設定
  • Auto Loader ジョブを実行するAPIのパラメータを設定します。
  • ジョブIDはJSON形式のリクエストボディに設定、アクセストークンは送信ヘッダーに設定します。

    • 

• Auto Loader ジョブをAPIで実行
  • Auto Loader ジョブをAPIで実行します。 ジョブは、非同期での実行となります。
  • APIの詳細はDatabricks社のサイトでご確認ください。
  • 

パターンの比較

スモールスタートパターンと大容量データパターン、Auto Loaderパターンを比較すると、このような違いがあります。

処理時間比較　*1 *2 *3 *4

共通

*1 処理時間の数値は一例であり、環境や実施タイミングにより変動します。掲載した数値はクラウド上の仮想サーバーにて各項目を5回ずつ処理し、最大と最小を除いた平均値です。

*2 スモールスタートパターンでは、DatabricksPutコンポーネントの「バッチ件数」プロパティを1000件に設定して処理を実施しています。

*3 Auto Loaderパターンの書き込み処理時間は、フローの実行時間とフローから非同期で実行したAuto Loaderの実行時間です。

*4 Auto Loaderは、増分データを継続的に取り込むユースケースに最適化された機能です。今回のように毎回全件を洗い替える処理では、読み込み履歴管理（チェックポイント更新）といったAuto Loader固有の内部処理が不要なオーバーヘッドとなります。今回の結果は、こうした特性を踏まえたものとなっています。
 

さいごに

この記事を参考にしていただくことで、皆さまのASTERIA WarpとDatabricksを使ったデータ活用の推進に貢献することができましたら幸いです。