Task G - Outdated Pages - CircleNet Analytics

Task G identifies “outdated” CircleNetPages by finding users who have not accessed CircleNet for 90 days (defined as 90 days × 24 hours = 2,160 time units from the maximum timestamp).

Problem Statement

Return IDs and nicknames of persons that have not accessed CircleNet for 90 days (no entries in ActivityLog in the last 90 days). The ActivityLog uses hourly timestamp granularity. SQL Equivalent:

WITH GlobalMax AS (
    SELECT MAX(ActionTime) as max_time
    FROM ActivityLog
),
UserLastAccess AS (
    SELECT ByWho, MAX(ActionTime) as last_access
    FROM ActivityLog
    GROUP BY ByWho
)
SELECT p.ID, p.NickName
FROM CircleNetPage p
LEFT JOIN UserLastAccess u ON p.ID = u.ByWho
CROSS JOIN GlobalMax g
WHERE u.last_access IS NULL 
   OR u.last_access < (g.max_time - 2160);  -- 90 days * 24 hours

Approaches

Simple
Optimized

Simple Implementation (3 Jobs)

The simple approach uses three sequential MapReduce jobs to find the global max, user max, and filter outdated users.Job 1: Find Global Maximum Timestamp (TaskGSimple.java:24-53)

public static class GlobalMaxMapper extends Mapper<LongWritable, Text, NullWritable, IntWritable> {
private final IntWritable out = new IntWritable();

@Override
protected void map(LongWritable key, Text value, Context context) 
    throws IOException, InterruptedException {
    String[] f = CsvUtils.split(value.toString());
    if (f.length >= 5) {
        out.set(CsvUtils.toInt(f[4], -1));  // ActionTime
        if (out.get() >= 0) {
            context.write(NullWritable.get(), out);
        }
    }
}
}

public static class MaxReducer extends Reducer<NullWritable, IntWritable, NullWritable, IntWritable> {
@Override
protected void reduce(NullWritable key, Iterable<IntWritable> values, Context context) 
    throws IOException, InterruptedException {
    int max = Integer.MIN_VALUE;
    for (IntWritable v : values) {
        if (v.get() > max) {
            max = v.get();
        }
    }
    context.write(NullWritable.get(), 
                 new IntWritable(max == Integer.MIN_VALUE ? 0 : max));
}
}

Job 2: Find Last Access Per User (TaskGSimple.java:55-85)

public static class UserMaxMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
private final Text outKey = new Text();
private final IntWritable outVal = new IntWritable();

@Override
protected void map(LongWritable key, Text value, Context context) 
    throws IOException, InterruptedException {
    String[] f = CsvUtils.split(value.toString());
    if (f.length >= 5) {
        int by = CsvUtils.toInt(f[1], -1);  // ByWho
        int t = CsvUtils.toInt(f[4], -1);   // ActionTime
        if (by > 0 && t >= 0) {
            outKey.set(String.valueOf(by));
            outVal.set(t);
            context.write(outKey, outVal);
        }
    }
}
}

public static class UserMaxReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) 
    throws IOException, InterruptedException {
    int max = Integer.MIN_VALUE;
    for (IntWritable v : values) {
        if (v.get() > max) {
            max = v.get();
        }
    }
    context.write(key, 
                 new IntWritable(max == Integer.MIN_VALUE ? 0 : max));
}
}

Read Global Max and Calculate Threshold (TaskGSimple.java:180-189):

FileSystem fs = FileSystem.get(conf);
BufferedReader br = new BufferedReader(
new InputStreamReader(fs.open(new Path(args[2] + "/part-r-00000"))));
String line = br.readLine();
br.close();
int globalMax = 0;
if (line != null) {
String[] p = CsvUtils.splitTab(line);
globalMax = CsvUtils.toInt(p[p.length - 1], 0);
}
int threshold = globalMax - 2160;  // 90 days * 24 hours

Job 3: Filter Outdated Users (TaskGSimple.java:87-141)

// Use MultipleInputs to read Pages and UserMax
public static class PageMapper extends Mapper<LongWritable, Text, Text, Text> {
@Override
protected void map(LongWritable key, Text value, Context context) 
    throws IOException, InterruptedException {
    String[] f = CsvUtils.split(value.toString());
    if (f.length >= 2) {
        context.write(new Text(f[0].trim()), 
                     new Text("P," + f[1].trim()));
    }
}
}

