Проблема с производительностью Spark - запись разделов в S3 как отдельных файлов

Question

Я запускаю искровое задание, задача которого - сканировать большой файл и разбивать его на более мелкие. Файл находится в формате Json Lines, и я пытаюсь разбить его по определенному столбцу (id) и сохранить каждый раздел как отдельный файл на S3. Размер файла составляет около 12 ГБ, но существует около 500000 различных значений id. Запрос занимает почти 15 часов. Что я могу сделать для повышения производительности? Spark - плохой выбор для такой задачи? Обратите внимание, что я могу убедиться, что источник - это фиксированное количество строк для каждого идентификатора.

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql.window import Window
from awsglue.utils import getResolvedOptions
from awsglue.transforms import *
from pyspark.sql.functions import udf, substring, instr, locate
from datetime import datetime, timedelta

    
sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
# Get parameters that were passed to the job
args = getResolvedOptions(sys.argv, ['INPUT_FOLDER', 'OUTPUT_FOLDER', 'ID_TYPE', 'DATASET_DATE'])

id_type = args["ID_TYPE"]
output_folder = "{}/{}/{}".format(args["OUTPUT_FOLDER"], id_type, args["DATASET_DATE"])
input_folder = "{}/{}/{}".format(args["INPUT_FOLDER"], id_type, args["DATASET_DATE"])


INS_SCHEMA = StructType([
    StructField("camera_capture_timestamp", StringType(), True),
    StructField(id_type, StringType(), True),
    StructField("image_uri", StringType(), True)
])


data = spark.read.format("json").load(input_folder, schema=INS_SCHEMA)

data = data.withColumn("fnsku_1", F.col("fnsku"))

data.coalesce(1).write.partitionBy(["fnsku_1"]).mode('append').json(output_folder)   

Я тоже пробовал переразбивать вместо слияния.

Я использую AWS Glue

Answer 1

Пожалуйста, рассмотрите следующие варианты как один из возможных. Было бы здорово посмотреть, помогло ли это:)

Во-первых, если вы объединитесь, как сказал @Lamanus в комментариях, это означает, что вы уменьшите количество разделов, следовательно, также уменьшите количество задач записи, следовательно, перетасуйте все данные в 1 задачу. Это может быть первым фактором, который нужно улучшить.

Чтобы побороть проблему, т.е. написать файл для каждого раздела и сохранить уровень распараллеливания, вы можете изменить логику на следующую:

object TestSoAnswer extends App {

  private val testSparkSession = SparkSession.builder()
    .appName("Demo groupBy and partitionBy").master("local[*]")
    .getOrCreate()
  import testSparkSession.implicits._

  // Input dataset with 5 partitions
  val dataset = testSparkSession.sparkContext.parallelize(Seq(
    TestData("a", 0), TestData("a", 1), TestData("b", 0), TestData("b", 1),
    TestData("c", 1), TestData("c", 2)
  ), 5).toDF("letter", "number")

  dataset.as[TestData].groupByKey(row => row.letter)
    .flatMapGroups {
      case (_, values) => values
    }.write.partitionBy("letter").mode("append").json("/tmp/test-parallel-write")

}

case class TestData(letter: String, number: Int)

Как это работает?

Сначала код выполняет перемешивание, чтобы собрать все строки, относящиеся к определенному ключу (также, как и для разделения), в одни и те же разделы. Таким образом, он будет выполнять запись сразу во все строки, принадлежащие ключу. Некоторое время назад я написал в блоге сообщение оpartitionByметод. Грубо говоря, внутренне он будет сортировать записи в данном разделе, а затем записывать их одну за другой в файл.

Таким образом, мы получаем план, подобный этому, где только 1 перемешивание, поэтому присутствует операция, требующая обработки:

== Physical Plan ==
*(2) SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, knownnotnull(assertnotnull(input[0, TestData, true])).letter, true, false) AS letter#22, knownnotnull(assertnotnull(input[0, TestData, true])).number AS number#23]
+- MapGroups TestSoAnswer$$$Lambda$1236/295519299@55c50f52, value#18.toString, newInstance(class TestData), [value#18], [letter#3, number#4], obj#21: TestData
   +- *(1) Sort [value#18 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(value#18, 200), true, [id=#15]
         +- AppendColumnsWithObject TestSoAnswer$$$Lambda$1234/1747367695@6df11e91, [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, knownnotnull(assertnotnull(input[0, TestData, true])).letter, true, false) AS letter#3, knownnotnull(assertnotnull(input[0, TestData, true])).number AS number#4], [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false) AS value#18]
            +- Scan[obj#2]

Выход TestSoAnswer выполненный дважды выглядит так:

test-parallel-write % ls
_SUCCESS letter=a letter=b letter=c
test-parallel-write % ls letter=a
part-00170-68245d8b-b155-40ca-9b5c-d9fb746ac76c.c000.json part-00170-cd90d64f-43c6-4582-aae6-fe443b6617f4.c000.json

test-parallel-write % ls letter=b
part-00161-68245d8b-b155-40ca-9b5c-d9fb746ac76c.c000.json part-00161-cd90d64f-43c6-4582-aae6-fe443b6617f4.c000.json

test-parallel-write % ls letter=c
part-00122-68245d8b-b155-40ca-9b5c-d9fb746ac76c.c000.json part-00122-cd90d64f-43c6-4582-aae6-fe443b6617f4.c000.json

Вы также можете контролировать количество записей, записываемых в файл с помощью этой конфигурации.

Изменить: не видел комментарий @mazaneicha, но действительно, вы можете попробовать с repartition("partitioning column")! Это даже более ясно, чем выражение группировки.

Лучший,

Бартош.

Answer 2

Если вы не собираетесь использовать Spark для чего-либо, кроме разделения файла на более мелкие версии самого себя, то я бы сказал, что Spark - плохой выбор. Лучше сделать это в AWS, следуя подходу, подобному тому, который приведен в этой публикации о переполнении стека.

Предполагая, что у вас есть доступный экземпляр EC2, вы должны запустить что-то вроде этого:

aws s3 cp s3://input_folder/12GB.json - | split -l 1000 - output.
aws s3 cp output.* s3://output_folder/

Если вы хотите продолжить обработку данных в Spark, вам нужно будет перераспределить данные на фрагменты размером от 128 МБ до 1 ГБ. При сжатии по умолчанию (мгновенное) обычно получается 20% от исходного размера файла. Итак, в вашем случае: между (12/5) ~3 и (12/5/8) ~20 разделами, поэтому:

data = spark.read.format("json").load(input_folder, schema=INS_SCHEMA) 

dataPart = data.repartition(12)

На самом деле это не очень большой набор данных для Spark, и поэтому он не должен быть таким громоздким.

Сохранение в виде паркета дает вам хорошую точку восстановления, а повторное считывание данных будет очень быстрым. Общий размер файла составит около 2,5 ГБ.