Apache Sparkを効率的に使用して、データをelasticsearchにプッシュします
xmlファイルに2700万件のレコードがあり、elasticsearchインデックスにプッシュしたい以下はspark scalaで記述されたコードスニペットです。sparkジョブjarを作成し、AWSEMRで実行します
どうすればスパークを効率的に使用してこの演習を完了することができますか？ご案内ください。
Sparkデータフレームにロードしている12.5GBのgzip圧縮されたxmlがあります。私はSparkを初めて使用します。（このgzipファイルを分割する必要がありますか？それともsparkエグゼキュータが処理しますか？）
class ReadFromXML {
def createXMLDF(): DataFrame = {
val spark: SparkSession = SparkUtils.getSparkInstance("Spark Extractor")
import spark.implicits._
val m_df: DataFrame = SparkUtils.getDataFrame(spark, "temp.xml.gz").coalesce(5)
var new_df: DataFrame = null
new_df = m_df.select($"CountryCode"(0).as("countryCode"), $"PostalCode"(0).as("postalCode"),
$"state"(0).as("state"), $"county"(0).as("county"),
$"city"(0).as("city"), $"district"(0).as("district"),
$"Identity.PlaceId".as("placeid"), $"Identity._isDeleted".as("deleted"),
$"FullStreetName"(0).as("street"), functions.explode($"Text").as("name"), $"name".getField("BaseText").getField("_VALUE")(0).as("nameVal")) .where($"LocationList.Location._primary" === "true")
.where("(array_contains(_languageCode, 'en'))")
.where(functions.array_contains($"name".getField("BaseText").getField("_languageCode"), "en"))
new_df.drop("name")
}
}
object PushToES extends App {
val spark = SparkSession
.builder()
.appName("PushToES")
.master("local[*]")
.config("spark.es.nodes", "awsurl")
.config("spark.es.port", "port")
.config("spark.es.nodes.wan.only", "true")
.config("spark.es.net.ssl", "true")
.getOrCreate()
val extractor = new ReadFromXML()
val df = extractor.createXMLDF()
df.saveToEs("myindex/_doc")
}
更新1：ファイルをそれぞれ68Mに分割し、この1つのファイルを読み取るのに3.7分かかります。gzip圧縮コーデックの代わりにsnappyを使用しようとしていませんでした。gzファイルをsnappyファイルに変換し、以下の構成に追加しました。
.config("spark.io.compression.codec", "org.apache.spark.io.SnappyCompressionCodec")
しかし、それは空のデータフレームを返します
df.printschemaは「ルート」だけを返します
更新2： lzo形式で実行できました。データフレームの解凍と読み込みにかかる時間は非常に短くなりました。
サイズ140MBの各lzo圧縮ファイルを繰り返し処理してデータフレームを作成することをお勧めしますか？または
データフレームに10個のファイルのセットをロードする必要がありますか？または
1つのデータフレームにそれぞれ140MBの200lzo圧縮ファイルをすべてロードする必要がありますか？はいの場合、マスターにロードされると思うので、マスターにどのくらいのメモリを割り当てる必要がありますか？
s3バケットからファイルを読み取るときに、「s3a」uriはパフォーマンスを向上させることができますか？または「s3」URIはEMRで大丈夫ですか？
更新3：10個のlzoファイルの小さなセットをテストするには..以下の構成を使用しました。EMRクラスターは全体で56分かかり、そのステップ（Sparkアプリケーション）は10個のファイルを処理するのに48分かかりました。
1マスター-m5.xlarge4 vCore、16 GiBメモリ、EBSのみのストレージEBSストレージ：32GiB
2コア-m5.xlarge4 vCore、16 GiBメモリ、EBSのみのストレージEBSストレージ：32GiB
以下から学習したSparkで調整されたパラメータ https://idk.dev/best-practices-for-successfully-managing-memory-for-apache-spark-applications-on-amazon-emr/
[
{
"Classification": "yarn-site",
"Properties": {
"yarn.nodemanager.vmem-check-enabled": "false",
"yarn.nodemanager.pmem-check-enabled": "false"
}
},
{
"Classification": "spark",
"Properties": {
"maximizeResourceAllocation": "false"
}
},
{
"Classification": "spark-defaults",
"Properties": {
"spark.network.timeout": "800s",
"spark.executor.heartbeatInterval": "60s",
"spark.dynamicAllocation.enabled": "false",
"spark.driver.memory": "10800M",
"spark.executor.memory": "10800M",
"spark.executor.cores": "2",
"spark.executor.memoryOverhead": "1200M",
"spark.driver.memoryOverhead": "1200M",
"spark.memory.fraction": "0.80",
"spark.memory.storageFraction": "0.30",
"spark.executor.extraJavaOptions": "-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'",
"spark.driver.extraJavaOptions": "-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'",
"spark.yarn.scheduler.reporterThread.maxFailures": "5",
"spark.storage.level": "MEMORY_AND_DISK_SER",
"spark.rdd.compress": "true",
"spark.shuffle.compress": "true",
"spark.shuffle.spill.compress": "true",
"spark.default.parallelism": "4"
}
},
{
"Classification": "mapred-site",
"Properties": {
"mapreduce.map.output.compress": "true"
}
}
]
回答
これが私の側からのヒントのいくつかです。
寄木細工の形式または任意の形式でデータを読み取ります。必要に応じて再パーティション化します。データ変換には時間がかかる場合があるため、Sparkで読み取り、処理してください。ロードを開始する前に、マップとフォーマットのデータを作成してみてください。これは、複雑なマップの場合に簡単にデバッグするのに役立ちます。
val spark = SparkSession
.builder()
.appName("PushToES")
.enableHiveSupport()
.getOrCreate()
val batchSizeInMB=4; // change it as you need
val batchRetryCount= 3
val batchWriteRetryWait = 10
val batchEntries= 10
val enableSSL = true
val wanOnly = true
val enableIdempotentInserts = true
val esNodes = [yourNode1, yourNode2, yourNode3]
var esConfig = Map[String, String]()
esConfig = esConfig + ("es.node"-> esNodes.mkString)(","))
esConfig = esConfig + ("es.port"->port.toString())
esConfig = esConfig + ("es.batch.size.bytes"->(batchSizeInMB*1024*1024).toString())
esConfig = esConfig + ("es.batch.size.entries"->batchEntries.toString())
esConfig = esConfig + ("es.batch.write.retry.count"->batchRetryCount.toString())
esConfig = esConfig + ("es.batch.write.retry.wait"->batchWriteRetryWait.toString())
esConfig = esConfig + ("es.batch.write.refresh"->"false")
if(enableSSL){
esConfig = esConfig + ("es.net.ssl"->"true")
esConfig = esConfig + ("es.net.ssl.keystore.location"->"identity.jks")
esConfig = esConfig + ("es.net.ssl.cert.allow.self.signed"->"true")
}
if (wanOnly){
esConfig = esConfig + ("es.nodes.wan.only"->"true")
}
// This helps if some task fails , so data won't be dublicate
if(enableIdempotentInserts){
esConfig = esConfig + ("es.mapping.id" ->"your_primary_key_column")
}
val df = "suppose you created it using parquet format or any format"
実際には、データはエグゼキュータレベルで挿入され、ドライバレベルでは挿入されないため、同時に開いている接続がそれほど多くないように、各エグゼキュータに2〜4コアのみを割り当ててみてください。使いやすさに応じて、ドキュメントのサイズやエントリを変えることができます。それらについてお読みください。
データをチャンクで書き込むと、将来的に大きなデータセットをロードするのに役立ち、データをロードする前にインデックスマップを作成してみてください。また、ESにはその機能があるため、ネストされたデータはほとんどありません。つまり、データに主キーを保持するようにしてください。
val dfToInsert = df.withColumn("salt", ceil(rand())*10).cast("Int").persist()
for (i<-0 to 10){
val start = System.currentTimeMillis
val finalDF = dfToInsert.filter($"salt"===i) val counts = finalDF.count() println(s"count of record in chunk $i -> $counts") finalDF.drop("salt").saveToES("indexName",esConfig) val totalTime = System.currentTimeMillis - start println(s"ended Loading data for chunk $i. Total time taken in Seconds : ${totalTime/1000}")
}
最終的なDFにエイリアスを付けて、実行ごとに更新してみてください。ロード時に本番サーバーの邪魔をしたくないので
記憶
これは一般的なものにすることはできません。しかし、あなたにキックスタートを与えるためだけに
データサイズまたは予算に応じて、10〜40のエグゼキュータを維持します。各エグゼキュータを8〜16GBのサイズと5GBのオーバーヘッドに保ちます。（これは、ドキュメントのサイズが大きい場合と小さい場合があるため、異なる場合があります）。必要に応じて、maxResultSizeを8GBに保ちます。ドライバーは5コアと30グラムのRAMを持つことができます
重要な事柄。
インデックスに従って構成を変更できるため、構成を変数に保持する必要があります
挿入はドライバーではなくエグゼキューターで発生するため、書き込み中は接続を少なくするようにしてください。各コアは1つの接続を開きます。
ドキュメントの挿入は、バッチエントリサイズまたはドキュメントサイズで行うことができます。複数の実行を行いながら、学習に従って変更してください。
ソリューションを堅牢にするようにしてください。すべてのサイズのデータを処理できる必要があります。読み取りと書き込みの両方を調整できますが、ロードを開始する前に、ドキュメントマップに従ってデータをフォーマットしてみてください。これは、データドキュメントが少し複雑でネストされている場合に、簡単にデバッグするのに役立ちます。
Spark-Submitのメモリは、ジョブの実行中に学習したとおりに調整することもできます。メモリとバッチサイズを変えて、挿入時間を調べてみてください。
最も重要なことはデザインです。ESを使用している場合は、終了クエリと要件を念頭に置いてマップを作成してください。
完全な答えではありませんが、コメントには少し長いです。私が提案したいいくつかのヒントがあります。
はっきりしていませんが、心配なのは実行時間だと思います。コメントで示唆されているように、クラスターにノード/エグゼキューターを追加することでパフォーマンスを向上させることができます。gzipファイルがsparkでパーティション化せずにロードされる場合は、適切なサイズに分割する必要があります。（小さすぎない-これにより処理が遅くなります。大きすぎない-エグゼキュータはOOMを実行します）。
parquetSparkを使用する場合に適したファイル形式です。XMLを寄木細工に変換できるかどうか。超圧縮で軽量です。
あなたのコメントを読んでcoalesce、完全なシャッフルをしません。合体アルゴリズムは、一部のパーティションから既存のパーティションにデータを移動することにより、ノードの数を変更します。このアルゴリズムでは、明らかにパーティションの数を増やすことはできません。repartition代わりに使用してください。操作にはコストがかかりますが、パーティションの数が増える可能性があります。その他の事実については、これを確認してください。https://medium.com/@mrpowers/managing-spark-partitions-with-coalesce-and-repartition-4050c57ad5c4