Как эффективно объединить фрейм данных PySpark?

Question

У меня есть два кадра данных в Pyspark, которые сливаются около двух дней.Первый - около 6 000 000 элементов x 2600 строк, а второй - около 30 элементов x 2600 строк.Я подозреваю, что так много времени занимает фактическая подготовка к слиянию до слияния.Вот мой код:

from pyspark.sql import SQLContext
import pyspark
from pyspark.sql.functions import col, split, create_map, lit
from pyspark.ml import Pipeline
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.feature import IndexToString, StringIndexer, VectorIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

sql_c = SQLContext(sc)

df = sql_c.read.option("maxColumns", 10000000).option("header", "true").options(samplingRatio=0.01).option("inferSchema", "true").csv('join_rows_no_prepended_new_line.csv')

df2 = sql_c.read.option("maxColumns", 10000000).option("header", "true").options(samplingRatio=0.01).option("inferSchema", "true").option("delimiter", "\t").csv('metadata_merged.txt')

#create a new column with a SampleID that matches the SampleID columns from the metadata df.
df = df.withColumn('#SampleID', split(df['# Gene Family'], '\_')[0])

df = df.drop("# Gene Family")
feature_cols = df.columns
df = df.join(df2, col("df.SampleID Gene Family")==col("df2.#SampleID"), how='inner')

Последняя строка - это однопоточная, работающая в течение двух дней.Есть ли лучший способ сделать это в Pyspark с точки зрения подготовки данных или что-то еще?

Спасибо.

Answer 1

• Spark SQL определенно не является подходящим инструментом для работы.

  Поскольку Spark SQL использует реляционную модель и планировщик запросов с оптимизатором, существует значительный объем памяти и вычислительные затраты с точки зренияЧисло столбцов.Нижняя граница является линейной (стоимость представления схемы), но на практике сложность планировщика запросов намного выше, а в худшем случае экспоненциальная.

  В результате Spark SQL можно удобно использовать, когдаколичество столбцов не превышает нескольких тысяч, хотя при необходимости его можно перенести в меньшие десятки тысяч.Миллионы столбцов просто не нужны.

• Возможно, неэффективный формат простого текста не подходит для работы.

• Вероятно, Spark MLне подходящий инструмент для этой работы.

  В общем, алгоритмы Spark ML могут достаточно хорошо работать с широкими собранными данными, если данные редки.Недостаточно информации, о которой идет речь, чтобы определить, так ли это на самом деле.

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

• Spark в целом может или не может быть подходящим инструментом для работы.

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

Некоторые из этих точек могут быть легко решены (например, использование RDD API для загрузки, анализа и сборки данных должно устранить узкое место оптимизатора), другие могут потребовать значительного объема работы (модели ансамблей с подмножествамифункции на коротких данных могут эффективно обучаться параллельно, при условии, что можно обеспечить эффективный выборочный доступ к данным).Остается вопрос, действительно ли это того стоит - размеры данных предполагают где-то в диапазоне 100 ГБ данных - ничего, что не может быть обработано в памяти на сервере среднего уровня.

---

* ЭтоКонечно, это не характерно для Spark.Большинство инструментов распределенной обработки делают подобные предположения по умолчанию.