Spark Dataset присоединиться к производительности

Question

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

    def joinFunction(dogs: Dataset[Dog]): Dataset[(Dog, Cat)] = {
      val cats: Dataset[Cat] = spark.table("dev_db.cat").as[Cat]
      dogs.joinWith(cats, ...)
    }

. Здесь моя основная проблема связана с spark.table("dev_db.cat")* 1008.*, как будто мы ссылаемся на все данные cat как

    SELECT * FROM dev_db.cat

, а затем делаем join на более позднем этапе.Или оптимизатор запросов будет непосредственно выполнять соединение без обращения ко всей таблице?Есть ли лучшее решение?

Answer 1

Вот несколько предложений для вашего случая:

a. Если у вас есть операции where, filter, limit, take и т.д., попробуйте применить их дообъединение двух наборов данных.Spark не может подавить фильтры такого рода, поэтому вы должны самостоятельно уменьшить максимальное количество целевых записей. Здесь отличный источник информации об оптимизаторе Spark.

b. Попробуйте найти наборы данных и свести к минимуму перемешанные данные, используя функцию repartition.Перераспределение должно основываться на ключах, которые участвуют в join, то есть:

dogs.repartition(1024, "key_col1", "key_col2")
dogs.join(cats, Seq("key_col1", "key_col2"), "inner")

c. Попробуйте использовать broadcast для меньшего набора данных, если вы уверены, что он можетпоместиться в память (или увеличить значение spark.broadcast.blockSize).Это обеспечивает определенный прирост производительности вашей программы Spark, поскольку она обеспечивает сосуществование двух наборов данных в одном узле.

Если вы не можете применить что-либо из вышеперечисленного, Spark не сможет определить, какие записи следует исключить, и поэтому будет сканировать все доступные строки из обоих наборов данных.

Answer 2

Вам нужно сделать объяснение и посмотреть, используется ли предикатное нажатие.Тогда вы можете судить, что ваша проблема правильна или нет.

Однако, как правило, теперь, если сложные типы данных не используются и / или несоответствия типов данных не очевидны, то происходит уменьшение.Вы можете увидеть это с помощью простого createOrReplaceTempView.См. https://databricks -prod-cloudfront.cloud.databricks.com / public / 4027ec902e239c93eaaa8714f173bcfc / 3741049972324885/4201913720573284/4413065072037724 / latest.html