如何在Scala中比较无匹配ID及属性的两个DataFrame

方案1：行哈希快速定位完全差异行 #

如果目标是找出两个DataFrame中完全相同或完全不同的行，直接给每行生成全局哈希值是最高效的方式，无需排序或逐行对比。

实现示例（Spark）：

# 为每个DataFrame添加行哈希列，组合所有列计算哈希
df1 = df1.withColumn("row_hash", hash(*df1.columns))
df2 = df2.withColumn("row_hash", hash(*df2.columns))

# 提取df1独有的行
df1_only = df1.join(df2, "row_hash", "left_anti").drop("row_hash")
# 提取df2独有的行
df2_only = df2.join(df1, "row_hash", "left_anti").drop("row_hash")
# 提取两行都存在的完全匹配行
matched_rows = df1.join(df2, "row_hash", "inner").drop("row_hash")

注意：哈希存在极小碰撞概率，若对精度要求极高，可改用sha2生成更长哈希值，比如sha2(concat_ws("|", *df1.columns), 256)。

方案2：统计特征分组+组内精细匹配 #

如果行之间并非完全相同，但存在相似统计特征（如时间戳相近、数值列分布相似），可先通过统计特征分组缩小匹配范围：

• 时间型列：用date_trunc按小时/天/周截断，生成时间分组标签
• 数值型列：用approxQuantile计算分位数，将数值列映射到分桶区间（如0-25%、25%-50%等）
• 类别型列：直接用列值作为分组标签

给两个DataFrame添加分组标签后按标签join，再在组内计算行之间的相似度（如时间差、数值差之和），筛选最相似的配对。

示例（Spark）：

# 对时间戳按小时分组
df1 = df1.withColumn("time_group", date_trunc("hour", df1.timestamp_col))
df2 = df2.withColumn("time_group", date_trunc("hour", df2.timestamp_col))

# 按时间分组join，组内计算时间差
joined = df1.join(df2, "time_group", "inner")
joined = joined.withColumn("time_diff", abs(unix_timestamp(df1.timestamp_col) - unix_timestamp(df2.timestamp_col)))

# 筛选组内时间差最小的配对
from pyspark.sql import Window
window = Window.partitionBy(df1["time_group"], df1["temp_id"]).orderBy("time_diff")
best_match = joined.withColumn("rank", row_number().over(window)).filter("rank = 1")

方案3：近似最近邻（LSH）匹配相似行 #

如果需要找到最相似的行（而非完全匹配），可使用LSH（局部敏感哈希）算法将每行转成特征向量，快速定位近似相似行：

1. 将所有列转成数值型：时间戳转成Unix时间戳，类别列用StringIndexer+OneHotEncoder编码
2. 用BucketedRandomProjectionLSH训练模型，生成特征向量的哈希桶
3. 对两个DataFrame的特征向量做近似相似性查询

示例（Spark）：

from pyspark.ml.feature import VectorAssembler, StringIndexer, OneHotEncoder
from pyspark.ml.feature import BucketedRandomProjectionLSH

# 处理类别列
indexer = StringIndexer(inputCol="category_col", outputCol="category_idx")
encoder = OneHotEncoder(inputCol="category_idx", outputCol="category_vec")

# 组装特征向量：数值列+时间戳(Unix)+编码后的类别列
assembler = VectorAssembler(
    inputCols=["num_col1", "num_col2", "unix_timestamp(timestamp_col)", "category_vec"],
    outputCol="features"
)

# 训练LSH模型并匹配相似行
lsh = BucketedRandomProjectionLSH(inputCol="features", outputCol="hashes", bucketLength=1.0)
model = lsh.fit(df1_transformed)
similarity_df = model.approxSimilarityJoin(df1_transformed, df2_transformed, threshold=10.0, distCol="distance")

该方案适合行数较多、无明确匹配键的场景，能高效缩小相似行范围。

方案4：小数据量下的全量配对+差异筛选 #

如果两个DataFrame行数都很小（如<10000行），可直接做笛卡尔积，计算每行对的差异得分，筛选差异最小的配对：

# 给每个DataFrame添加临时ID
df1 = df1.withColumn("temp_id1", monotonically_increasing_id())
df2 = df2.withColumn("temp_id2", monotonically_increasing_id())

# 笛卡尔积并计算差异得分
cross_join = df1.crossJoin(df2)
cross_join = cross_join.withColumn(
    "diff_score",
    abs(df1.num_col1 - df2.num_col1) + abs(df1.num_col2 - df2.num_col2) +
    abs(unix_timestamp(df1.timestamp_col) - unix_timestamp(df2.timestamp_col)) +
    when(df1.category_col != df2.category_col, 1).otherwise(0)
)

# 筛选每个df1行对应的最小差异df2行
window = Window.partitionBy("temp_id1").orderBy("diff_score")
best_match = cross_join.withColumn("rank", row_number().over(window)).filter("rank = 1")

注意：大数据量下笛卡尔积会导致数据爆炸，绝对不要使用。

内容的提问来源于stack exchange，提问作者Dan