Hello all -
I'm trying to run the sample PageView Kafka streams example,
(ref -
https://github.com/apache/kafka/tree/0.11.0/streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview
)
and have a question ..
There is a leftJoin between PageView (Kstream) and UserProfile(Ktable) as
shown below... The join should give - PageViewByRegions
*What key should the join on ? *
*There seems to be no common key (eg. user) between the 2 classes -
PageView and UserProfile*
*Also, what should a sample input data in the PageView & UserProfile
specific topics be ?*
Do we need to add surrogate (id) key to the input for these, to enable the
left join ?
Code :
*------------- static Classes ------------ *
>
> static public class PageView {
> public String user;
> public String page;
> public Long timestamp;
> }
>
> static public class UserProfile {
> public String region;
> public Long timestamp;
> }
*----------Join between the views( i.e. KStream - PageView) & users (KTable
- UserProfile) *
KStream<String, PageViewByRegion> kstream1 =
> *views .leftJoin(users, *
> //PageView - first value type
> //UserProfile - 2nd value type
> //PageViewByRegion - Joined Value
> new ValueJoiner<PageView, UserProfile, PageViewByRegion>() {
> @Override
> public PageViewByRegion apply(PageView view, UserProfile
> profile) {
> PageViewByRegion viewByRegion = new PageViewByRegion();
> viewByRegion.user = view.user;
> viewByRegion.page = view.page;
> System.out.println(" viewByRegion.user " +
> viewByRegion.user);
> System.out.println(" viewByRegion.page " +
> viewByRegion.page);
>
> if (profile != null) {
> viewByRegion.region = profile.region;
> } else {
> viewByRegion.region = "UNKNOWN";
> }
>
> System.out.println(" viewByRegion.page " +
> viewByRegion.region);
>
> return viewByRegion;
> }
> }
> )
