ポイント: Apache SparkデータをPythonを使用してElasticSearchに書き込むための指示が提供されており、ユーザーのためにプロセスが簡素化されています。このガイドには、ES-HadoopのダウンロードやPySparkの設定、Apacheログファイルの解析やElasticSearchにデータを保存するためのコードスニペットなどが含まれています。
- 互換性のためにES-Hadoopをダウンロードします。
- ElasticSearch形式でデータを保存するための特定の設定を使用します。
ここでは、Pythonを使用してApache SparkデータをElasticSearch（ES）に書き込む方法を説明します。ApacheログデータをESに書き込みます。
このトピックは、インターネット上のすべての悪い、複雑な例が原因で複雑になっています。しかし、ここでは簡単にします。
1つの複雑な要因は、SparkがScalaおよびJavaでElasticSearchへの書き込みをネイティブにサポートしているのに対し、Pythonではサポートしていないことです。そのため、ElasticSearchによって書かれたES-Hadoopをダウンロードする必要があります。こちらで入手できますここ。
次に、それをスコープに持ち込んで、次のようにpysparkで利用可能にします:
pyspark --jars elasticsearch-hadoop-6.4.1.jar
PySparkを次のようにPython 3を使用するように設定します:
export PYSPARK_PYTHON=/usr/bin/python3
ElasticSearchへの書き込みを理解するための鍵は、ESがJSONデータベースである一方、1つの要件があることです。データはこの形式である必要があります:
{ "id: { the rest of your json}}
以下にその変換を行う方法を示します。
下部には完全なコードがあり、オンラインではこちらで確認できます。ここではセクションごとに説明します:
Apacheログファイルの解析
ApacheログをSpark RDDに読み込みます。次に、各文字列を正規表現グループに読み込むparse()関数を作成し、必要なフィールドを選択して辞書として返します。
rdd = sc.textFile("/home/ubuntu/walker/apache_logs")
regex='^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$'
p=re.compile(regex)
def parse(str):
s=p.match(str)
d = {}
d['ip']=s.group(1)
d['date']=s.group(4)
d['operation']=s.group(5)
d['uri']=s.group(6)
return d
言い換えれば、最初にテキストファイルログをRDDに読み込むと、次のようになります:
['83.149.9.216 - - [17/May/2015:10:05:03 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"']
その後、rddのmap()関数を使用して、各行をparse()関数に渡してこれを生成します。
rdd2 = rdd.map(parse)
rdd2.take(1)
[{'date': '17/May/2015:10:05:03 +0000', 'ip': '83.149.9.216', 'operation': 'GET', 'uri': '/presentations/logstash-monitorama-2013/images/kibana-search.png'}]
今、これはJSONのように見えますが、まだJSONではありません。次に、json.dumpsを使用します。Python文書の技術的な説明によれば、これは「objをJSON形式のストリームとしてシリアル化する」となっています。
また、IDを追加します。以下のES設定で、ESにどのフィールドがユニークなドキュメント識別子になるかを伝えます: "es.mapping.id": "doc_id".
最初にSHAダイジェストを計算して、ユニークな数値としてそのIDを作成します。
結果は次のように返されます。IDは非常に長いSHA番号で、前にJSONが続きます。
rdd3.take(1)
[('a5b086b04e1cc45fb4a19e2a641bf99ea3a378599ef62ba12563b75c', '{"date": "17/May/2015:10:05:03 +0000", "ip": "83.149.9.216", "operation": "GET", "doc_id": "a5b086b04e1cc45fb4a19e2a641bf99ea3a378599ef62ba12563b75c", "uri": "/presentations/logstash-monitorama-2013/images/kibana-search.png"}')]
次に、ElasticSearch設定を指定します。重要な項目は次のとおりです:
| "es.resource" : 'walker/apache' | "walker"はインデックスで、"apache"はタイプです。全体を合わせて「インデックス」と呼ばれることがよくあります。 |
| "es.mapping.id": "doc_id" | ここでは、ESにどのドキュメントをドキュメントIDとして使用するかを示しています。これは_idフィールドと同じことを意味します。 |
残りのフィールドは自己説明的です。
次に、saveAsNewAPIHadoopFile()メソッドを使用してRDDをESに保存します。文法は常にESに対して同じであるため、すべての部分を理解する必要はありません。
es_write_conf = {
"es.nodes" : "localhost",
"es.port" : "9200",
"es.resource" : 'walker/apache',
"es.input.json": "yes",
"es.mapping.id": "doc_id"
}
rdd3.saveAsNewAPIHadoopFile(
path='-',
outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat", keyClass="org.apache.hadoop.io.NullWritable",
valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
conf=es_write_conf)
rdd3 = rdd2.map(addID)
def addId(data):
j=json.dumps(data).encode('ascii', 'ignore')
data['doc_id'] = hashlib.sha224(j).hexdigest()
return (data['doc_id'], json.dumps(data))
次に、コマンドラインからESにクエリを実行し、1つのドキュメントを確認できます:
curl http://localhost:9200s/walker/apache/_search?pretty=true&?q=*
{
"_index" : "walker",
"_type" : "apache",
"_id" : "227e977849bfd5f8d1fca69b04f7a766560745c6cb3712c106d590c2",
"_score" : 1.0,
"_source" : {
"date" : "17/May/2015:10:05:32 +0000",
"ip" : "91.177.205.119",
"operation" : "GET",
"doc_id" : "227e977849bfd5f8d1fca69b04f7a766560745c6cb3712c106d590c2",
"uri" : "/favicon.ico"
}
こちらが完全なコードです:
import json
import hashlib
import re
def addId(data):
j=json.dumps(data).encode('ascii', 'ignore')
data['doc_id'] = hashlib.sha224(j).hexdigest()
return (data['doc_id'], json.dumps(data))
def parse(str):
s=p.match(str)
d = {}
d['ip']=s.group(1)
d['date']=s.group(4)
d['operation']=s.group(5)
d['uri']=s.group(6)
return d
regex='^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$'
p=re.compile(regex)
rdd = sc.textFile("/home/ubuntu/walker/apache_logs")
rdd2 = rdd.map(parse)
rdd3 = rdd2.map(addID)
es_write_conf = {
"es.nodes" : "localhost",
"es.port" : "9200",
"es.resource" : 'walker/apache',
"es.input.json": "yes",
"es.mapping.id": "doc_id"
}
rdd3.saveAsNewAPIHadoopFile(
path='-',
outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat", keyClass="org.apache.hadoop.io.NullWritable",
valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
conf=es_write_conf)
ElasticSearchガイド
- 初心者のためのElasticSearchチュートリアル: ElasticSearchの基本
- ElasticSearchでApache Hiveを使用する
- Apache SparkでElasticSearchを使用する
- Curl ElasticSearchコマンドチートシート
- Spark ElasticSearch Hadoopの更新およびUpsertの例と説明
- Apache SparkデータをElasticSearchにPythonで書き込む方法
- ElasticSearch検索構文とブーリアンおよび集約検索
- ElasticSearchネストされたクエリ: 埋め込まれたドキュメントを検索する方法
- ElasticSearch結合: Has_Child、Has_parentクエリ
- Elastic Version 7クラスタのセットアップ方法
- nginxとElasticSearchのためのFilebeatの設定方法
- BeatsとLogstashを使用してElasticSearchにログを送信する
- CSVファイルをLogstashを使ってElasticSearchにロードする方法
- ElasticSearchの集約について説明
- ELKスタックとは何ですか？
- Logstash 101: データ処理パイプラインでのLogstashの使用
- Kibanaを使用してElasticSearchでLuceneおよびKibanaクエリ言語を実行する
- ElasticSearch機械学習
- Elastic Enterprise Searchのインストール方法
- GitHubでElastic Enterprise Searchを使用する方法
これらの投稿は私自身の見解であり、BMCの立場、戦略、または意見を必ずしも表すものではありません。