public static class MaxMapOutMapper extends Mapper<LongWritable, Text, Text, Text> {
@Override
protected void map(LongWritable key, Text value, Context context) 
    throws IOException, InterruptedException {
    String[] f = CsvUtils.splitTab(value.toString());
    if (f.length >= 2) {
        context.write(new Text(f[0].trim()), 
                     new Text("M," + f[1].trim()));
    }
}
}

public static class OutdatedReducer extends Reducer<Text, Text, NullWritable, Text> {
private int threshold;

@Override
protected void setup(Context context) {
    threshold = context.getConfiguration().getInt("task.g.threshold", 0);
}

@Override
protected void reduce(Text key, Iterable<Text> values, Context context) 
    throws IOException, InterruptedException {
    String nick = null;
    Integer last = null;
    for (Text t : values) {
        String[] p = CsvUtils.split(t.toString());
        if ("P".equals(p[0])) {
            nick = p[1];
        } else if ("M".equals(p[0])) {
            last = CsvUtils.toInt(p[1], -1);
        }
    }
    // Output if user has page but no recent activity
    if (nick != null && (last == null || last < threshold)) {
        context.write(NullWritable.get(), new Text(key + "," + nick));
    }
}
}

Characteristics:

3 sequential jobs with high overhead
Separate job to find global max
Reduce-side join in Job 3
Multiple HDFS reads by driver

Optimized Implementation (2 Jobs)

The optimized version eliminates the global max job and uses map-side join.Job 1: User Max with Combiner (TaskGOptimized.java:25-55)

public static class UserMaxMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
private final Text outKey = new Text();
private final IntWritable outVal = new IntWritable();

@Override
protected void map(LongWritable key, Text value, Context context) 
    throws IOException, InterruptedException {
    String[] f = CsvUtils.split(value.toString());
    if (f.length >= 5) {
        int by = CsvUtils.toInt(f[1], -1);
        int t = CsvUtils.toInt(f[4], -1);
        if (by > 0 && t >= 0) {
            outKey.set(String.valueOf(by));
            outVal.set(t);
            context.write(outKey, outVal);
        }
    }
}
}

public static class MaxReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) 
    throws IOException, InterruptedException {
    int max = Integer.MIN_VALUE;
    for (IntWritable v : values) {
        if (v.get() > max) {
            max = v.get();
        }
    }
    context.write(key, 
                 new IntWritable(max == Integer.MIN_VALUE ? 0 : max));
}
}

// Combiner pre-computes local max per user
job1.setCombinerClass(MaxReducer.class);
job1.setReducerClass(MaxReducer.class);

Combiner for MAX: MAX operation is associative and commutative, making it safe for combiner use. Pre-computing local max per user at each mapper reduces shuffle I/O.

Calculate Global Max in Driver (TaskGOptimized.java:120-133):

// Read Job 1 output and find global max
FileSystem fs = FileSystem.get(conf);
BufferedReader br = new BufferedReader(
new InputStreamReader(fs.open(new Path(args[2] + "/part-r-00000"))));
String line;
int globalMax = 0;
while ((line = br.readLine()) != null) {
String[] p = CsvUtils.splitTab(line);
if (p.length >= 2) {
    int t = CsvUtils.toInt(p[1], 0);
    if (t > globalMax) {
        globalMax = t;
    }
}
}
br.close();
int threshold = globalMax - 2160;

Optimization: Instead of a separate MapReduce job to find global max, the driver scans Job 1 output (only ~200K user records, not 10M activity records). This is faster than launching another job.

Job 2: Map-Side Join Filter (TaskGOptimized.java:57-96)

