Kafka

-----BEGIN PGP SIGNED MESSAGE----- Hash: SHA512 Are you aware of KIP-557: https://cwiki.apache.org/confluence/display/KAFKA/KIP-557%3A+Add+emit+on +change+support+for+Kafka+Streams Seems it will address your use case? - -Matthias On 2/25/20 6:45 PM, Adam Rinehart wrote: > Bruno and Guozhang, > > Thank you for the replies. Between the 2 of you, I think I know how > to code what I wanted. I'm going with > > stream.flatTransform(...).groupByKey().aggregate() > > because an additional requirement that I hadn't stated in the > original message was I was planning on using a punctuate method to > delay some values from being forwarded until certain time-outs or > conditions were met. I can't use punctuate with the ...Values() > versions, per the documentation. > > Basically, I'm going to be transforming the events from one of many > change events to a Success, Fail, or Lost message. But I only want...

Bruno and Guozhang, Thank you for the replies. Between the 2 of you, I think I know how to code what I wanted. I'm going with stream.flatTransform(...).groupByKey().aggregate() because an additional requirement that I hadn't stated in the original message was I was planning on using a punctuate method to delay some values from being forwarded until certain time-outs or conditions were met. I can't use punctuate with the ...Values() versions, per the documentation. Basically, I'm going to be transforming the events from one of many change events to a Success, Fail, or Lost message. But I only want to report actual state changes, so if I get multiple Fail messages, I only need to see the first one. So I'm going to use a persistent KeyValueStore to store a tracking object when a given key comes in, and check the table to see if I'm going to actually emit a change. In the punctuate method, it will iterate over the persistent KeyValueStore to se...

Hi, This is really getting interesting. Now if we don't want a record to be emitted downstream only way we can do is via transform or (flatTransform). Since we are now reverting the fix for null record in transformValues and rather change the docs, doesn't this add bit of confusion for users. Confluent docs says that: transformValues is preferable to transform because it will not cause data re-partitioning. So in many cases if just the record's value structure is sufficient to determine whether we should emit it downstream or not, we would still be forced to use transform and unnecessarily cause data re-partitioning. Won't this be in-efficient. Thanks Sachin On Tue, Feb 25, 2020 at 10:52 PM Bruno Cadonna < bruno@confluent.io > wrote: > Hello Guozhang and Adam, > > Regarding Guozhang's proposal please see recent discussions about > `transformValues()` and returning `null` from the transformer: > > https://is...

Hello Guozhang and Adam, Regarding Guozhang's proposal please see recent discussions about `transformValues()` and returning `null` from the transformer: https://issues.apache.org/jira/browse/KAFKA-9533?focusedCommentId=17044602&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-17044602 . With the current behavior, the commands should be: `stream.transformValues(...).filter((k,v) -> return v != null).groupByKey().aggregate()` Best, Bruno On Tue, Feb 25, 2020 at 2:58 AM Guozhang Wang < wangguoz@gmail.com > wrote: > > Hello Adam, > > It seems your intention is to not "avoid emitting if the new aggregation > result is the same as the old aggregation" but to "avoid processing the > aggregation at all if it state is already some certain value", right? > > In this case I think you can try sth. like this: > > *stream.transformValues().groupByKey().aggregate()* > ...

you can use kafka-console-consumer that comes with your kafka deployment, or you can install kafkacat (which I found more simple to use) brew install kafkacat kafkacat -b your.broker.com:yourPORT -t yourtopic -c max-messages On Tue, Feb 25, 2020 at 9:03 AM Doaa K. Amin <doaaelkordy@yahoo.com.invalid> wrote: > > Hello, > I'm new to kafka and I'd like to write data from kafka to a CSV file in a Mac. Please, advise. > Thank You & Kindest Regards,Doaa. -- Richard Rossel Atlanta - GA

Hello, I'm new to kafka and I'd like to write data from kafka to a CSV file in a Mac. Please, advise. Thank You & Kindest Regards,Doaa.

Hello Adam, It seems your intention is to not "avoid emitting if the new aggregation result is the same as the old aggregation" but to "avoid processing the aggregation at all if it state is already some certain value", right? In this case I think you can try sth. like this: *stream.transformValues().groupByKey().aggregate()* where transformValues is just used as a slight complicated "filter" operation, in which you can access the state store that "aggregate" is connected to, and read / check if the corresponding entry is already `success`, if yes let `transformValue` to return `null` which means forward nothing to the downstream. The reason to use transformValues instead of transform is to make sure you do not introduce unnecessary repartitioning here. Guozhang On Mon, Feb 24, 2020 at 2:01 PM Adam Rinehart < adam.rinehart@gmail.com > wrote: > So I am trying to process incoming events, that may or may ...