Spark执行分组取Top3查询报错，求适配MySQL原逻辑的方案

嘿，这个问题我太熟悉了！你碰到的org.apache.spark.sql.AnalysisException: cannot recognize input near 'num' ':' '=' in expression specification报错，本质是Spark SQL不支持MySQL那套用户变量（@num、@group）的赋值语法——这种带:=的变量写法是MySQL专属的，Spark SQL的语法解析器完全认不出它，自然就报错了。

不过别担心，Spark里实现「每个分类取前3条记录」的需求，有更贴合分布式计算的标准方案：用窗口函数，比MySQL的变量写法简洁得多，还更稳定。

具体实现代码 #

假设你已经把MySQL的products表注册成了Spark临时视图（比如命名为products_temp），直接用下面的代码就能实现需求：

val query = """
SELECT product_id, product_category_id, product_price
FROM (
    SELECT 
        product_id, 
        product_category_id, 
        product_price,
        -- 窗口函数：按分类分组，组内按价格排序（可按需改升序/降序）
        ROW_NUMBER() OVER (PARTITION BY product_category_id ORDER BY product_price DESC) AS rn
    FROM products_temp
) ranked_products
WHERE rn <= 3
"""
// 执行查询并展示结果
val top3ByCategory = spark.sql(query)
top3ByCategory.show()

代码逻辑解释 #

• ROW_NUMBER() OVER (PARTITION BY product_category_id ORDER BY product_price DESC)：这是核心的窗口函数逻辑：
  • PARTITION BY product_category_id：把数据按商品分类分组，和你MySQL逻辑里的@group = product_category_id作用一致；
  • ORDER BY product_price DESC：每个分组内按商品价格降序排序（如果需要升序，把DESC改成ASC即可）；
  • ROW_NUMBER()：给每个分组内的记录分配唯一的连续序号，序号从1开始。
• 外层查询通过WHERE rn <=3过滤掉序号大于3的记录，最终就得到了每个分类下的前3条记录。

可选调整：支持并列排名 #

如果你的业务允许同价格的商品并列进入Top3（比如两个同价商品都算第2名），可以把ROW_NUMBER()换成RANK()或者DENSE_RANK()：

• RANK()：会跳过重复的序号（比如两个第2名后，下一个是第4名）；
• DENSE_RANK()：不会跳过重复序号（两个第2名后，下一个是第3名）。

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