public static class OutdatedMapJoin extends Mapper<LongWritable, Text, NullWritable, Text> {
private final Map<String, Integer> userLast = new HashMap<String, Integer>();
private int threshold;
private final Text out = new Text();

@Override
protected void setup(Context context) throws IOException {
    threshold = context.getConfiguration().getInt("task.g.threshold", 0);
    
    // Load user last access times from distributed cache
    URI[] files = context.getCacheFiles();
    if (files != null) {
        FileSystem fs = FileSystem.get(context.getConfiguration());
        for (URI u : files) {
            BufferedReader br = new BufferedReader(
                new InputStreamReader(fs.open(new Path(u.getPath()))));
            String line;
            while ((line = br.readLine()) != null) {
                String[] p = CsvUtils.splitTab(line);
                if (p.length >= 2) {
                    userLast.put(p[0].trim(), CsvUtils.toInt(p[1], 0));
                }
            }
            br.close();
        }
    }
}

@Override
protected void map(LongWritable key, Text value, Context context) 
    throws IOException, InterruptedException {
    String[] f = CsvUtils.split(value.toString());
    if (f.length >= 2) {
        String id = f[0].trim();
        String nick = f[1].trim();
        Integer last = userLast.get(id);
        // Output if no activity or last activity before threshold
        if (last == null || last < threshold) {
            out.set(id + "," + nick);
            context.write(NullWritable.get(), out);
        }
    }
}
}

// Configure as map-only job
job2.setNumReduceTasks(0);
job2.addCacheFile(new Path(args[2] + "/part-r-00000").toUri());

Optimization Benefits:

Eliminates Job 1 (global max computed in driver)
Combiner reduces Job 1 shuffle from 10M to ~200K
Map-side join eliminates Job 3 shuffle
Map-only Job 2 saves additional I/O
40-50% faster execution

Performance Comparison

Metric	Simple	Optimized	Improvement
Number of Jobs	3	2	33% reduction
Global Max Calculation	Separate job	In driver	1 job saved
Job 1 Shuffle	10M records	~200K (with combiner)	98% reduction
Join Strategy	Reduce-side	Map-side	Eliminates shuffle
Job 2 Type	Reduce job	Map-only	50% I/O saved
Execution Time	Baseline	40-50% faster	Significant

Running the Task

export JAR=/home/ds503/ds503_bdm-1.0-SNAPSHOT.jar
export PAGES=/circlenet/pages/CircleNetPage.csv
export ACTIVITY=/circlenet/activitylog/ActivityLog.csv
export OUT=/circlenet/output

# Run simple version (3 jobs)
hadoop jar $JAR circlenet.taskG.TaskGSimple \
  $PAGES $ACTIVITY \
  $OUT/taskG/tmp_globalmax_simple \
  $OUT/taskG/tmp_usermax_simple \
  $OUT/taskG/simple

# View results
hdfs dfs -cat $OUT/taskG/simple/part-r-00000 | head -20

Sample Output

12456,InactiveUser1
34567,DormantPage
78901,OldAccount
...

Format: UserID,NickName (users with no activity in last 90 days)

Understanding the 90-Day Calculation

ActionTime granularity: hours
90 days = 90 * 24 hours = 2,160 time units
Threshold = GlobalMaxTime - 2160

Examples:
- GlobalMax = 1,000,000
- Threshold = 997,840
- User last access = 997,500 → Outdated (< threshold)
- User last access = 998,000 → Active (≥ threshold)
- User with no access → Outdated (NULL < threshold)

Key Takeaways

Simple approach: 3 jobs with separate global max computation
Optimization techniques:
- Combiner for MAX: Pre-compute local max (98% shuffle reduction)
- Driver computation: Calculate global max in driver instead of separate job
- Map-side join: Load user timestamps into memory
- Map-only job: Eliminate reduce phase for filtering
Design pattern: When intermediate result is small, compute in driver vs. launching new job
Performance: 40-50% faster with optimized approach
Output filename: Map-only jobs produce part-m-* instead of part-r-*

Get Started

Dataset

Analytics Tasks

Guides

Task G - Outdated Pages

Problem Statement

Approaches

Simple Implementation (3 Jobs)

Optimized Implementation (2 Jobs)

Performance Comparison

Running the Task

Sample Output

Understanding the 90-Day Calculation

Key Takeaways

Build docs developers (and LLMs) love

Get Started

Dataset

Analytics Tasks

Guides

​Problem Statement

​Approaches

​Simple Implementation (3 Jobs)

​Optimized Implementation (2 Jobs)

​Performance Comparison

​Running the Task

​Sample Output

​Understanding the 90-Day Calculation

​Key Takeaways

Build docs developers (and LLMs) love

Problem Statement

Approaches

Simple Implementation (3 Jobs)

Optimized Implementation (2 Jobs)

Performance Comparison

Running the Task

Sample Output

Understanding the 90-Day Calculation

Key Takeaways