ポイント: ElasticSearchはApache Sparkとシームレスに統合され、JSONデータに対する分析を可能にします。組織はログ処理のためにElasticSearchを利用し、運用およびサイバーセキュリティを強化しています。Sparkのバージョン2.1以降、ユーザーはコネクタを活用して高度なデータ分析を行うことができ、基本的なLuceneクエリを超えた分析が可能です。
- ElasticSearchはログ処理に広く使用されています。
- Apache Sparkはバージョン2.1以降、ElasticSearchのネイティブサポートを提供しています。
ElasticSearchはログ処理システムで人気のあるJSONデータベースです。たとえば、組織は通常、ElasticSearchをlogstashまたはfilebeatと一緒に使用して、Webサーバーログ、Windowsイベント、Linux syslog、その他のデータを送信します。そして、Kibanaウェブインターフェースを使用してログイベントをクエリします。これらすべては、サイバーセキュリティ、運用などにとって重要です。現在、Spark 2.1以降、SparkはElasticSearchのネイティブサポートを含んでおり、これをElasticsearch Hadoopと呼びます。つまり、Apache PigやHiveを使用してElasticSearchのJSONドキュメントと作業できます。ElasticSearch Sparkは2.1以前に存在していたコネクタであり、現在もサポートされています。ここでは、ElasticSearch Sparkの使用方法を示します。これらのコネクタは、ElasticSearchデータに対して分析を実行することを可能にします。ElasticSearch自体はLuceneクエリのみをサポートしており、つまり自然言語クエリです。したがって、サイバーセキュリティイベントをフラグ付けしたり、他の分析を実行したりするために、予測および分類モデルを書くこともできます。これはESが自分だけでは行わないことです。
ElasticSearch Sparkのインストール
まず、ElasticSearchが必要です。Dockerイメージを使用するのではなく、事前構築されたバージョンをダウンロードしてください。ダウンロードしたバージョンは認証がオフになっており、いくつかのステップを省くことができます。
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.5.2.tar.gz
次に、Apache Sparkが必要です。ここからダウンロードし、解凍します。どちらのソフトウェアも構成は不要です。また、使用するためにSparkを起動する必要はありません。Hadoopを起動する必要もありません。ElasticSearchを起動するだけで済みます。非ルートユーザーとして次のコマンドを実行してください:
/usr/share/elasticsearch/elasticsearch-5.5.2/bin/elasticsearch
次に、ElasticSearch jarファイルをダウンロードします:
wget http://central.maven.org/maven2/org/elasticsearch/elasticsearch-spark-20_2.11/5.5.2/elasticsearch-spark-20_2.11-5.5.2.jar
これで、コマンドラインオプションとしてjarファイルを指定して、対話型Sparkシェルを起動します:
spark-shell --jars /home/walker/Documents/jars/elasticsearch-spark-20_2.11-5.5.2.jar
これで、ElasticSearchのウェブサイトからそのままサンプルプログラムをコピーして、ElasticSearchにデータを作成します。このテキストをspark-shellにコピーしてください。
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.elasticsearch.spark._
val sc = new SparkContext(conf)
conf.set("es.index.auto.create", "true")
val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)
val airports = Map("arrival" -> "Otopeni", "SFO" -> "San Fran")
sc.makeRDD(Seq(numbers, airports)).saveToEs("spark/docs")
これにより、ElasticSearchにRDDを書き込むだけです。これで、Scalaの数行を書いて、そのデータをESから読み込み、再度RDDにします。
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.elasticsearch.spark._
import org.apache.spark.SparkConf
val RDD = sc.esRDD("spark/docs")
RDD.first()
これにより、以下の行が返されます。RDDはMapです。また、ScalaのケースクラスやJavaBeanでもかまいません。したがって、ここに保存できるオブジェクトの種類があり、ES形式はJSONであるため、理にかなっています。
RDD: org.apache.spark.rdd.RDD[(String, scala.collection.Map[String,AnyRef])] = ScalaEsRDD[0] at RDD at AbstractEsRDD.scala:37
res4: (String, scala.collection.Map[String,AnyRef]) = (AV5b8VamL-9DBQqpL3kV,Map(arrival -> Otopeni, SFO -> San Fran))
これでESインデックスをクエリして、sparkインデックスが存在することを確認できます:
curl 'localhost:9200/_cat/indices?v'
応答:
health status index uuid pri rep docs.count docs.deleted store.size pri.store.size
yellow open spark n9aTT2HcTJ-RiRZ5Ct1Wiw 5 1 2 0 7.9kb 7.9kb
ElasticSearchガイド
- 初心者のためのElasticSearchチュートリアル: ElasticSearchの基本
- Apache HiveをElasticSearchと一緒に使用する
- Apache Sparkを使用したElasticSearch
- Curl ElasticSearchコマンドチートシート
- Spark ElasticSearch Hadoopの更新とUpsertの例と説明
- Pythonを使用してApache SparkデータをElasticSearchに書き込む方法
- ElasticSearch検索構文とブールおよび集約検索
- ElasticSearchネストされたクエリ: 埋め込まれたドキュメントを検索する方法
- ElasticSearchジョイン: Has_Child, Has_parentクエリ
- Elasticバージョン7クラスタのセットアップ方法
- nginxとElasticSearchのためにFilebeatを構成する方法
- BeatsとLogstashを使用してElasticSearchにログを送信する
- Logstashを使用してCSVファイルをElasticSearchにロードする方法
- ElasticSearchの集約について説明
- ELKスタックとは何ですか？
- Logstash 101: データ処理パイプラインでのLogstashの使用
- LuceneとKibanaクエリ言語を使用してElasticSearchでクエリを実行するためのKibanaの使用
- ElasticSearchの機械学習
- Elastic Enterprise Searchをインストールする方法
- GitHubでElastic Enterprise Searchを使用する方法
これらの投稿は私自身のものであり、BMCの立場、戦略、または意見を必ずしも表すものではありません。