Spark SQL flatMap后RowFactory生成空行问题求助

我完全理解你的困惑：控制台明明打印出了拆分后的用户-物品-评分数据，但最终的Dataset<Row>调用show()却输出空行。这种情况通常是手动flatMap操作中Row编码或类型匹配的细节问题导致的，下面咱们一步步拆解并解决：

可能的核心原因 #

你的控制台能打印数据，说明flatMap的循环逻辑已经执行，且rows列表里确实有元素。问题大概率出在RowEncoder的类型匹配或者Spark对Row的序列化/反序列化处理上：

• 你手动创建的Row实例和RowEncoder定义的StructType可能存在隐性的类型不匹配（比如原始rating是Double类型，但你用getFloat(1)转成float再强转double，虽然最终类型是Double，但Spark的编码逻辑可能识别异常）；
• 手动构建Row并使用RowEncoder时，字段的顺序、类型必须和StructType完全一致，哪怕微小的差异都会导致Spark静默丢弃数据。

更可靠的解决方案：使用Spark内置的explode函数 #

手动flatMap处理数组拆分容易出错，Spark提供了更简洁、稳定的explode函数来实现这种“一行拆多行”的需求，完全避开手动Row编码的问题：

方式1：DataFrame API实现 #

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.*;
import org.apache.spark.sql.types.DataTypes;

// 第一步：将recommendations数组拆分为单独的行
Dataset<Row> explodedDF = userRecs.select(
    col("user"),
    explode(col("recommendations")).alias("rec")
);

// 第二步：提取拆分后结构体中的字段，并指定类型
Dataset<Row> recomenderResult = explodedDF.select(
    col("user"),
    col("rec.iID").cast(DataTypes.IntegerType).alias("item"),
    col("rec.rating").cast(DataTypes.DoubleType).alias("relevance")
);

// 验证结果
recomenderResult.show();
recomenderResult.write().json("temp2");

方式2：Spark SQL实现 #

// 将原数据集注册为临时视图
userRecs.createOrReplaceTempView("user_recs");

// 通过SQL语句完成数组拆分和字段提取
Dataset<Row> recomenderResult = spark.sql("""
    SELECT 
        user,
        rec.iID AS item,
        rec.rating AS relevance
    FROM user_recs
    LATERAL VIEW explode(recommendations) AS rec
""");

recomenderResult.show();
recomenderResult.write().json("temp2");

如果一定要用手动flatMap的修复方案 #

如果你坚持使用原有的flatMap方式，建议做以下调整：

1. 确认原数据集的Schema：先打印userRecs的Schema，确保recommendations字段是ArrayType，且内部的结构体是iID: IntegerType和rating: DoubleType：

   userRecs.printSchema();
   

2. 使用Java Bean代替RowEncoder：自定义一个简单的Java类来封装结果，避免Row编码的复杂性：

   // 定义Java Bean类
   public class UserRecommendation {
       private String user;
       private Integer item;
       private Double relevance;
   
       // 无参构造器（必须）
       public UserRecommendation() {}
   
       // 带参构造器
       public UserRecommendation(String user, Integer item, Double relevance) {
           this.user = user;
           this.item = item;
           this.relevance = relevance;
       }
   
       // Getter和Setter方法（必须）
       public String getUser() { return user; }
       public void setUser(String user) { this.user = user; }
       public Integer getItem() { return item; }
       public void setItem(Integer item) { this.item = item; }
       public Double getRelevance() { return relevance; }
       public void setRelevance(Double relevance) { this.relevance = relevance; }
   }
   

3. 修改flatMap逻辑，使用Bean Encoder：

   import org.apache.spark.sql.Encoders;
   import org.apache.spark.sql.Encoder;
   
   Encoder<UserRecommendation> encoder = Encoders.bean(UserRecommendation.class);
   Dataset<UserRecommendation> resultDS = userRecs.flatMap((FlatMapFunction<Row, UserRecommendation>) row -> {
       String user = row.getString(0);
       List<Row> recs = row.getList(1);
       List<UserRecommendation> rows = new ArrayList<>();
       for (Row rec : recs) {
           // 直接按原始类型获取，避免不必要的类型转换
           Integer item = rec.getInt(0);
           Double relevance = rec.getDouble(1);
           System.out.println(user + " : " + item + " : " + relevance);
           rows.add(new UserRecommendation(user, item, relevance));
       }
       return rows.iterator();
   }, encoder);
   
   // 转换为DataFrame并验证
   Dataset<Row> recomenderResult = resultDS.toDF();
   recomenderResult.show();
   recomenderResult.write().json("temp2");
   

这样调整后，就能确保类型匹配的准确性，避免Spark丢弃数据的问题。

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