使用 Java 读取 spark redis 保存的数据
Read data saved by spark redis using Java
我使用 spark-redis 将数据集保存到 Redis。
然后我使用 Spring data redis:
读取这些数据
我保存到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
public interface RatingResultRepository extends JpaRepository<RatingResult, String> {
}
我无法读取使用 Spring data redis 保存在 Redis 中的数据,因为 spark-redis 保存的结构数据和 spring data redis 不相同(我检查了键的值spark-redis 创建的和spring data redis 使用命令不同:redis-cli -p 6379 keys \*
and redis-cli hgetall $key
)
那么如何读取使用 Java 或 Java 中的任何库保存的数据?
以下适合我。
正在写入来自 spark-redis 的数据。
我在这里使用 Scala,但它与您在 Java 中使用的基本相同。我唯一改变的是我添加了一个 .option("key.column", "id")
来指定哈希 ID。
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()
在 spring-data-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;
@Override
public String toString() {
return "RatingResult{" +
"id='" + id + '\'' +
", user=" + user +
", product='" + product + '\'' +
", productN=" + productN +
", rating=" + rating +
", prediction=" + prediction +
'}';
}
}
我使用 CrudRepository 而不是 JPA:
@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}
您可能会注意到 id
字段未填充,因为存储的 spark-redis 具有哈希 ID 而不是哈希属性。
我使用 spark-redis 将数据集保存到 Redis。 然后我使用 Spring data redis:
读取这些数据我保存到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
public interface RatingResultRepository extends JpaRepository<RatingResult, String> {
}
我无法读取使用 Spring data redis 保存在 Redis 中的数据,因为 spark-redis 保存的结构数据和 spring data redis 不相同(我检查了键的值spark-redis 创建的和spring data redis 使用命令不同:redis-cli -p 6379 keys \*
and redis-cli hgetall $key
)
那么如何读取使用 Java 或 Java 中的任何库保存的数据?
以下适合我。
正在写入来自 spark-redis 的数据。
我在这里使用 Scala,但它与您在 Java 中使用的基本相同。我唯一改变的是我添加了一个 .option("key.column", "id")
来指定哈希 ID。
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()
在 spring-data-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;
@Override
public String toString() {
return "RatingResult{" +
"id='" + id + '\'' +
", user=" + user +
", product='" + product + '\'' +
", productN=" + productN +
", rating=" + rating +
", prediction=" + prediction +
'}';
}
}
我使用 CrudRepository 而不是 JPA:
@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}
您可能会注意到 id
字段未填充,因为存储的 spark-redis 具有哈希 ID 而不是哈希属性。