Gunakan Apache Spark secara efisien untuk mendorong data ke elasticsearch
Saya memiliki 27 juta catatan dalam file xml, yang ingin saya masukkan ke dalam indeks elasticsearch Di bawah ini adalah cuplikan kode yang ditulis dalam spark scala, saya akan membuat stoples pekerjaan percikan dan akan berjalan di AWS EMR
Bagaimana saya dapat menggunakan percikan api secara efisien untuk menyelesaikan latihan ini? Tolong bimbing.
Saya memiliki xml gzip 12,5 gb yang saya muat ke dalam bingkai data percikan. Saya baru mengenal Spark .. (Haruskah saya membagi file gzip ini? Atau pelaksana spark akan mengurusnya?)
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")
}
Pembaruan 1: Saya telah memecah file masing-masing dalam 68M dan untuk membaca file tunggal ini dibutuhkan 3,7 menit Saya membuang waktu mencoba menggunakan snappy alih-alih codec kompresi gzip Jadi ubah file gz menjadi file snappy dan tambahkan di bawah ini dalam konfigurasi
.config("spark.io.compression.codec", "org.apache.spark.io.SnappyCompressionCodec")
Tapi itu mengembalikan dataframe kosong
df.printschema hanya mengembalikan "root"
Pembaruan 2: Saya telah berhasil menjalankan dengan format lzo .. dibutuhkan waktu yang sangat sedikit untuk mendekompresi dan memuat dalam kerangka data.
Apakah ide yang bagus untuk mengulang setiap file terkompresi lzo berukuran 140 MB dan membuat kerangka data? atau
haruskah saya memuat 10 file dalam kerangka data? atau
haruskah saya memuat semua 200 file terkompresi lzo masing-masing 140MB dalam satu dataframe ?. jika ya maka berapa banyak memori yang harus dialokasikan untuk master karena saya pikir ini akan dimuat pada master?
Saat membaca file dari bucket s3, "s3a" uri dapat meningkatkan performa? atau "s3" uri ok untuk EMR?
Pembaruan 3: Untuk menguji satu set kecil 10 file lzo .. Saya menggunakan konfigurasi di bawah ini. Kluster EMR memerlukan 56 menit secara keseluruhan dari langkah mana (aplikasi Spark) membutuhkan waktu 48 menit untuk memproses 10 file
1 Master - m5.xlarge 4 vCore, memori 16 GiB, penyimpanan khusus EBS Penyimpanan EBS: 32 GiB
2 Core - m5.xlarge 4 vCore, memori 16 GiB, penyimpanan khusus EBS Penyimpanan EBS: 32 GiB
Dengan parameter yang disetel Spark di bawah ini dipelajari dari 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"
}
}
]
Jawaban
Berikut beberapa tips dari sisi saya.
Membaca data dalam format parket atau format apapun. Partisi ulang sesuai kebutuhan Anda. Konversi data dapat memakan waktu, jadi bacalah dalam percikan dan kemudian proses. Cobalah untuk membuat peta dan memformat data sebelum mulai memuat. Ini akan membantu proses debug yang mudah dalam kasus peta yang kompleks.
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"
Sebenarnya data dimasukkan pada level pelaksana dan bukan pada level driver coba berikan hanya 2-4 core untuk setiap pelaksana agar tidak banyak koneksi yang terbuka pada saat yang bersamaan. Anda dapat memvariasikan ukuran atau entri dokumen sesuai kemudahan Anda. Silakan baca tentang mereka.
menulis data dalam potongan ini akan membantu Anda dalam memuat dataset besar di masa depan dan mencoba membuat peta indeks sebelum memuat data. Dan lebih suka sedikit data bersarang karena Anda memiliki fungsionalitas itu di ES Maksud saya mencoba menyimpan beberapa kunci utama dalam data Anda.
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}")
}
Cobalah untuk memberikan beberapa alias ke DF terakhir Anda dan perbarui itu di setiap proses. Karena Anda tidak ingin mengganggu server produksi Anda pada saat memuat
Penyimpanan
Ini tidak bisa generik. Tapi hanya sebagai permulaan
pertahankan 10-40 pelaksana sesuai ukuran data atau anggaran Anda. pertahankan ukuran masing-masing pelaksana 8-16gb dan overhead 5 gb. (Ini dapat bervariasi karena dokumen Anda bisa berukuran besar atau kecil). Jika perlu, pertahankan maxResultSize 8gb. Pengemudi dapat memiliki 5 inti dan 30 g ram
Hal-hal penting.
Anda perlu menyimpan config dalam variabel karena Anda dapat mengubahnya sesuai Indeks
Penyisipan terjadi pada pelaksana bukan pada driver, Jadi cobalah untuk menjaga koneksi yang lebih rendah saat menulis. Setiap inti akan membuka satu koneksi.
penyisipan dokumen bisa dengan ukuran entri batch atau ukuran dokumen. Ubah sesuai pembelajaran Anda saat melakukan banyak lari.
Cobalah untuk membuat solusi Anda kuat. Itu harus bisa menangani semua data ukuran. Membaca dan menulis keduanya dapat disetel tetapi cobalah untuk memformat data Anda sesuai peta dokumen sebelum mulai memuat. Ini akan membantu dalam proses debug yang mudah, Jika dokumen data sedikit rumit dan bersarang.
Memori percikan-kirim juga dapat disetel sesuai pembelajaran Anda saat menjalankan pekerjaan. Coba saja lihat waktu penyisipan dengan memvariasikan memori dan ukuran batch.
Yang terpenting adalah desain. Jika Anda menggunakan ES daripada membuat peta Anda sambil tetap mengingat permintaan dan persyaratan akhir.
Bukan jawaban lengkap tapi masih agak panjang untuk berkomentar. Ada beberapa tips yang ingin saya sarankan.
Tidak jelas tapi saya anggap kekhawatiran Anda dengar adalah waktu eksekusi. Seperti yang disarankan dalam komentar, Anda dapat meningkatkan kinerja dengan menambahkan lebih banyak node / eksekutor ke cluster. Jika file gzip dimuat tanpa mempartisi di spark, maka Anda harus membaginya ke ukuran yang wajar. (Tidak terlalu kecil - Ini akan membuat pemrosesan menjadi lambat. Tidak terlalu besar - pelaksana akan menjalankan OOM).
parquet
adalah format file yang bagus saat bekerja dengan Spark. Jika Anda dapat mengubah XML Anda menjadi parket. Ini super terkompresi dan ringan.
Membaca komentar Anda, coalesce
tidak dilakukan secara acak. Algoritma penggabungan mengubah jumlah node dengan memindahkan data dari beberapa partisi ke partisi yang ada. Algoritma ini jelas tidak bisa menambah jumlah partisi. Gunakan repartition
sebagai gantinya. Pengoperasiannya mahal tetapi dapat meningkatkan jumlah partisi. Lihat ini untuk fakta lebih lanjut:https://medium.com/@mrpowers/managing-spark-partitions-with-coalesce-and-repartition-4050c57ad5c4