Azure Databricks でElasticsearch のデータに接続してデータ処理を行う方法
Databricks は、Apache Spark によるデータ処理機能を提供するクラウドベースのサービスです。CData JDBC ドライバと組み合わせることで、Databricks を使用してリアルタイムElasticsearch のデータのデータエンジニアリングとデータサイエンスを実行できます。この記事では、Azure で CData JDBC ドライバをホストし、Databricks からリアルタイムElasticsearch のデータに接続してデータを処理する方法を説明します。
最適化されたデータ処理機能を組み込んだ CData JDBC ドライバは、リアルタイムElasticsearch のデータとのインタラクションにおいて卓越したパフォーマンスを発揮します。Elasticsearch に対して複雑な SQL クエリを発行すると、ドライバーはフィルタや集計などのサポートされている SQL 操作を直接Elasticsearchにプッシュし、サポートされていない操作(主に SQL 関数や JOIN 操作)は組み込みの SQL エンジンを使用してクライアント側で処理します。動的メタデータクエリ機能により、ネイティブのデータ型を使用してElasticsearch のデータの操作・分析が可能です。
Elasticsearch データ連携について
CData を使用すれば、Elasticsearch のライブデータへのアクセスと統合がこれまでになく簡単になります。お客様は CData の接続機能を以下の目的で利用しています:
- SQL エンドポイントと REST エンドポイントの両方にアクセスでき、接続を最適化し、Elasticsearch データの読み書きに関してより多くのオプションを提供します。
- v2.2 以降およびオープンソース Elasticsearch サブスクリプションを含む、ほぼすべての Elasticsearch インスタンスに接続できます。
- SCORE() 関数を明示的に要求することなく、常にクエリ結果の関連性スコアを受け取ることができます。これにより、サードパーティツールからのアクセスが簡素化され、クエリ結果のテキスト関連性のランキングを簡単に確認できます。
- 複数のインデックスを検索でき、クライアントマシンではなく Elasticsearch がクエリと結果の管理・処理を担当します。
ユーザーは、Crystal Reports、Power BI、Excel などの分析ツールと Elasticsearch データを統合し、当社のツールを活用して、Elasticsearch を含むすべてのデータソースへの単一のフェデレートアクセスレイヤーを実現しています。
CData の Elasticsearch ソリューションの詳細については、ナレッジベース記事をご覧ください:CData Elasticsearch Driver Features & Differentiators
はじめに
CData JDBC ドライバを Azure にインストール
Databricks でリアルタイムElasticsearch のデータを操作するには、Azure Data Lake Storage(ADLS)を通じてドライバーをインストールします。(以前のバージョンの記事で説明していた DBFS を介した接続方法は非推奨となっていますが、廃止日は公開されていません。)
- JDBC JAR ファイルを任意の Blob コンテナにアップロードします(例:「databrickslibraries」ストレージアカウントの「jdbcjars」コンテナ)。
- ストレージアカウントから「セキュリティとネットワーク」を展開し、「アクセスキー」をクリックしてアカウントキーを取得します。使用するキーを表示してコピーしてください。
- コンテナに移動し、JAR を保存している特定のコンテナを開き、JDBC JAR ファイルのエントリを選択して JAR ファイルの URL を取得します。ファイルの詳細が開き、URL をクリップボードにコピーするボタンがあります。この値は以下のようになります(「blob」の部分はストレージアカウントの種類によって異なる場合があります):
https://databrickslibraries.blob.core.windows.net/jdbcjars/cdata.jdbc.salesforce.jar
- Databricks クラスターの「Configuration」タブで「Edit」ボタンをクリックし、「Advanced options」を展開します。そこで、以下の Spark オプション(JAR URL のドメイン名から派生)に、コピーしたアカウントキーを値として追加し、「Confirm」をクリックします:
spark.hadoop.fs.azure.account.key.databrickslibraries.blob.core.windows.net
- Databricks クラスターの「Libraries」タブで「Install new」をクリックし、ADLS オプションを選択します。ドライバー JAR の ABFSS URL(これも JAR URL のドメイン名から派生)を指定し、「Install」をクリックします。ABFSS URL は以下のようになります:
abfss://[email protected]/cdata.jdbc.salesforce.jar
Databricks からElasticsearchに接続
JAR ファイルがインストールされたら、Databricks でリアルタイムElasticsearch のデータを操作する準備が整いました。まず、ワークスペースで新しいノートブックを作成します。ワークブックに名前を付け、言語として Python が選択されていることを確認し(デフォルトで選択されているはずです)、「Connect」をクリックして「General Compute」から JDBC ドライバーをインストールしたクラスターを選択します(デフォルトで選択されているはずです)。
Elasticsearchへの接続を設定
JDBC ドライバのクラスを参照し、JDBC URL で使用する接続文字列を構築してElasticsearchに接続します。また、JDBC URL に RTK プロパティを設定する必要があります(Beta ドライバーを使用している場合を除く)。このプロパティの設定方法については、インストールに含まれるライセンスファイルを参照してください。
driver = "cdata.jdbc.elasticsearch.ElasticsearchDriver" url = "jdbc:elasticsearch:RTK=5246...;Server=127.0.0.1;Port=9200;User=admin;Password=123456;"
組み込みの接続文字列デザイナー
JDBC URL の構築には、Elasticsearch JDBC Driver に組み込まれている接続文字列デザイナーを使用できます。JAR ファイルをダブルクリックするか、コマンドラインから JAR ファイルを実行してください。
java -jar cdata.jdbc.elasticsearch.jar
接続プロパティを入力し、接続文字列をクリップボードにコピーします。
Elasticsearch 接続プロパティの取得・設定方法
接続するには、Server およびPort 接続プロパティを設定します。 認証には、User とPassword プロパティ、PKI (public key infrastructure)、またはその両方を設定します。 PKI を使用するには、SSLClientCert、SSLClientCertType、SSLClientCertSubject、およびSSLClientCertPassword プロパティを設定します。
CData 製品は、認証とTLS/SSL 暗号化にX-Pack Security を使用しています。TLS/SSL で接続するには、Server 値に'https://' を接頭します。Note: PKI を 使用するためには、TLS/SSL およびクライアント認証はX-Pack 上で有効化されていなければなりません。
接続されると、X-Pack では、設定したリルムをベースにユーザー認証およびロールの許可が実施されます。
Elasticsearch のデータの読み込み
接続を設定したら、CData JDBC ドライバと接続情報を使用してElasticsearch のデータをデータフレームとして読み込むことができます。
remote_table = spark.read.format ( "jdbc" ) \ .option ( "driver" , driver) \ .option ( "url" , url) \ .option ( "dbtable" , "Orders") \ .load ()
Elasticsearch のデータの表示
読み込んだElasticsearch のデータを display 関数で確認してみましょう。
display (remote_table.select ("OrderName"))
Azure Databricks でElasticsearch のデータを分析
Databricks SparkSQL でデータを処理したい場合は、読み込んだデータを一時ビューとして登録します。
remote_table.createOrReplaceTempView ( "SAMPLE_VIEW" )
以下の SparkSQL で分析用のElasticsearch のデータを取得できます。
result = spark.sql("SELECT SAMPLE_VIEW.Freight, Customers.ContactName FROM Customers INNER JOIN SAMPLE_VIEW ON Customers.CustomerId=SAMPLE_VIEW.CustomerId")
Elasticsearch からのデータは、対象のノートブック内でのみ利用可能です。他のユーザーと共有したい場合は、テーブルとして保存してください。
remote_table.write.format ( "parquet" ) .saveAsTable ( "SAMPLE_TABLE" )
CData JDBC Driver for Elasticsearch の30日間の無償トライアルをダウンロードして、Azure Databricks でリアルタイムElasticsearch のデータを活用してみてください。ご不明な点があれば、サポートチームまでお気軽にお問い合わせください。