如何使用 Apache Spark 遍历 Elasticsearch 源？

Question

我正在尝试通过将 Elasticsearch 与 Apache Spark 集成来构建推荐系统。我正在使用 Java。我使用 movilens 数据集作为示例数据。我也将数据索引到 Elasticsearch。到目前为止，我已经能够从 Elasticsearch 索引中读取输入，如下所示：

    SparkConf conf = new SparkConf().setAppName("Example App").setMaster("local");
conf.set("spark.serializer", org.apache.spark.serializer.KryoSerializer.class.getName());
        conf.set("es.nodes", "localhost");
        conf.set("es.port", "9200");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaPairRDD<String, Map<String, Object>> esRDD = JavaEsSpark.esRDD(sc, "movielens/recommendation");

使用 esRDD.collect() 函数，我可以看到我正在正确地从弹性搜索中检索数据。现在我需要将 Elasticsearch 结果中的用户 ID、项目 ID 和偏好提供给 Spark 的推荐。如果我使用的是 csv 文件，我可以这样做：

String path = "resources/user_data.data";
        JavaRDD<String> data = sc.textFile(path);
        JavaRDD<Rating> ratings = data.map(
          new Function<String, Rating>() {
            public Rating call(String s) {
              String[] sarray = s.split("   ");
              return new Rating(Integer.parseInt(sarray[0]), Integer.parseInt(sarray[1]), 
                                Double.parseDouble(sarray[2]));
            }
          }
        );

如果我需要遍历存储在 esRDD 中的弹性搜索输出并创建与上述类似的映射，那么什么是等效映射？如果有任何我可以参考的示例代码，那将有很大帮助。

Answer 1

很抱歉没有直接回答 Spark 问题，但如果您错过了，这里有使用 elasticsearch 对 MovieLens 数据进行推荐的描述：http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/_significant_terms_demo.html

Answer 2

您尚未指定 ElasticSearch 中数据的格式。但是我们假设它有字段userId、movieId 和rating，所以示例文档看起来像{"userId":1,"movieId":1,"rating":4}。

那么你应该能够做到（忽略空检查等）：

JavaRDD<Rating> ratings = esRDD.map(
    new Function<Map<String, Object>, Rating>() {
        public Rating call(Map<String, Object> m) {
              Int userId = Integer.parseInt(m.get("userId"));
              Int movieId = Integer.parseInt(m.get("movieId"));
              Double rating = Double.parseDouble(m.get("rating"));
              return new Rating(userId, movieId, rating);
        }
    }
);