ポイント: このガイドでは、Apache Sparkを使用してPythonでElasticSearchドキュメントを更新する方法を詳述しています。簡単な例と必要な前提条件を提供し、Python用のElasticsearch-Hadoopコネクタの使用を強調しています。ガイドには、GitHubで利用可能なコードスニペットが含まれており、ドキュメントIDの取得や更新操作などの重要な概念を説明しています。
- ElasticSearchドキュメントを更新するための例コードがGitHubで利用可能です。
- 前提条件にはElasticSearch、Apache Spark、Hadoop-ElasticSearch jarファイルが含まれます。
ここでは、Apache Spark DataframeおよびRDDからElasticSearchドキュメントを更新するためのPythonコードを書く方法を説明します。
インターネット上にはいくつかの指示があります。ElasticSearchによって書かれたものは理解が難しく、例も提供されていません。ここでは、できるだけシンプルな例を作成します。
このコードは、ElasticSearch（ES）JSONドキュメントに追加のフィールドを追加します。つまり、ドキュメントを更新します。SparkはScalaとJavaのためのネイティブサポートを提供しています。しかし、Pythonの場合はElasticSearchによって書かれたElasticsearch-Hadoopコネクタを使用する必要があります。それにより、この操作はより複雑になります。
GitHubのコード
この演習のコードはここにあります：
前提条件
- ES。バイナリをダウンロードし、apt-get installを使用しないでください。そこに保存されているバージョンは非常に古いです。
- Apache Spark。
- Hadoop-ElasticSearch jarファイル。ここからダウンロードすると、さまざまな言語用のjarsが提供されます。
データを追加
最初に、ESに2つのデータレコードを追加する必要があります。キーは表記です：school/doc/1
これは、インデックスschoolに、タイプdoc、ID1を追加することを意味します。
curl -XPUT --header 'Content-Type: application/json' http://localhost:9200/school/doc/1 -d '{
"school" : "Clemson"
}'
curl -XPUT --header 'Content-Type: application/json' http://localhost:9200/schools/doc/2 -d '{
"school" : "Harvard"
}'
以下のPythonコードをspark-submitジョブとして実行するためのコードです。pysparkでコードを1行ずつステップ実行するためには必要ありません。
#!/bin/bash
NUM_CORES=*
DRIVER_MEM=3g
JARS="/usr/share/spark/spark-2.3.2-bin-hadoop2.7/jars/elasticsearch-hadoop-6.4.2/dist"
export SPARK_HOME="/usr/share/spark/spark-2.3.2-bin-hadoop2.7"
export PATH=$PATH:$SPARK_HOME/bin
export CODEPATH="/home/ubuntu/Documents/esearch"
cd $CODEPATH
$SPARK_HOME/bin/spark-submit --master local[$NUM_CORES] --driver-memory
$DRIVER_MEM --jars $JARS/elasticsearch-hadoop-6.4.2.jar
$CODEPATH/updateData.py
コードの説明
read.formatはESへの接続を開きます。重要な点は"es.read.metadata", "true"です。メタデータが必要です。これにより、更新操作に必要なドキュメントの_idが提供されます。
reader = spark.read.format("org.elasticsearch.spark.sql").option("es.read.metadata", "true").option("es.nodes.wan.only","true").option("es.port","9200").option("es.net.ssl","false").option("es.nodes", "http://localhost")
次に、インデックスschool内のすべてのドキュメントを読み取ります。
df = reader.load("school")
df.show()
次に、Harvardと等しい1つのレコードを見つけるためにフィルタリングします。
df.filter(df["school"] == "Harvard").show()
今やdf.filter()操作はその場で動作します。つまり、新しいデータフレームを返さずに、現在のデータフレーム上で操作します。以下のように、フィルタ操作の後にHarvardのみが表示されます。
df.show()
+---------+-------+--------------------+
| school| _metadata|
+---------+-------+--------------------+
| Harvard|[_index -> school...|
| Clemson|[_index -> school...|
+---------+-------+--------------------+
df.filter(df["school"] == "Harvard").show()
+---------+-------+--------------------+
| school| _metadata|
+---------+-------+--------------------+
| Harvard|[_index -> school...|
+---------+-------+--------------------+
メタデータから_idを取得する必要があります。これはマップです。したがって、これをRDDに変換し、collect()メソッドを使用してRowオブジェクトを作成します。次に、最初の行、2番目の列をインデックスで読み取り、名前で_idを取得します。その後、df.withColumnとlitを使用して、その値をデータフレームdf内の定数値として新しい列として書き込みます。
注意: データフレームには1行のみがあります。もしそれがもっとあれば、以下のコード全体でマップ操作を実行して、データフレーム内のすべてのレコードを更新する必要があります。
r=df.rdd.collect()
id = r[0][1]['_id']
df2=df.withColumn("_id", lit(id))
次に、辞書としてesconfオブジェクトを作成します。インターネット上のコード例では、これをJSONとして見ることができます。しかし、以下の表記はよりクリーンで作業しやすいです。
注目すべき重要な項目は：
| es.update.script.inline | ctx._source.locationはlocationというフィールドを更新または作成することを意味します。ctx_sourceはそれを行うためのESオブジェクトです。 |
| es.update.script.params | location:は、インラインスクリプトes.update.script.inlineに渡されるパラメータ値です。<>はリテラルを書くことを意味します。フィールド値を書く場合は、これを省略します。 |
| es.mapping.id | これはESに対してデータフレームのid列を見て、それをドキュメントID_idとして使用するように指示します。ESはそれを使用して、ESで更新したいドキュメントを見つけます。 |
| es.write.operation | upsertは、文書が存在しない場合は追加し、そうでなければ更新することを意味します。updateはそれを更新することを意味します。 |
esconf={}
esconf["es.mapping.id"] = "_id"
esconf["es.nodes"] = "localhost"
esconf["es.port"] = "9200"
esconf["es.update.script.inline"] = "ctx._source.location = params.location"
esconf["es.update.script.params"] = "location:"
esconf["es.write.operation"] = "upsert"
これによりデータが書き込まれます。**esconfは辞書escconfを読み取ることを意味します。mode("append")は、既存のドキュメントにフィールドを追加することを意味します。
df2.write.format("org.elasticsearch.spark.sql").options(**esconf).mode("append").save("school/info")
今、ドキュメントをルックアップして、locationフィールドがCambridgeに更新されたことに気付きます。アイビーリーグのスノッブの群れ。
curl -X GET 'http://localhost:9chool/info/_search'
{"took":1,"timed_out":false,"_shards":{"total":5,"successful":5,"skipped":0,"failed":0},"hits":{"total":1,"max_score":1.0,"hits":[{"_index":"school","_type":"info","_id":"2","_score":1.0,"_source":{"school":"Harvard","location":"Cambridge"}}]}}
完全なコード
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
from pyspark.sql.functions import lit
conf = SparkConf().setAppName("updateSchools")
sc = SparkContext(conf=conf)
sc.setLogLevel("INFO")
spark = SQLContext(sc)
reader = spark.read.format("org.elasticsearch.spark.sql").option("es.read.metadata", "true").option("es.nodes.wan.only","true").option("es.port","9200").option("es.net.ssl","false").option("es.nodes", "http://localhost")
df = reader.load("school")
df.show()
df.filter(df["school"] == "Harvard").show()
r=df.rdd.collect()
id = r[0][1]['_id']
df2=df.withColumn("_id", lit(id))
esconf={}
esconf["es.mapping.id"] = "_id"
esconf["es.nodes"] = "localhost"
esconf["es.port"] = "9200"
esconf["es.update.script.inline"] = "ctx._source.location = params.location"
esconf["es.update.script.params"] = "location:"
esconf["es.write.operation"] = "upsert"
df2.write.format("org.elasticsearch.spark.sql").options(**esconf).mode("append").save("school/info")
ElasticSearchガイド
- 初心者のためのElasticSearchチュートリアル：ElasticSearchの基本
- ElasticSearchとApache Hiveの使用
- ElasticSearchとApache Sparkの使用
- Curl ElasticSearchコマンドチートシート
- Spark ElasticSearch Hadoop 更新とアップサートの例と説明
- 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の使用
- Kibanaを使用してElasticSearchでLuceneおよびKibanaクエリ言語を実行する
- ElasticSearch機械学習
- Elastic Enterprise Searchをインストールする方法
- GitHubとElastic Enterprise Searchを使用する方法
これらの投稿は私自身の意見であり、必ずしもBMCの立場、戦略、または意見を表しているわけではありません。