Let's assume that I have a few streams defined like:
Dataset<Row> allMessages = df.select(.....)
Dataset<Row> messagesOfType1 = df.select() //some unique conditions
Dataset<Row> messagesOfType2 = df.select() //some other unique conditions
and some logic for processing like:
StreamingQuery firstQuery = messagesOfType1
.writeStream()
.foreach(new CustomForEachWiriter(ConfigurationForType1())) // class that extends ForeachWriter[T] and save data into external RDBMS table
.start();
StreamingQuery secondQuery = messagesOfType2
.writeStream()
.foreach(new CustomForEachWiriter(ConfigurationForType2())) // class that extends ForeachWriter[T] and save data into external RDBMS table (may be even another database than before)
.start();
As I mentioned above CustomForEachWriter is responsible for message processing and storing into RDBMS. In the simplest approach I can manage db connection inside dedicated method implemented in CustomForEachWriter.
But I'm not really fan of this approach so I'm wondering how to introduce connection pooling. Usually Connection and DataSource doesn't implement Serializable interface so I can't simply pass Connection retrieved from connection pools on driver (exception TaskNotSerialized will be thrown). It also looks that when new data arrives new object of CustomEachWriter is created.
Could you please explain to me what is the right approach in such cases for structured streaming ?
Thank you
Kind regards