Видео

Hi, I haven't but thanks for the tip.

Sinking data from Apache Kafka to an MQTT broker. Do you do anything with either of these?

Hi! We have a Postgres CDC connector that you can use to source data based on changes to the DB and then insert those changes into the Kafka topic. github.com/quixio/quix-samples/tree/main/python/sources/postgres_cdc You can also roll your own connectors. If you can connect to your DB with Python, you can use Quix Streams to pump the data into the topic(s) or if you prefer another language you can use any language you want, but you'll have to know how to build the docker container for it and use a supported Kafka client. If you want to go into more details about your needs drop into our Slack -> quix.io/slack-invite

Thanks for the sub!

Hey Adam. Looks like the team responded in Slack. If you have any more questions or ideas please feel free to pick their brains there :-)

Hi thanks for the question. To do it currently, you'd have to manually seek and commit that offset for each partition. But QuixStreams doesn't support that at the moment.

@@QuixStreams many thanks !

Hi @vijayanandsundaresan3776. Happy to help. Could you please come over to the community channel here join.slack.com/t/stream-processing/shared_invite/zt-2lxwg3a0l-6NIZuvkFVKrm6UTH_97iSA. We'll be on hand to help you solve the problem in the #quix-help channel.

I see, because the producer hasn't sent messages, with 'auto_offset_rest=latest' will only read new messages sent after it starts. 'earliest' will work, but after consuming the messages once, running script again, no messages obtained..

@@yifeitong same issue here. found any solution?

@@sridharstreakssri try auto_offset_reset earliest ?

There's a problem with the connection to the Kafka broker. This can be due to incorrect broker address, network issues, or broker unavailability. Check out the docs here: quix.io/docs/manage/troubleshooting.html#data-is-not-being-received-into-a-topic Or ask the community for help here join.slack.com/t/stream-processing/shared_invite/zt-2lxwg3a0l-6NIZuvkFVKrm6UTH_97iSA

Thank fo watching. Please share our content on X and Twitter to spread the word!

Thanks @Apex543. Please share on socials to help boost the numbers!

It's Neovim with some custom configuration. We recommend getting started with the Neovim Kickstart project and tailor to your preferences

Thank you for sharing. That is a great use case for explode! If you partition and produce using different keys (provide a key in sdf.to_topic()), you could run multiple consumers downstream to process in parallel. This depends on if you care about ordering (perhaps you do for search results unless the result’s rank is encoded). You may also care about exactly-once processing guarantees, because if there’s failure, this avoids events being reprocessed. In general when processing large result sets I recommend leveraging state

@@DataSurfer yes you are absolutely correct about processing once! Do you guys have a community slack we could have a chat on?

​@@vikramtatke5930 Yes, you can come speak to the whole team on Slack. The link is here (and also in the video description): quix.io/slack-invite

@@DataSurfer that's great thanks a lot!! :) if it's alright for me to suggest, you could make a video on this too :)

@@vikramtatke5930 You're more than welcome to make suggestions :) However, it would be more fun if you could share some more details and we build out the solution together. Let's chat in Slack!

Thank you! The main difference is that the Producer object is a low-level object with no guarantees. Using StreamingDataFrame provides automatic management of resilient and durable application state with checkpoints and changelogs. This means you can provide exactly-once processing guarantees for the items in the array if required. For further reading, check the docs for "Stateful Applications"

@@DataSurfer For sure! Thanks a lot!

We’ve just released a new Sinks API. The release has a CSV Sink connector included. Pls check out the docs

@@michaelrosam9271 That sounds great! Thanks!

If you haven't already seen it check out the video about the 2.9 release. ruclips.net/video/VoDQtO8mirc/видео.html

Have you got anything specific you want us to explain or demonstrate?

Thank you!

Thanks! Great idea, I'll add it to the list.

Hi, Expand is used when you want to return more than one 'row' from your processing function. The function will receive 1 row or data, then you might decide that should represent 2 or more. Maybe you received a binary value that you have decoded and it contains more data. Or you have received a json payload that actually represents more than 1 row. So you want to unpack this into the individual rows. Another example, if you have a sentence and you want to process each word individually, you can use expand=True to split the sentence into words and treat each word as a separate event/row. Then in the next lines of code or in another service you can process the, new, individual rows as needed. If that's still not clear, happy to jump on a call or do another video focusing on this. Let me know.

@@QuixStreams Yeah yeah I got it. So the messages might be nested and we need to kinda unpack it and we can do that via the expand=True right?

That's correct yes.

QuixStreams! sdf.run listens for the termination signals and keeps the listeners open for business.

It should just work automatically. The consumer can detect the compression type when it reads each batch and decompress it on the fly. The only thing you need to ensure is support. So if your producers compress with, say, `lz4`, you need to make sure all your consumer libraries support reading `lz4`. With a very common algorithm like `gzip`, that's pretty much guaranteed. For the more exotic ones, I'd double check before you ship.

@@QuixStreams Thanks for the response. That is a good design choice from Kafka, especially if all your components are using the same quixstreams library as the compression support should be the same.

Absolutely. 👍

Yup, that's absolutely true. However, if you were parallelising, you might decide that it's better to have 3 Python producers sharing a workload instead of introducing an extra component like Camel. And with these tips, you might be looking at 3 producers instead of 10. 🙂 (That's the shape of it at least. I'm not actually sure how you'd load-balance this specific data-source, but I thought it was an interesting one to explore the concepts with. 😉)

Glad you enjoyed it! There's a video here that goes into performing realtime stream transformations with a panda-like API: ruclips.net/video/5sqegy_EPa0/видео.html It's note exactly the same as Pandas, but it the underlying data's stored in Kafka it'll probably solve your problem in the same way. 🙂

I go through how to migrate Pandas code to Quix Streams code in a video. Hopefully it provides some inspiration: ruclips.net/video/FC-DrNbe5fA/видео.html

Yes sir, will do! We actually have many more in the pipeline. Subscribe and stay tuned!

Will do. Watch this space.

Quix Streams is a library, not a server-side engine like Spark. It uses container orchestration via Kubernetes to achieve the same horizontal scale as Spark. The benefit being it is easy to develop and debug a Quix Streams application. You can deploy it how you like, but the best way is in a docker container in K8’s

Any transformation you can do on top of the data

Yes that's correct. It could be determining a new value based on existing values in the data set or some average based on previous values that you have in state. Really any processing you can achieve using Python.

We noticed you've joined the Quix Community on Slack where Tomas has helped you with a solution for stream joins. For future reference to anyone who would like support, please join us on Slack: quix.io/slack-invite

Thank you 😊

Thanks 👍

Hi, this code will run forever, unless there is an issue like the machine it's running on being turned off, it's also not production code so adding some error handling etc would certainly be a good idea. If you want to run this in production with resiliency and SLA's check out Quix Cloud -> quix.io/signup