使用 [spark-redis][1] 将 Dataset 保存到 Redis。
然后我使用 [Spring data redis][2] 读取这些数据：
我保存到 Redis 的对象：
    @Getter
    @Setter
    @AllArgsConstructor
    @NoArgsConstructor
    @Builder
    @RedisHash("collaborative_filtering")
    public class RatingResult implements Serializable {
    	private static final long serialVersionUID = 8755574422193819444L;
    
    	@Id
    	private String id;
    
    	@Indexed
    	private int user;
    
    	@Indexed
    	private String product;
    
    	private double productN;
    	private double rating;
    	private float prediction;
    
    	public static RatingResult convert(Row row) {
    		int user = row.getAs("user");
    		String product = row.getAs("product");
    		double productN = row.getAs("productN");
    		double rating = row.getAs("rating");
    		float prediction = row.getAs("prediction");
    		String id = user + product;
    
    		return RatingResult.builder().id(id).user(user).product(product).productN(productN).rating(rating)
    				.prediction(prediction).build();
    	}
    
    }
使用 spark-redis 进行对象保存：
    JavaRDD<RatingResult> result = ...
    ...
    sparkSession.createDataFrame(result, RatingResult.class).write().format("org.apache.spark.sql.redis")
    			.option("table", "collaborative_filtering").mode(SaveMode.Overwrite).save();
Repository：
    @Repository
    public interface RatingResultRepository extends JpaRepository<RatingResult, String> {
    
    }
我无法使用 Spring data redis 读取通过 spark-redis 保存在 Redis 中的数据，因为结构数据由于 spark-redis 和 spring data redis 的不同而不同（我使用命令进行检查：`redis-cli -p 6379 keys \*` 和 `redis-cli hgetall $key`，spark-redis 和 spring data redis 创建的键的值不同）
那么如何使用 Java 或者 Java 中的任何库来读取已经保存的这些数据呢？
  [1]: https://github.com/RedisLabs/spark-redis/blob/master/doc/java.md
  [2]: https://www.baeldung.com/spring-data-redis-tutorial

    val ratingResult = new RatingResult("1", 1, "product1", 2.0, 3.0, 4)
    val result: JavaRDD[RatingResult] = spark.sparkContext.parallelize(Seq(ratingResult)).toJavaRDD()
    spark
      .createDataFrame(result, classOf[RatingResult])
      .write
      .format("org.apache.spark.sql.redis")
      .option("key.column", "id")
      .option("table", "collaborative_filtering")
      .mode(SaveMode.Overwrite)
      .save()

@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@Builder
@RedisHash("collaborative_filtering")
public class RatingResult implements Serializable {
    private static final long serialVersionUID = 8755574422193819444L;
    @Id
    private String id;
    @Indexed
    private int user;
    @Indexed
    private String product;
    private double productN;
    private double rating;
    private float prediction;
    @Override
    public String toString() {
        return "RatingResult{" +
                "id='" + id + '\'' +
                ", user=" + user +
                ", product='" + product + '\'' +
                ", productN=" + productN +
                ", rating=" + rating +
                ", prediction=" + prediction +
                '}';
    }
}

@Repository
public interface RatingResultRepository extends CrudRepository<RatingResult, String> {
}

     RatingResult found = ratingResultRepository.findById("1").get();
     System.out.println("found = " + found);

found = RatingResult{id='null', user=1, product='product1', productN=2.0, rating=3.0, prediction=4.0}