Программа Spark JAVA - PullRequest
Новая
       11

Программа Spark JAVA

0 голосов
/ 06 сентября 2018

Я пытаюсь решить эту проблему с помощью Spark Java:

Напишите некоторый искровой код, который возвращает количество поездок для каждого Пользовательский ИД. Поездка - это последовательность транзакций, в которых больше нет более 7 дней между каждой транзакцией.

Ввод: (customerid, имя, имя, пол, дата)
Выход: (customerid, numTrips)

Код, который я пробовал: 

SimpleDateFormat sdf = new SimpleDateFormat("MM/dd/yyyy");
TripData td=new TripData();
JavaRDD<String> getData= jsc.read().textFile("E:/PROJECTS/SPARK/RESOURCES/TRIPS_MOCK_DATA.txt").javaRDD();
JavaPairRDD<Integer,Iterable<Long>> tripByKey=  getData.mapToPair(line->
    {
        String[] parts=SPACES.split(line);
        return new Tuple2<>(Integer.parseInt(parts[0]),sdf.parse(parts[4]).getTime());
    }).groupByKey().sortByKey();

Решение: Я создал пару Map с CustomerId и списком дат, затем я хочу отсортировать даты и найти разницу в датах не более 7 дней.

Я новичок в Spark, у меня нет намеков на дальнейшие действия, может кто-нибудь помочь мне?

Пример ввода и вывода:

Sample

1 Ответ

0 голосов
/ 12 июня 2019 
JavaPairRDD<Integer,Iterable<Long>> tripByKey=  getData.mapToPair(line->
    {
        String[] parts=SPACES.split(line);
        return new Tuple2<>(Integer.parseInt(parts[0]),sdf.parse(parts[4]).getTime());
    }).groupByKey();

После преобразования groupByKey у нас будет пользователь (ключ), все даты для этого пользователя (значение как итеративное). Теперь, чтобы получить количество поездок на пользователя, нам нужно будет отсортировать эти даты и использовать нашу логику для получения поездок на пользователя. 

JavaPairRDD<Integer, Integer> tripsCountPerUser = tripByKey.mapValues(func);


 Function<Iterable<Long>, Integer> fun = (Iterable<Long> itr ) -> {
        List<Long> dates = new ArrayList<>();
        for (Long i:itr) {
            dates.add(i);
        }
        Collections.sort(dates);
        long day = 86400000l ;
        long days7 = day * 7;

        int count = 0;
        Long firstDay = null;
        for (Long dt : dates) {
            if(firstDay == null)
            {
                firstDay = dt;
                count = 1;
            }
            else {
                Long diffMs = dt - firstDay ;

                if(diffMs > days7 ) {
                    firstDay = dt;
                    count ++;
                }
            }
        }
        return count;
    };

Приведенный выше код не является оптимизированным кодом и может быть выполнен несколькими различными способами. Приведенный выше код предназначен для решения проблемы. Надеюсь, это поможет.

Предположение: -

2018-01-01, 2018-01-08, 2018-01-09, 2018-01-11 рассматривают две поездки, которые

  1. 2018-01-01, 2018-01-08 (в течение 7 дней)
  2. 2018-01-09, 2018-01-11 (7 дней)
...