GCP Dataproc: низкая пропускная способность сети с использованием разъема для хранения

Question

Обновление

При загрузке файлов с использованием DataFrame я достиг гораздо более высокой производительности.У меня не было возможности выяснить, почему это так, но чтение, как это, а затем преобразование в RDD является лучшим решением, которое я нашел до сих пор.

sparkSession.read.text("gs://bucket/some/sub/directory/prefix*")

---

Я пытаюсь сделатьпростое чтение файлов в GCS-хранилище в Dataproc Spark и вставка в таблицу Hive.Я получаю очень низкую пропускную способность сети (макс. 1 МБ / с) при загрузке файлов из корзины.

Кластер: 3 x n1-standard-4 (один мастер).

Bucketимеет 1440 GZIP-пунктов, ок.4MB каждый.

Я загружаю в искру, используя

sc.textFile("gs://bucket/some/sub/directory/prefix*")

Мой кластер dataproc и GCS bucket находятся в одной области / зоне.Ведро только региональное (не мультирегиональное).

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

Если бы я былчтобы загрузить те же данные с помощью gsutil cp (работающего на главном экземпляре виртуальной машины dataproc), требуется всего ~ 30 секунд.

Есть некоторые настройки, которые мне не хватает, или, возможно, подход sc.textFile (...)крайне неоптимально для GCS?

Спасибо

Answer 1

Этот пост должен ответить на несколько вопросов о различиях производительности RDD и DataFrame: https://www.linkedin.com/pulse/apache-spark-rdd-vs-dataframe-dataset-chandan-prakash

В целом:

• Производительность ввода-вывода GCS может варьироваться в зависимости от нагрузки
• Производительность ввода-вывода может варьироваться в зависимости от виртуальной машины зоны GCE
• IO зависит от количества процессоров и размера диска

В моем собственном тестировании этого поста gsutil cp для локальногодиск работал медленнее, в то время как различные распределенные команды были значительно быстрее, чем ваш набор данных (1440 текстовых файлов по 4 МБ случайных данных):

import timeit
i1 = sc.textFile("gs://my-bucket/input/*")

// Ordered by fastest first to slowest last:
timeit.timeit(lambda: spark.read.text("gs://.../input/*").count(), number=1)

timeit.timeit(lambda: i1.count(), number=1)

timeit.timeit(lambda: spark.read.text("gs://.../input/*").rdd.count(), number=1)