PySpark SQL子查询报错：用作表达式的子查询返回多行

Hey there! Let's break down exactly what's causing this error and how to fix it—it's a common gotcha with Spark SQL's subquery rules.

The Root Cause #

That error pops up because Spark requires subqueries used as expressions (like in your SELECT clause) to return exactly one row and one column (a scalar value). When you tried to pull aml_freq_a, aml_freq_b, aml_freq_c from train using subqueries, Spark found that one or more of those subqueries was returning multiple rows instead of a single value per row in test.

For example, if your query looked something like this:

SELECT
  test.*,
  (SELECT aml_freq_a FROM train WHERE train.a = test.a) AS aml_freq_a,
  (SELECT aml_freq_b FROM train WHERE train.b = test.b) AS aml_freq_b,
  (SELECT aml_freq_c FROM train WHERE train.c = test.c) AS aml_freq_c
FROM test

The problem is likely one of two things:

1. Your train table has duplicate entries for the same a/b/c value with the same (or different) frequency values—so the subquery returns multiple rows for a single match in test.
2. You forgot to add a proper join condition in the subquery, so it's pulling all rows from train's frequency columns instead of matching to test's rows.

What You Missed #

You probably didn't account for Spark's strict scalar subquery requirement. Unlike some other SQL engines that might implicitly pick a row (like the first one), Spark throws this error to avoid silent data issues. Additionally, using subqueries for this kind of column enrichment is less reliable than using joins—joins are designed to handle row-to-row associations safely.

Fixes to Try #

1. Use Joins (Recommended) #

This is the most robust approach. First, make sure you have a clean view of train with unique a/b/c values and their corresponding frequencies (if you haven't already):

-- Create separate views for each column's frequency
CREATE OR REPLACE TEMP VIEW train_a_freq AS
SELECT a, COUNT(*) AS aml_freq_a FROM train GROUP BY a;

CREATE OR REPLACE TEMP VIEW train_b_freq AS
SELECT b, COUNT(*) AS aml_freq_b FROM train GROUP BY b;

CREATE OR REPLACE TEMP VIEW train_c_freq AS
SELECT c, COUNT(*) AS aml_freq_c FROM train GROUP BY c;

Then join these views to your test DataFrame:

SELECT
  t.*,
  ta.aml_freq_a,
  tb.aml_freq_b,
  tc.aml_freq_c
FROM test t
LEFT JOIN train_a_freq ta ON t.a = ta.a
LEFT JOIN train_b_freq tb ON t.b = tb.b
LEFT JOIN train_c_freq tc ON t.c = tc.c;

This ensures each frequency value is matched exactly once per row in test, and avoids any subquery row limits.

2. Force Subqueries to Return a Single Row #

If you really want to stick with subqueries, use an aggregate function like MAX() or FIRST() to guarantee a single value is returned. This works only if all frequency values for a given a/b/c are identical (which they should be if you calculated frequencies correctly):

SELECT
  t.*,
  (SELECT MAX(aml_freq_a) FROM train WHERE train.a = t.a) AS aml_freq_a,
  (SELECT MAX(aml_freq_b) FROM train WHERE train.b = t.b) AS aml_freq_b,
  (SELECT MAX(aml_freq_c) FROM train WHERE train.c = t.c) AS aml_freq_c
FROM test t;

Note: Use this only if you're 100% sure there are no conflicting frequency values for the same a/b/c in train.

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