Эффективное использование Apache Spark для передачи данных в elasticsearch
У меня есть 27 миллионов записей в xml-файле, которые я хочу поместить в индекс elasticsearch. Ниже приведен фрагмент кода, написанный на Spark Scala. Я буду создавать jar-файл с искровым заданием и запускать его на AWS EMR.
Как я могу эффективно использовать искру для выполнения этого упражнения? Пожалуйста, направьте.
У меня есть сжатый xml размером 12,5 ГБ, который я загружаю в искровой фреймворк. Я новичок в 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 каждый, и для чтения этого единственного файла требуется 3,7 минуты. Я пытался использовать snappy вместо кодека сжатия gzip. Поэтому преобразовал файл gz в файл snappy и добавил ниже в конфигурации.
.config("spark.io.compression.codec", "org.apache.spark.io.SnappyCompressionCodec")
Но он возвращает пустой фрейм данных
df.printschema возвращает только «корень»
Обновление 2: мне удалось запустить формат lzo ... на распаковку и загрузку в фрейм данных уходит гораздо меньше времени.
Это хорошая идея - перебирать каждый сжатый файл lzo размером 140 МБ и создавать фрейм данных? или же
я должен загрузить набор из 10 файлов в фрейм данных? или же
должен ли я загружать все 200 сжатых файлов lzo размером 140 МБ каждый в один фрейм данных ?. если да, то сколько памяти следует выделить мастеру, поскольку я думаю, что он будет загружен на мастер?
При чтении файла из ведра s3, uri "s3a" может улучшить производительность? или "s3" uri подходит для EMR?
Обновление 3: для тестирования небольшого набора из 10 файлов lzo. Я использовал конфигурацию ниже. Кластеру EMR потребовалось 56 минут, из которых шаг (приложение Spark) занял 48 минут, чтобы обработать 10 файлов.
1 главный - m5.xlarge 4 виртуальных ядра, 16 ГиБ памяти, только хранилище EBS Хранилище EBS: 32 ГиБ
2 ядра - m5.xlarge 4 виртуальных ядра, 16 ГиБ памяти, только хранилище EBS Хранилище EBS: 32 ГиБ
Ниже приведены настроенные параметры 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"
}
}
]
Ответы
Вот несколько советов с моей стороны.
Считайте данные в формате паркета или любом другом формате. Разбейте его на разделы в соответствии с вашими потребностями. Преобразование данных может занять время, поэтому считайте его в искре, а затем обработайте. Перед загрузкой попробуйте создать карту и отформатировать данные. Это поможет облегчить отладку в случае сложной карты.
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-16 ГБ и накладными расходами 5 ГБ. (Это может варьироваться, поскольку ваш документ может быть большим или маленьким по размеру). При необходимости оставьте maxResultSize 8gb. Драйвер может иметь 5 ядер и 30 г RAM
Важные вещи.
Вам необходимо сохранить конфигурацию в переменной, так как вы можете изменить ее в соответствии с индексом
Вставка происходит на исполнителе, а не на драйвере, поэтому старайтесь поддерживать меньшее соединение во время записи. Каждое ядро открывает одно соединение.
вставка документа может быть с размером записи партии или размером документа. Измените его в соответствии с вашим обучением, выполняя несколько пробежек.
Постарайтесь сделать свое решение надежным. Он должен уметь обрабатывать данные любого размера. Чтение и запись можно настроить, но попробуйте отформатировать данные в соответствии с картой документа перед началом загрузки. Это поможет облегчить отладку, если документ данных немного сложен и вложен.
Память искры-отправки также может быть настроена в соответствии с вашим обучением во время выполнения заданий. Просто попробуйте посмотреть время вставки, варьируя объем памяти и размер пакета.
Самое главное - это дизайн. Если вы используете ES, тогда создавайте свою карту, учитывая конечные запросы и требования.
Не полный ответ, но все же немного для комментария. Я хотел бы дать несколько советов.
Это не ясно, но я предполагаю, что вы беспокоитесь о времени выполнения. Как предлагается в комментариях, вы можете улучшить производительность, добавив в кластер больше узлов / исполнителей. Если файл gzip загружается без разбиения в Spark, вам следует разделить его до разумного размера. (Не слишком маленький - это замедлит обработку. Не слишком большой - исполнители будут запускать OOM).
parquet
- хороший формат файла для работы со Spark. Если вы можете преобразовать ваш XML в паркет. Он очень сжатый и легкий.
Читая ваши комментарии, coalesce
не тасует их полностью. Алгоритм объединения изменяет количество узлов, перемещая данные из некоторых разделов в существующие разделы. Очевидно, что этот алгоритм не может увеличить количество разделов. repartition
Вместо этого используйте . Операция дорогостоящая, но может увеличить количество разделов. Проверьте это, чтобы узнать больше:https://medium.com/@mrpowers/managing-spark-partitions-with-coalesce-and-repartition-4050c57ad5c4