Блог

Я работаю над переносом данных из elasticsearch в hdfs. Размер данных составляет около 200 ГБ, а объем данных — 80 миллионов.

вот мой код. очень просто. просто прочитайте es, а затем запишите в hdfs.

 // 1. config setting
public Map<String, String> esConfigParam() {
        return new ImmutableMap.Builder<String, String>()
                .put("es.nodes", params.getEsNodes())
                .put("es.mapping.date.rich", "false")
                .put("pushdown", "true")
                .put("es.scroll.size", params.getScrollSize())
                .put("es.read.field.as.array.include","label")
                .put("es.input.json", "true")
                .put("es.read.metadata", "true").build();
    }

// 2. load ES data
Dataset<Row> dataSet = JavaEsSparkSQL.esDF(session.getSqlContext(), indexAlias, esConfigParam());

// 3. write to hdfs
dataSet.write()
            .mode(SaveMode.Overwrite)
            .option("compression", "gzip")
            .parquet(params.getWritePath());
  

Я рассматриваю приведенное ниже как точку настройки для повышения производительности.

1. Настройка Spark: исполнитель-ядер 5 / количество исполнителей 16 / исполнитель-памяти 4g / драйвер-памяти 4g
2. Настройка чтения ES : params.getScrollSize()=2000

В такой ситуации это занимает около 30 минут. Пожалуйста, скажите мне, как сократить время, улучшив производительность записи. (например. применить colesce (10) ??)

Очень интересный вопрос!

В рамках обсуждения я предполагаю, что в вашем кластере 4 узла, по 16 ядер на каждый узел и по 64 ГБ на каждый узел.

(для вашего конкретного случая вам необходимо знать эту информацию и применять приведенный ниже пример)

Вы не можете использовать все ресурсы своего кластера по трем причинам:

• Каждому узлу требуется, по крайней мере, 1 ядро и 1 ГБ для запуска процессов OS и YARN.
• Возможно, другие приложения могут одновременно использовать кластер.
• Мы предполагаем, что AM запущен в одном из узлов.

Что ж, после этого у вас действительно есть 4 узла, 15 ядер на узел и 63 ГБ на узел, свободных для использования.

Вы можете подумать, что хорошей идеей должно быть взять —num-executors 4 —executor-core 15 —executor-memory 63G, но нет!

Причины:

• Сначала вы должны учитывать нагрузку на память (примерно 7% от памяти исполнителя), то есть 63 ГБ 7% = 67,41 > 64 ГБ
• Во-вторых, вы будете использовать все ядра в узлах, но вам понадобится 1 дополнительное ядро в одном из них для запуска AM (диспетчер приложений)
• И, наконец, 15 ядер на исполнителя могут привести к снижению пропускной способности ввода-вывода HDFS.

Хорошим подходом было бы —число исполнителей 11 —количество ядер исполнителя 4 —объем памяти исполнителя 19 ГБ:

• Узел 1,2,3: будет использовать 3 исполнителя и 12 ядер на каждом узле (3 ядра свободны для SO и других процессов)
• Узел 4: будет использовать 2 исполнителя и 8 ядер (остальные 8 ядер свободны для AM, SO и других процессов)
• Каждый исполнитель будет использовать 19 ГБ 7% (накладные расходы) = 20,33 ГБ
• Узел 1,2,3: будет использовать 20,33 * 3 исполнителя = 60,99 ГБ (3 ГБ бесплатно)
• Узел 4: будет использовать 40,66 ГБ (23,44 ГБ бесплатно для AM, SO и других процессов)

Это не единственная конфигурация, которую вы можете использовать, есть и другие.

В результате настройка spark всегда является сложной задачей. Вы должны знать ресурсы своего кластера (память, узлы и ядра).

Дополнительная информация в официальном блоге cloudera:https://blog.cloudera.com/how-to-tune-your-apache-spark-jobs-part-2 /