You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Gaspar Muñoz <gm...@stratio.com> on 2016/05/04 14:43:54 UTC
Kafka Streams: KStream - KTable Left Join
Hi there,
I am not able to perform a Left Join between a KStream and KTable in Kafka
Streams.
Exception in thread "main"
org.apache.kafka.streams.errors.TopologyBuilderException: Invalid topology
building: KSTREAM-FILTER-0000000003 and KTABLE-SOURCE-0000000005 are not
joinable
at
org.apache.kafka.streams.kstream.internals.AbstractStream.ensureJoinableWith(AbstractStream.java:44)
at
org.apache.kafka.streams.kstream.internals.KStreamImpl.leftJoin(KStreamImpl.java:383)
My code looks
KStream<String, String> dataStream = builder
.stream(stringDeserializer, stringDeserializer,
conf.getString(INPUT_TOPIC))
.map(App::keyByAap)
KTable<String, String> aapTable = builder.table("lookup_topic");
KStream<String, String> result = dataStream.leftJoin(aapTable, new
ValueJoinerAap());
My ValueJoiner simply concat two Strings. I think the problem is in map
where I change the key in order to join by this field. In the internals of
KafkaStreams we can see that map function set to null source nodes
@Override
public <K1, V1> KStream<K1, V1> map(KeyValueMapper<K, V, KeyValue<K1,
V1>> mapper) {
String name = topology.newName(MAP_NAME);
topology.addProcessor(name, new KStreamMap<>(mapper), this.name);
return new KStreamImpl<>(topology, name, null);
}
And after that, in the left join precondition, in function Set<String>
allSourceNodes = ensureJoinableWith((AbstractStream<K>) other); the code
checks if any source nodes is null and, obviously, the KStream which I
applied the map has the source to null so throw the exception.
if (thisSourceNodes == null || otherSourceNodes == null)
throw new TopologyBuilderException(this.name + " and " +
other.name + " are not joinable");
The question is, how can I change the key without lose parent data in order
to perform a join with KTable after that?
Thanks.
--
Gaspar Muñoz
Vía de las dos Castillas, 33, Ática 4, 3ª Planta
28224 Pozuelo de Alarcón, Madrid
Tel: +34 91 828 6473 // www.stratio.com // *@stratiobd
<https://twitter.com/StratioBD>*
Re: Kafka Streams: KStream - KTable Left Join
Posted by Gaspar Muñoz <gm...@stratio.com>.
Thanks, I'll try it. It would be nice if you could improve the error
message.
2016-05-04 20:13 GMT+02:00 Matthias J. Sax <ma...@confluent.io>:
> +1
>
> I had the same thought and put it on my personal agenda already.
>
> -Matthias
>
> On 05/04/2016 06:37 PM, Jay Kreps wrote:
> > Is it possible to make the error message give more an explanation?
> >
> > -Jay
> >
> > On Wed, May 4, 2016 at 8:46 AM, Matthias J. Sax <ma...@confluent.io>
> > wrote:
> >
> >> Hi,
> >>
> >> I am still new to Kafka Streams by myself, but from my understanding if
> >> you change the key, your partitioning changes, ie, is not valid anymore.
> >> Thus, the joins (which assumes co-located data) cannot be performed
> >> (this is the reason why sources get set to null). You can write to an
> >> intermediate topic via .through(...) to get a valid partitioning:
> >>
> >> KStream dataStream = builder.stream(...).map(...).through(...);
> >>
> >> Afterward, your join should work.
> >>
> >> -Matthias
> >>
> >>
> >> On 05/04/2016 04:43 PM, Gaspar Muñoz wrote:
> >>> Hi there,
> >>>
> >>> I am not able to perform a Left Join between a KStream and KTable in
> >> Kafka
> >>> Streams.
> >>>
> >>> Exception in thread "main"
> >>> org.apache.kafka.streams.errors.TopologyBuilderException: Invalid
> >> topology
> >>> building: KSTREAM-FILTER-0000000003 and KTABLE-SOURCE-0000000005 are
> not
> >>> joinable
> >>> at
> >>>
> >>
> org.apache.kafka.streams.kstream.internals.AbstractStream.ensureJoinableWith(AbstractStream.java:44)
> >>> at
> >>>
> >>
> org.apache.kafka.streams.kstream.internals.KStreamImpl.leftJoin(KStreamImpl.java:383)
> >>>
> >>>
> >>> My code looks
> >>>
> >>> KStream<String, String> dataStream = builder
> >>> .stream(stringDeserializer, stringDeserializer,
> >>> conf.getString(INPUT_TOPIC))
> >>> .map(App::keyByAap)
> >>>
> >>> KTable<String, String> aapTable = builder.table("lookup_topic");
> >>>
> >>> KStream<String, String> result = dataStream.leftJoin(aapTable, new
> >>> ValueJoinerAap());
> >>>
> >>>
> >>> My ValueJoiner simply concat two Strings. I think the problem is in
> map
> >>> where I change the key in order to join by this field. In the internals
> >> of
> >>> KafkaStreams we can see that map function set to null source nodes
> >>>
> >>> @Override
> >>> public <K1, V1> KStream<K1, V1> map(KeyValueMapper<K, V, KeyValue<K1,
> >>> V1>> mapper) {
> >>> String name = topology.newName(MAP_NAME);
> >>>
> >>> topology.addProcessor(name, new KStreamMap<>(mapper), this.name);
> >>>
> >>> return new KStreamImpl<>(topology, name, null);
> >>> }
> >>>
> >>>
> >>> And after that, in the left join precondition, in function Set<String>
> >>> allSourceNodes = ensureJoinableWith((AbstractStream<K>) other); the
> code
> >>> checks if any source nodes is null and, obviously, the KStream which I
> >>> applied the map has the source to null so throw the exception.
> >>>
> >>> if (thisSourceNodes == null || otherSourceNodes == null)
> >>> throw new TopologyBuilderException(this.name + " and " +
> >>> other.name + " are not joinable");
> >>>
> >>>
> >>> The question is, how can I change the key without lose parent data in
> >> order
> >>> to perform a join with KTable after that?
> >>>
> >>> Thanks.
> >>>
> >>
> >>
> >
>
>
--
Gaspar Muñoz
Vía de las dos Castillas, 33, Ática 4, 3ª Planta
28224 Pozuelo de Alarcón, Madrid
Tel: +34 91 828 6473 // www.stratio.com // *@stratiobd
<https://twitter.com/StratioBD>*
Re: Kafka Streams: KStream - KTable Left Join
Posted by "Matthias J. Sax" <ma...@confluent.io>.
+1
I had the same thought and put it on my personal agenda already.
-Matthias
On 05/04/2016 06:37 PM, Jay Kreps wrote:
> Is it possible to make the error message give more an explanation?
>
> -Jay
>
> On Wed, May 4, 2016 at 8:46 AM, Matthias J. Sax <ma...@confluent.io>
> wrote:
>
>> Hi,
>>
>> I am still new to Kafka Streams by myself, but from my understanding if
>> you change the key, your partitioning changes, ie, is not valid anymore.
>> Thus, the joins (which assumes co-located data) cannot be performed
>> (this is the reason why sources get set to null). You can write to an
>> intermediate topic via .through(...) to get a valid partitioning:
>>
>> KStream dataStream = builder.stream(...).map(...).through(...);
>>
>> Afterward, your join should work.
>>
>> -Matthias
>>
>>
>> On 05/04/2016 04:43 PM, Gaspar Muñoz wrote:
>>> Hi there,
>>>
>>> I am not able to perform a Left Join between a KStream and KTable in
>> Kafka
>>> Streams.
>>>
>>> Exception in thread "main"
>>> org.apache.kafka.streams.errors.TopologyBuilderException: Invalid
>> topology
>>> building: KSTREAM-FILTER-0000000003 and KTABLE-SOURCE-0000000005 are not
>>> joinable
>>> at
>>>
>> org.apache.kafka.streams.kstream.internals.AbstractStream.ensureJoinableWith(AbstractStream.java:44)
>>> at
>>>
>> org.apache.kafka.streams.kstream.internals.KStreamImpl.leftJoin(KStreamImpl.java:383)
>>>
>>>
>>> My code looks
>>>
>>> KStream<String, String> dataStream = builder
>>> .stream(stringDeserializer, stringDeserializer,
>>> conf.getString(INPUT_TOPIC))
>>> .map(App::keyByAap)
>>>
>>> KTable<String, String> aapTable = builder.table("lookup_topic");
>>>
>>> KStream<String, String> result = dataStream.leftJoin(aapTable, new
>>> ValueJoinerAap());
>>>
>>>
>>> My ValueJoiner simply concat two Strings. I think the problem is in map
>>> where I change the key in order to join by this field. In the internals
>> of
>>> KafkaStreams we can see that map function set to null source nodes
>>>
>>> @Override
>>> public <K1, V1> KStream<K1, V1> map(KeyValueMapper<K, V, KeyValue<K1,
>>> V1>> mapper) {
>>> String name = topology.newName(MAP_NAME);
>>>
>>> topology.addProcessor(name, new KStreamMap<>(mapper), this.name);
>>>
>>> return new KStreamImpl<>(topology, name, null);
>>> }
>>>
>>>
>>> And after that, in the left join precondition, in function Set<String>
>>> allSourceNodes = ensureJoinableWith((AbstractStream<K>) other); the code
>>> checks if any source nodes is null and, obviously, the KStream which I
>>> applied the map has the source to null so throw the exception.
>>>
>>> if (thisSourceNodes == null || otherSourceNodes == null)
>>> throw new TopologyBuilderException(this.name + " and " +
>>> other.name + " are not joinable");
>>>
>>>
>>> The question is, how can I change the key without lose parent data in
>> order
>>> to perform a join with KTable after that?
>>>
>>> Thanks.
>>>
>>
>>
>
Re: Kafka Streams: KStream - KTable Left Join
Posted by Jay Kreps <ja...@confluent.io>.
Is it possible to make the error message give more an explanation?
-Jay
On Wed, May 4, 2016 at 8:46 AM, Matthias J. Sax <ma...@confluent.io>
wrote:
> Hi,
>
> I am still new to Kafka Streams by myself, but from my understanding if
> you change the key, your partitioning changes, ie, is not valid anymore.
> Thus, the joins (which assumes co-located data) cannot be performed
> (this is the reason why sources get set to null). You can write to an
> intermediate topic via .through(...) to get a valid partitioning:
>
> KStream dataStream = builder.stream(...).map(...).through(...);
>
> Afterward, your join should work.
>
> -Matthias
>
>
> On 05/04/2016 04:43 PM, Gaspar Muñoz wrote:
> > Hi there,
> >
> > I am not able to perform a Left Join between a KStream and KTable in
> Kafka
> > Streams.
> >
> > Exception in thread "main"
> > org.apache.kafka.streams.errors.TopologyBuilderException: Invalid
> topology
> > building: KSTREAM-FILTER-0000000003 and KTABLE-SOURCE-0000000005 are not
> > joinable
> > at
> >
> org.apache.kafka.streams.kstream.internals.AbstractStream.ensureJoinableWith(AbstractStream.java:44)
> > at
> >
> org.apache.kafka.streams.kstream.internals.KStreamImpl.leftJoin(KStreamImpl.java:383)
> >
> >
> > My code looks
> >
> > KStream<String, String> dataStream = builder
> > .stream(stringDeserializer, stringDeserializer,
> > conf.getString(INPUT_TOPIC))
> > .map(App::keyByAap)
> >
> > KTable<String, String> aapTable = builder.table("lookup_topic");
> >
> > KStream<String, String> result = dataStream.leftJoin(aapTable, new
> > ValueJoinerAap());
> >
> >
> > My ValueJoiner simply concat two Strings. I think the problem is in map
> > where I change the key in order to join by this field. In the internals
> of
> > KafkaStreams we can see that map function set to null source nodes
> >
> > @Override
> > public <K1, V1> KStream<K1, V1> map(KeyValueMapper<K, V, KeyValue<K1,
> > V1>> mapper) {
> > String name = topology.newName(MAP_NAME);
> >
> > topology.addProcessor(name, new KStreamMap<>(mapper), this.name);
> >
> > return new KStreamImpl<>(topology, name, null);
> > }
> >
> >
> > And after that, in the left join precondition, in function Set<String>
> > allSourceNodes = ensureJoinableWith((AbstractStream<K>) other); the code
> > checks if any source nodes is null and, obviously, the KStream which I
> > applied the map has the source to null so throw the exception.
> >
> > if (thisSourceNodes == null || otherSourceNodes == null)
> > throw new TopologyBuilderException(this.name + " and " +
> > other.name + " are not joinable");
> >
> >
> > The question is, how can I change the key without lose parent data in
> order
> > to perform a join with KTable after that?
> >
> > Thanks.
> >
>
>
Re: Kafka Streams: KStream - KTable Left Join
Posted by "Matthias J. Sax" <ma...@confluent.io>.
Hi,
I am still new to Kafka Streams by myself, but from my understanding if
you change the key, your partitioning changes, ie, is not valid anymore.
Thus, the joins (which assumes co-located data) cannot be performed
(this is the reason why sources get set to null). You can write to an
intermediate topic via .through(...) to get a valid partitioning:
KStream dataStream = builder.stream(...).map(...).through(...);
Afterward, your join should work.
-Matthias
On 05/04/2016 04:43 PM, Gaspar Muñoz wrote:
> Hi there,
>
> I am not able to perform a Left Join between a KStream and KTable in Kafka
> Streams.
>
> Exception in thread "main"
> org.apache.kafka.streams.errors.TopologyBuilderException: Invalid topology
> building: KSTREAM-FILTER-0000000003 and KTABLE-SOURCE-0000000005 are not
> joinable
> at
> org.apache.kafka.streams.kstream.internals.AbstractStream.ensureJoinableWith(AbstractStream.java:44)
> at
> org.apache.kafka.streams.kstream.internals.KStreamImpl.leftJoin(KStreamImpl.java:383)
>
>
> My code looks
>
> KStream<String, String> dataStream = builder
> .stream(stringDeserializer, stringDeserializer,
> conf.getString(INPUT_TOPIC))
> .map(App::keyByAap)
>
> KTable<String, String> aapTable = builder.table("lookup_topic");
>
> KStream<String, String> result = dataStream.leftJoin(aapTable, new
> ValueJoinerAap());
>
>
> My ValueJoiner simply concat two Strings. I think the problem is in map
> where I change the key in order to join by this field. In the internals of
> KafkaStreams we can see that map function set to null source nodes
>
> @Override
> public <K1, V1> KStream<K1, V1> map(KeyValueMapper<K, V, KeyValue<K1,
> V1>> mapper) {
> String name = topology.newName(MAP_NAME);
>
> topology.addProcessor(name, new KStreamMap<>(mapper), this.name);
>
> return new KStreamImpl<>(topology, name, null);
> }
>
>
> And after that, in the left join precondition, in function Set<String>
> allSourceNodes = ensureJoinableWith((AbstractStream<K>) other); the code
> checks if any source nodes is null and, obviously, the KStream which I
> applied the map has the source to null so throw the exception.
>
> if (thisSourceNodes == null || otherSourceNodes == null)
> throw new TopologyBuilderException(this.name + " and " +
> other.name + " are not joinable");
>
>
> The question is, how can I change the key without lose parent data in order
> to perform a join with KTable after that?
>
> Thanks.
>