こんにちは。CData Software Japanリードエンジニアの杉本です。

前回は Apache の全文検索エンジン、SolrでのCData JDBC Driver利用方法を紹介しました。

www.cdatablog.jp

でも、Apache Solrを紹介したからには、Elasticsearchも外せないですね。

今回の記事ではCData JDBC Driver、そしてLogstash を使って Elasticsearchへデータを転送する方法を紹介したいと思います。

LogStashとは？

LogStash は分散処理マルチテナント対応検索エンジン「ElasticSeach」などを提供するElastic社がオープンソースで開発・提供しているデータ処理パイプラインツールです。

www.elastic.co

様々なフォーマット、サービス、ログなどのデータをElasticSeachへ転送することによく利用されているツールですが、柔軟なPlugin機能により、ログやElasticSeach以外にも数多くのInput・Output先に対応しています。

そんなLogstashではJDBC Driverを利用できるPluginが提供されており、MySQLやPostgreSQLなどのRDBデータを転送することも可能になっています。

今回はそのJDBC 用のPluginとCData JDBC Driverを組み合わせて、ElasticSeachへデータを取り込む方法を紹介します。

本記事のシナリオ

今回取り込む対象のデータソースは Atlassian が提供する課題 & プロジェクト追跡ソフトウェアのJIRAとしました。

Logstashは前述の通り、様々なInput Pluginが提供されており、多種多様なサービスからデータを取り込むことが可能ですが、現在JIRAはデータソースとして対応していません。

https://www.elastic.co/guide/en/logstash/current/input-plugins.html

もちろん、Elasticsearch に取り込むので、Elasticsearch のREST APIとJIRA の REST APIを組み合わせて実装することも可能ですが、各APIへの連携実装や保守・メンテナンスのコストが多くかかってしまいます。

そこでこの記事では Logstash がもともと備えているJDBCのインターフェースを利用することでこのボトルネックを解消します。

CData Software では SalesforceやKintone、JIRAなどに接続することができる各種JDBC Driverを200種類ほど提供しています。

https://www.cdata.com/jp/jdbc/#drivers

この各種JDBC Driverを用いることで、各APIの仕様を意識せず Logstash からのデータ取り込みを実行でき、シームレスに Elasticsearch へJIRAのデータを転送できるようになります。

手順

必要なもの

• Elasticsearch
• Logstash
• Kibana
• JIRA アカウント（JIRA Core）
• CData JIRA JDBC Driver

Elasticsearch・Logstash・Kibana はそれぞれ7.8.0 を使用しました。予めWindows版をダウンロードして、ZIPファイルを解凍した状態になっています。

また、この記事ではJDBC Input Pluginを利用して、CData JDBC Driverからデータを取り込みます。

https://www.elastic.co/guide/en/logstash/5.4/plugins-inputs-jdbc.html

JDBC Pluginは最新のLogstashだとデフォルトでついてきますが、バージョンによっては追加する必要があります。

JIRA JDBC Driver のインストール

最初にJIRAからデータを取得するためのインターフェースとなるJDBC Driverをインストールします。トライアルは以下のURLから入手できます。

https://www.cdata.com/jp/drivers/jira/jdbc/

ダウンロード後、exeファイルを実行し、セットアップを進めます。

CData JIRA JDBC Driver のインストールが完了すると、「C:\Program Files\CData\CData JDBC Driver for JIRA 2019J\lib\」フォルダにJDBC Driverの本体であるjarファイルが生成されます。

そして下記２つのファイルを

C:\Program Files\CData\CData JDBC Driver for JIRA 2019J\lib\cdata.jdbc.jira.jar

C:\Program Files\CData\CData JDBC Driver for JIRA 2019J\lib\cdata.jdbc.jira.lic

Logstashの「/logstash-core/lib/jars/」に移動します。

JIRAに接続するためのAPIトークンを取得

次にJIRAへAPIアクセスを行うためのAPIトークンを取得します。APIトークンの取得方法については下記URLを参照してみてください。

https://ja.confluence.atlassian.com/cloud/api-tokens-938839638.html

Elasticsearch・Kibanaの起動

今回はJiraからElasticsearchにデータを取り込むので、予めElasticsearchを起動しておきましょう。また後ほどKibana上で動作確認をするので、こちらも立ち上げておきます。

> Elasticsearch-7.8.0\bin\Elasticsearch

> kibana-7.8.0-windows-x86_64\bin\kibana

起動後、「http://localhost:9200/」を参照して、Elasticsearchにアクセスできることを確認します。

「http://localhost:5601/」でKibanaにもアクセスできました。

Logstash用のconfファイルの作成

それでは、Logstashでデータ転送を行うための設定ファイルを作成していきたいと思います。

今回はJIRAの課題（Issue）データをElasticsearchに転送します。

まず取得するデータを確認しましょう。

CData JIRA JDBC DriverはJIRAのAPIをSQLで実行できるようにしている製品なので、以下のヘルプにあるように各APIリソース毎にスキーマを提供しています。

http://cdn.cdata.com/help/BJE/jp/jdbc/pg_alltables.htm

JIRAのIssueが取れるのは「Issues」テーブルなので、これを取得するクエリを記述します。Logstashだけだとちょっとどんなデータが取れるかわかりづらいので、JDBCのSQLを手軽に実行できるDbVizualizerで少し見てみましょう。

