Hi,
In our system, we have multiple data generators creating file content in a
shared file system, with an indication of the DataSourceId in the filename.
The need is to have a fair scheduling mechanism to read files generated by
all sources, parse, flatten & enrich (using Reference Data) the data
records in the files, batch the enriched records and write into database.
I use the IPartitionedTridentSpout. The topology looks like this:
TransactionalTridentEsrSpout spout = new
TransactionalTridentEsrSpout(NUM_OF_PARTITIONS);
TridentTopology topology = *new* TridentTopology();
topology.newStream("FileHandlerSpout", spout)
.each(*new* Fields("filename", "esr"), *new*Utils.PrintFilter())
.parallelismHint(NUM_OF_PARTITIONS)
.shuffle()
.each(*new* Fields("filename", "record"),
*new*RecordFlattenerAndEnricher(),
*new* elds("record-enriched"))
.each(*new* Fields("filename", "record-enriched"),
*new*Utils.PrintFilter())
.project(*new* Fields(record-enriched")) // pass only required
fields
.parallelismHint(PARALLELISM_HINT_FOR_ESR_FLATTENER_ENRICHER)
.shuffle()
.aggregate(*new* Fields("record-enriched"),
*new*BlockWriterToDb(),
*new* Fields("something"))
.each(*new* Fields("something"), *new* Utils.PrintFilter())
.parallelismHint(PARALLELISM_HINT_FOR_GP_WRITER);
Since the data files are very big (usually 1 million records), I read small
batches of 10K records. For every transactionId generated by Coordinaor, my
Emitter emits the next 10K records of the current/next file in its
partition. The final BlockWriter would aggregate enriched records into a
buffer and on "complete" method call, will write the buffer to DB.
The topology works fine, but I have the following question:
The parallelismHint for the ParttionedTridentSpout, which impacts the
number of Emitters, is set to the number of partitions. The parallelismHint
for the next two layers (the FlattenerAndEnricher and BlockWriterToDb)
needs to be set to higher values because we have lot of work to be done
there. Since there is no groupBy need here, I use Shuffle() between all
stages here. When a particular downstream bolt dies, Trident is expected to
call the Emitter with appropriate old metadata asking it to re-emit. But
since a shuffle has happened, the specific records that were part of one
Emitter's emit would have landed in multiple downstream bolts. So, how can
Trident call the appropriate emitters for re-emit so the exact same records
are re-emitted. Even if Trident calls the appropriate emitters, the Emitter
is going to re-emit the whole 10K batch, out of which some records have
only failed. How is this whole sequence handled by Storm and how do we
design the application logic here to handle fault tolerance for "exactly
once" semantics
Thanks & Regards
MK
