TopWikipediaSessions { public static void run(Options options) { Pipeline p = Pipeline.create(options); double samplingThreshold = 0.1; p.apply(TextIO.read().from(options.getWikiInput())) .apply(MapElements.via(new ParseTableRowJson())) .apply(new ComputeTopSessions(samplingThreshold)) .apply("Write", TextIO.write().to(options.getOutput())); p.run().waitUntilFinish(); }  static void run(Options options); static void main(String[] args);  }
@Test @Category(ValidatesRunner.class) public void testComputeTopUsers() { PCollection<String> output = p.apply( Create.of( Arrays.asList( new TableRow().set("timestamp", 0).set("contributor_username", "user1"), new TableRow().set("timestamp", 1).set("contributor_username", "user1"), new TableRow().set("timestamp", 2).set("contributor_username", "user1"), new TableRow().set("timestamp", 0).set("contributor_username", "user2"), new TableRow().set("timestamp", 1).set("contributor_username", "user2"), new TableRow().set("timestamp", 3601).set("contributor_username", "user2"), new TableRow().set("timestamp", 3602).set("contributor_username", "user2"), new TableRow() .set("timestamp", 35 * 24 * 3600) .set("contributor_username", "user3")))) .apply(new TopWikipediaSessions.ComputeTopSessions(1.0)); PAssert.that(output) .containsInAnyOrder( Arrays.asList( "user1 : [1970-01-01T00:00:00.000Z..1970-01-01T01:00:02.000Z)" + " : 3 : 1970-01-01T00:00:00.000Z", "user3 : [1970-02-05T00:00:00.000Z..1970-02-05T01:00:00.000Z)" + " : 1 : 1970-02-01T00:00:00.000Z")); p.run().waitUntilFinish(); }