Uploding data to kafka producer-CodePudding

I am new to Kafka in Spring Boot, I have been through many tutorials and got fair knowledge about the same.

Currently I have been assigned a task and I am facing an issue. Hope to get some help here.
The scenario is as follows.
1)I have a DB which is getting updated continuously with millions of data.
2)I have to hit the DB after every 5 mins and pick the recently updated data and send it to Kafka.

Condition- The old data that I have picked in my previous iteration should not be picked in my next DB call and Kafka pushing.

I am done with the part of Spring Scheduling to pick the data by using findAll() of spring boot JPA, but how can I write the logic so that it does not pick the old DB records and just take the new record and push it to kafka.

My DB table also have a field called "Recent_timeStamp" of type "datetime"

CodePudding user response：

Its hard to tell without really seeing your logic and the way you work with the database, but from what you've described you should do just "findAll" here. Instead you should treat your DB table as a time-driven data:

Since it has a field of timestamp, make sure there is an index on it
Instead of "findAll" execute something like:

   SELECT <...>
   FROM <YOUR_TABLE>
   WHERE RECENT_TIMESTAMP > ?
   ORDER BY RECENT_TIMESTAMP ASC

In this case you'll get the records ordered by the increasing timestamp Now the ? denotes the last memorized timestamp that you've handled

So you'll have to maintain the state here

Another option is to query the data whose timestamp is "less" than 5 minutes, in this case the query will look like this (pseudocode since the actual syntax varies):

   SELECT <...>
   FROM <YOUR_TABLE>
   WHERE RECENT_TIMESTAMP < now() - 5 minutes
   ORDER BY RECENT_TIMESTAMP ASC

The first method is more robust because if your spring boot application is "down" for some reason you'll be able to recover and query all your records from the point it has failed to send the data. On the other hand you'll have to save this kind of pointer in some type of persistent storage. The second solution is "easier" in a sense that you don't have a state to maintain but on the other hand you will miss the data after the restart.

In both of the cases you might want to use some kind of pagination because basically you don't know how many records you'll get from the database and if the amount of records exceeds your memory limits, the application with end up with OutOfMemory error thrown.

A Completely different approach is throwing the data to kafka when you write to the database instead of when you read from it. At that point you might have a data chunk of (probably) reasonably limited size and in general you don't need the state because you can store to db and send to kafka from the same service, if the architecture of your application permits to do so.

CodePudding user response：

You can look into kafka connect component if it serves your purpose.

Kafka Connect is a tool for scalably and reliably streaming data between Apache Kafka® and other data systems. It makes it simple to quickly define connectors that move large data sets in and out of Kafka. Kafka Connect can ingest entire databases or collect metrics from all your application servers into Kafka topics, making the data available for stream processing with low latency. An export connector can deliver data from Kafka topics into secondary indexes like Elasticsearch, or into batch systems–such as Hadoop for offline analysis.