如何在Spark SQL中优化非equi-join？

Question

我有两个数据帧，我需要使用具有两个连接谓词的非相等连接(即不等式连接)将它们连接在一起。

一个数据帧是直方图DataFrame[bin: bigint, lower_bound: double, upper_bound: double]

另一个数据帧是观察值DataFrame[id: bigint, observation: double]的集合

我需要确定每个观察值落入直方图的哪个柱状图中，如下所示：

observations_df.join(histogram_df, 
    (
        (observations_df.observation >= histogram_df.lower_bound) &
        (observations_df.observation < histogram_df.upper_bound)
    )
   )

基本上，它是非常慢的，我正在寻找一些建议，关于如何让它更快。

下面是演示该问题的一些示例代码。observations_df包含100000行，当histogram_df中的行数变得适当大时(假设是number_of_bins = 500000)，它就会变得非常非常慢，我敢肯定这是因为我正在进行非等联接。如果您运行此代码，然后使用number_of_rows的值，请从较低的值开始，然后增加，直到性能明显变慢为止

from pyspark.sql.functions import lit, col, lead
from pyspark.sql.types import *
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import rand
from pyspark.sql import Window
spark = SparkSession \
    .builder \
    .getOrCreate()

number_of_bins = 500000

bin_width = 1.0 / number_of_bins
window = Window.orderBy('bin')
histogram_df = spark.range(0, number_of_bins)\
    .withColumnRenamed('id', 'bin')\
    .withColumn('lower_bound', 0 + lit(bin_width) * col('bin'))\
    .select('bin', 'lower_bound', lead('lower_bound', 1, 1.0).over(window).alias('upper_bound'))
observations_df = spark.range(0, 100000).withColumn('observation', rand())
observations_df.join(histogram_df, 
        (
            (observations_df.observation >= histogram_df.lower_bound) &
            (observations_df.observation < histogram_df.upper_bound)
        )
       ).groupBy('bin').count().head(15)

Answer 1

对于spark join，不建议使用不相等的join。通常，我会生成一个新列作为此类操作的连接键。然而，对于您的情况，您不需要连接来确定每个观察值落入直方图的哪个柱状图中，因为每个柱状图的上下界都可以预先计算，并且您可以使用观察值计算柱状图。

你可以参考pyspark: passing multiple dataframe fields to udf