Как динамически составлять, где условие в PySpark

Question

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

Например, таблица 1 выглядит следующим образом:

+-----+-----------+
|Acct |Util_Change|
+-----+-----------+
|1    |0.5        |         
+-----+-----------+
|2    |0.8        |
+-----+-----------+

Таблица 2 выглядит как

+----------+-----------+-----------+
|Low_Change|High_Change|CLS        |
+----------+-----------+-----------+
|>0        |0.3        |T1         | # This means the util_change should be>0 and <=0.3  
+----------+-----------+-----------+
|>0.3      |<0.7       |T2         | # This means the util_change should be>0.3 and <0.7  
+----------+-----------+-----------+
|0.7       |1          |T3         | # This means the util_change should be>=0.7 and <=1  
+----------+-----------+-----------+

Я хочу объединить таблицу 1 и таблицу 2, сопоставив table1.Util_change сLow_change и High_change в таблице 2. Как видите, операторы сравнения определены в таблице 2.

Каким должен быть лучший способ кодировать это в PySpark?

Вот код для создания двух таблиц:

product = [(1, 0.5), (2, 0.8)]
sp = sqlContext.createDataFrame(product, ["Acct", "Util_Change"])

grid = [('>0', '0.3', 'T1'), ('>0.3', '<0.7', 'T2'), ('0.7', '1', 'T3')]
sp2 = sqlContext.createDataFrame(grid, ["Low_Change", "High_Change", "CLS"]