java - Spark JAVA程序

Question

我正在尝试使用 Spark Java 解决这个问题:

Write some spark code which returns the number of trips for each customerId. A trip is a sequence of travel transactions with no more than 7 days in between each transaction.

输入:(客户 ID、姓名、姓名、性别、日期)
输出:(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();

解决方案:我创建了一个包含 CustomerId 和日期列表的映射对，接下来我想对日期进行排序并发现日期差异小于 7 天。

我是 Spark 新手，没有得到任何进一步继续的提示，有人可以帮助我吗？

示例输入和输出:

Answer 1

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年1月9日、2018年1月11日(7天)