Как перебирать источник Elasticsearch с помощью Apache Spark?

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 до рекомендации 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

Извиняюсь за не отвечая на вопрос Спарк непосредственно, но в случае, если вы пропустили его, есть описание делать рекомендации по данным MovieLens использованием elasticsearch здесь: 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); 
     } 
    } 
);