Page History

final RemoteTableDescriptor outputTableDesc = new RemoteTableDescriptor<Integer, EnrichedPageView, EnrichedPageView>("enriched-page-view-table-1");
 
final Table<KV<Integer, Profile>> joinTable = appDesc.
   getTable(outputTableDesc);
 
appDesc.getInputStream(isd)
 .map(pv -> new KV<>(pv.getMemberId(), pv))
 .join(joinTable, new PageViewToProfileJoinFunction())
 .map(m -> new KV(m.getMemberId(), UpdatePairUpdateMessage.of(m, m)))
 .sendUpdateTo(outputTable);