Задержка искры Apache между рабочими местами

Question

Как вы можете видеть, мое небольшое приложение имеет 4 задания, которые выполняются общей продолжительностью 20,2 секунды, однако между заданиями 1 и 2 существует большая задержка, из-за которой общее время превышает одну минуту.Задание № 1 runJob в SparkHadoopMapReduceWriter.scala: 88 выполняет массовую загрузку HFiles в таблицу HBase.Вот код, который я использовал для загрузки файлов

val outputDir = new Path(HBaseUtils.getHFilesStorageLocation(resolvedTableName))
val job = Job.getInstance(hBaseConf)
job.getConfiguration.set(TableOutputFormat.OUTPUT_TABLE, resolvedTableName)
job.setOutputFormatClass(classOf[HFileOutputFormat2])
job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])
job.setMapOutputValueClass(classOf[KeyValue])
val connection = ConnectionFactory.createConnection(job.getConfiguration)
val hBaseAdmin = connection.getAdmin
val table = TableName.valueOf(Bytes.toBytes(resolvedTableName))
val tab = connection.getTable(table).asInstanceOf[HTable]
val bulkLoader = new LoadIncrementalHFiles(job.getConfiguration)
preBulkUploadCallback.map(callback => callback())
bulkLoader.doBulkLoad(outputDir, hBaseAdmin, tab, tab.getRegionLocator)

Если у кого-то есть какие-либо идеи, я был бы очень рад

Answer 1

Спасибо за ваш вклад, ребята, я уменьшил количество HFiles, созданных в задании 0. Это уменьшило отставание примерно на 20%. Я использовал

HFileOutputFormat2.configureIncrementalLoad(job, tab, tab.getRegionLocator)

, который автоматически вычисляет количество задач сокращения, чтобы соответствовать текущему количеству областей для таблицы. Я скажу, что мы используем HBase, поддерживаемый S3 в AWS EMR, вместо классической HDFS. Я собираюсь выяснить, может ли это способствовать отставанию.

Answer 2

Уменьшите количество областей для выходной таблицы в Hbase, что приведет к уменьшению количества задач для вашей второй работы.

TableOutputFormat определяет разделение на основе количества областей для данной таблицы в Hbase

Answer 3

Задание № 1 runJob в SparkHadoopMapReduceWriter.scala: 88 выполняет массовую загрузку

Это не совсем так.Эта работа просто создает HFiles вне HBase.Разрыв, который вы видите между этой работой и следующей, может быть объяснен фактической массовой загрузкой в bulkLoader.doBulkLoad.Эта операция включает только передачу метаданных и обычно выполняется быстрее (по моему опыту), поэтому вам следует проверить журналы драйверов, чтобы увидеть, где они зависают.

Answer 4

Я вижу, что в задании 1 есть 26 задач, которые основаны на количестве созданных hfiles.Несмотря на то, что задание 2 показывает выполненное в 2 с, копирование этих файлов в целевое местоположение занимает некоторое время, и поэтому вы получаете задержку между заданиями 2 и 3. Этого можно избежать, сократив количество задач в задании 1.