Лучшая практика для изменения большого фрейма данных pyspark с одним значением из другого фрейма данных на основе условия

Question

У меня есть два фрейма данных, один большой, а другой простой столбец / строка. Рассмотрим второй фрейм данных, содержащий одно значение, которое может выступать в качестве верхней границы. Первый -

`+----+-----+-----+----+
|col1|col2 |test1|val. |
+----+-----+-----+----+
|   a|  foo|   p1|  24|
|   b|  bar|   p2|  56|
|   c|  baz|   p3|   8|
+----+-----+-----+----+

Второй - с одним значением

+--------------+
| upper_bound  |
+--------------+
|25            |
+--------------+

Я хочу добавить новый логический столбец в первый DataFrame, используя upper_bound второго кадра данных и условие (ie: если одно значение val равно> = upper_bound)

Желаемый результат:

+----+-----+-----+----+-----+
|col1|col2 |test1|val |bound|
+----+-----+-----+----+-----+
|   a|  foo|   p1|  24|0    | 
|   b|  bar|   p2|  56|1    |
|   c|  baz|   p3|   8|0    |
+----+-----+-----+----+-----+

Каков наилучший способ сделать это в pyspark? Получаю ли я данные из второго фрейма данных и использую withColumn и filter?

Это не очень хорошая практика для secondFrame.take(1)[0]['upper_bound']

Answer 1

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

Это не очень хорошая практика для secondFrame.take(1)[0]['upper_bound']

Я думаю более быстрое решение (и большее количество памяти), которое вы можете получить, если вы точно знаете значение,

df.withColumn("bound", psf.col("val")>=psf.lit(25))

или, если вы не хотите более общую версию,

df.withColumn("bound", psf.col("val")>=secondFrame.take(1)[0]['upper_bound'])

Если вы действительно не хотите использовать secondFrame.take(1)[0]['upper_bound'], что является самым простым путем, вы можете сделать широковещательное соединение. Нечто подобное может сделать это:

df2 = rdd.toDF(['upper_bound']).withColumn('dummycol', psf.lit(1))
df  = df.withColumn('dummycol', psf.lit(1))
df  = (df.join(df2, ['dummycol'], how = "left_outer")
          .withColumn("bound", psf.col("val") >= psf.col("upper_bound"))