接続手順は以下のURLを参照してみてください。

https://www.cdata.com/jp/kb/tech/jira-jdbc-dbv.rst

以下のように「SELECT * FROM Issues」のSQLを実行することで、JIRA API経由でデータを取得できます。

またWhere句も「SELECT *,Updated as TrackingColumn FROM Issues WHERE Updated >= '2020-07-16 11:06:00.224000';」といった形で使用できます。

このクエリを元にLogstashのデータ処理定義である「logstash.conf」ファイルを以下のように作成しました。InputはJDBC、OutputはElasticsearchになっていて、30秒ごとに最新のデータを取得する設定にしています。

input {
  jdbc {
    jdbc_driver_library => "../logstash-core/lib/jars/cdata.jdbc.jira.jar"
    jdbc_driver_class => "Java::cdata.jdbc.jira.JIRADriver"
    jdbc_connection_string => "jdbc:jira:Url=https://XXXXX.atlassian.net;APIToken=XXXXX;[email protected];"
    jdbc_user => ""
    jdbc_password => ""
    schedule => "*/30 * * * * *"
    statement => "SELECT * FROM Issues WHERE Updated >= :sql_last_value"
    last_run_metadata_path => "../.logstash_jdbc_last_run"
  }
}

filter {
  mutate {
    copy => { "id" => "[@metadata][_id]"}
    remove_field => ["id"]
  }
}

output {
  Elasticsearch {
      index => "jira_issues"
      document_id => "%{[@metadata][_id]}"
  }
}

いくつかのパラメータについて解説しますと

• jdbc_driver_library：対象となるJDBC jarファイルを指定します。先の手順で「/logstash-core/lib/jars/」に配置した「cdata.jdbc.jira.jar」を指定してください。
• jdbc_driver_class：JDBC Driverのクラス名を指定します。ここでは「Java::cdata.jdbc.jira.JIRADriver」と指定します。Driver Class 名だけでなく、頭に「Java::」が必要です。
• jdbc_connection_string：JIRA Driverに接続するための各種認証情報を接続文字列として入力します。先の手順で取得したAPI Tokenと接続先のJIRA URL、ログインユーザーIDを指定してください。
• jdbc_user & jdbc_password：CData JDBC Driverでは利用しませんが、LogstashのConfigファイルの仕様上必須なので、空文字を指定します。
• schedule：CRON式でデータの取得間隔を指定します。今回は30秒ごととしました。
• statement：データを取得するためのSQLを指定します。「:sql_last_value」には前回の実行時間が入力されるので、それを元に差分だけ取得するSQLにしています。
• last_run_metadata_path：前回の実行時間が入力されるファイルの場所を指定します。「:sql_last_value」で使用される値です。

Class名や接続文字列などは JIRA Driver のヘルプが参考になります。

「filter」では取得したJIRAのIdをMetadataフィールドに格納しています。

「output 」ではElasticsearchに転送する先となるIndex名と予め「filter」でコピーしておいたIdをElasticsearchのdocument_idとして指定しました。

以上で「logstash.conf」ファイルの作成は完了です。

Logstashの実行

それでは作成した「logstash.conf」ファイルを元にLogstashを実行してみます。

> logstash-7.8.0\bin\logstash -f logstash.conf

実行後、以下のようなログが出ていれば、処理が実行されています。

[2020-07-16T22:25:08,593][INFO ][logstash.inputs.jdbc     ][main][56098d6447be2ed079765ec6c2c7b73d0586b7be4c608fdf52c99ca0c49b883d] (6.786616s) SELECT *,Updated as TrackingColumn FROM Issues WHERE Updated >= '1970-01-01 00:00:00.000000' ORDER BY Updated ASC
[2020-07-16T22:25:32,300][INFO ][logstash.inputs.jdbc     ][main][56098d6447be2ed079765ec6c2c7b73d0586b7be4c608fdf52c99ca0c49b883d] (1.887040s) SELECT *,Updated as TrackingColumn FROM Issues WHERE Updated >= '2020-07-16 13:25:01.173000' ORDER BY Updated ASC

一番最初のクエリはデータ全体を取得するためにデフォルト日付「1970-01-01」以上で取得していますが、2回め以降は30秒刻みで最新のデータを取得するクエリになります。

Kibanaに移動して、実際に転送されたデータも見てみましょう。

DevToolsで以下のクエリを実行してみました。

GET jira_issues/_search
{
  "query": {
    "match_all": {}
  }
}

全部で344件、全部データが格納されていました！

ちゃんと以下のJIRAのIssueが取り込まれていることがわかります。

おわりに

とても手軽に ElasticSeach へ JIRA のデータ取り込むことができました。

CData では JIRA 以外にも数多くの JDBC Driver を提供しています。

kintone や Sansan といった日本のクラウドサービスや Salesforce、Gmail などKBやナレッジが非定形で蓄積されているようなデータソースにも対応しているので、エンタープライズ検索のデータとして活用できるのではないかなと思います。

https://www.cdata.com/jp/jdbc/

すべて30日間のトライアルがあるので、是非一度試してみてください。

参考記事

https://www.elastic.co/jp/blog/how-to-keep-Elasticsearch-synchronized-with-a-relational-database-using-logstash

関連コンテンツ