You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@camel.apache.org by Michele <mi...@finconsgroup.com> on 2016/03/28 11:15:27 UTC
Best Strategy to process a large number of rows in File
Hi everyone,
a business requirement of my project is to process a file from directory,
split and store each single row in Active MQ.
Then a consumer will consume the queue invoking a Rest Service by jetty
client.
So, there is a little problem... when the file csv has many lines (~50000
rows) to process.
I already optimized my route following this topic
http://www.davsclaus.com/2011/11/splitting-big-xml-files-with-apache.html in
order to process quickly the file with split and store in AMQ.
But, now i need to process messages stored in AMQ slowly in order to don't
overload a Rest Service Interface... Ex. n messages at time each five
minutes
Could you please provide me the best strategy to do it or example or
suggestions to study?
I'm seeing ThrottlingInflightRoutePolicy and AMQ Parameter Configuration but
i don't know if this is right way.
I work with JBossFuse 6.2.0 (Camel 2.15.1 and Active MQ 5.11)
Thanks in advance
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi Jens,
exactly, approximately 1.6 msg/s.
The processes responsible for reading the queue is single-threaded??
No, I defined a PooledConnectionFactory with concurrentConsumers 10 and 8
max connections, limited with prefetchSize of 10 to avoid OOM.
In addition, I added a custom parallelProcessing on splitter to speed up
defined like this :
<bean id="myThreadPoolExecutor"
class="java.util.concurrent.ThreadPoolExecutor">
<constructor-arg index="0" value="3"/>
<constructor-arg index="1" value="6"/>
<constructor-arg index="2" value="0"/>
<constructor-arg index="3" value="MILLISECONDS"/>
<constructor-arg index="4"><bean
class="java.util.concurrent.LinkedBlockingQueue"/></constructor-arg>
</bean>
More than 3 => Out Of Memory.
I also defined a throttle policy (to avoid overlaod on RS Service) on route
that consumes the queue but it seems that it is never reached.
<route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
<from uri="activemq:queue:CBIKIT?destination.consumer.prefetchSize=10" />
<throttle timePeriodMillis="1000" asyncDelayed="true">
<constant>5</constant>
<enrich uri="direct:crm-login" strategyRef="OAuthStrategy" />
<setHeader
headerName="Content-Type"><constant>application/json</constant></setHeader>
<setHeader
headerName="CamelHttpMethod"><constant>POST</constant></setHeader>
<setHeader headerName="CompleteActionPath">
<simple>jetty:{{crm.rs.host.name}}:{{crm.rs.port}}{{crm.rs.context.path}}{{crm.rs.serialnumber.path}}?httpClient.idleTimeout=30000&jettyHttpBindingRef=CustomJettyHttpBinding</simple>
</setHeader>
<log message="Sending SerialNumber from CBIKIT to CRT -
${in.header.CamelSplitIndex}!" loggingLevel="DEBUG" />
<log message="Sending SerialNumber from CBIKIT to CRT
${in.header.CamelSplitIndex} - ${body}!" />
<recipientList>
<header>CompleteActionPath</header>
</recipientList>
<log message="Message Received (JSON Format)
${in.header.CamelSplitIndex}: ${body}" />
<unmarshal ref="Gson" />
<choice>
<when>
<simple>${in.header.Result} == 'ERROR'</simple>
<log message="CRM Response: ${body}" loggingLevel="ERROR"/>
<bean ref="FailureMessageBean" method="enrichByCRMError" />
<to uri="direct:messageDiscarded"/>
</when>
</choice>
</throttle>
</route>
Current Active MQ configuration (I'm studying Active MQ guide to improve
performance):
<policyEntry queue="CBIKIT" producerFlowControl="true"
optimizedDispatch="true" reduceMemoryFootprint="true" memoryLimit="256mb" >
</policyEntry>
PersistenceAdapter
<kahaDB directory="${data}/kahadb" indexCacheSize="40000"
indexWriteBatchSize="4000" journalMaxFileLength="512mb"/>
Thanks a lot for your support.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781173.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
"When I log a message that contains 1000 lines I see
line\rline2\rline3\r...line999
The split works when token is \\r. Why?"
I'm not positive but it appears that something in the marshaling is
escaping the backslash on the \r so when it writes it out you get \\r but
I'm not sure. You could try this as a test:
<tokenize token="\r" group="1000" regex="true"/>
<transform>
<simple>${body}.split("\r")</simple>
</transform>
<to uri="activemq:queue:Cbikit.Key" />
I'm using the Java String split mechanics there and I'm not sure that
syntax works in that place but there is an equivalent if it doesn't. That
should send a String[] across that then you can just use.
<split streaming="true">
<to uri="sedaQueue:queue.ReadyToProcess?blockWhenFull=true" />
</split>
Which should simply understand it is splitting an array at that point.
On Tue, Apr 26, 2016 at 1:32 PM, Michele <mi...@finconsgroup.com>
wrote:
> Hi Brad,
>
> as you suggested and I split big files into chunks of 1000 lines changing
> my
> route like this:
>
> <route id="cr-cbikit-1">
> <from
>
> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
> />
> <split streaming="true" parallelProcessing="false"
> >
> <tokenize token="\r" group="1000"
> regex="true"/>
> <to uri="activemq:queue:Cbikit.Key" />
> </split>
> </route>
>
> <route id="cr-cbikit-2">
> <from
> uri="activemq:queue:Cbikit.Key?destination.consumer.prefetchSize=1"
> />
> <split streaming="true">
> <tokenize token="\r" />
> <to
> uri="sedaQueue:queue.ReadyToProcess?blockWhenFull=true" />
> </split>
> </route>
>
> <route id="cr-cbikit-3">
> <from
> uri="sedaQueue:queue.ReadyToProcess?concurrentConsumers=3" />
> process....
> </route>
>
> This works fine! Memory Usage and Processing Time are very good.
> But, I found a problem: Second route doesn't work when token is equal to
> \r.
> When I log a message that contains 1000 lines I see
> line\rline2\rline3\r...line999
> The split works when token is \\r. Why? The file affected by this issue is
> a
> csv.
>
> Thanks a lot again for your support.
>
> Best regards
>
> Michele
>
>
>
>
>
>
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781750.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi Brad,
as you suggested and I split big files into chunks of 1000 lines changing my
route like this:
<route id="cr-cbikit-1">
<from
uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
/>
<split streaming="true" parallelProcessing="false" >
<tokenize token="\r" group="1000" regex="true"/>
<to uri="activemq:queue:Cbikit.Key" />
</split>
</route>
<route id="cr-cbikit-2">
<from uri="activemq:queue:Cbikit.Key?destination.consumer.prefetchSize=1"
/>
<split streaming="true">
<tokenize token="\r" />
<to uri="sedaQueue:queue.ReadyToProcess?blockWhenFull=true" />
</split>
</route>
<route id="cr-cbikit-3">
<from uri="sedaQueue:queue.ReadyToProcess?concurrentConsumers=3" />
process....
</route>
This works fine! Memory Usage and Processing Time are very good.
But, I found a problem: Second route doesn't work when token is equal to \r.
When I log a message that contains 1000 lines I see
line\rline2\rline3\r...line999
The split works when token is \\r. Why? The file affected by this issue is a
csv.
Thanks a lot again for your support.
Best regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781750.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
You may want to tell it to chunk the file then. That's something you can
add to the splitter/tokenizer.
On Mon, Apr 18, 2016 at 4:12 PM, Michele <mi...@finconsgroup.com>
wrote:
> Hi,
>
> I have different mediations. Each mediation handles an incoming file type
> (csv, txt delimited or fixed length) during the prorcessing.
> In same csv and txt file Tokenizer works fine with \n or \r\n or \r.
> Few minutes ago, I found solution for a txt delimited file adding a
> charset=iso-8859-1 option to uri of file component avoiding converterBodyTo
> and as token \n.
>
> Thank you very much for your help/suggestions. The memory usage is
> significantly improved.
>
> Now, I'll try to improve ActiveMq performance because the speed of process
> on Producer side (Read-Split-StoreInQueue) is very slow (Put in queue
> ~2msg/s in some process). My goal is to store the whole file splitted line
> by line in queue as soon as possible to increase reliability.
>
> Thank you a lot again.
>
> Best Regards
>
> Michele
>
>
>
>
>
>
>
>
>
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781298.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
I have different mediations. Each mediation handles an incoming file type
(csv, txt delimited or fixed length) during the prorcessing.
In same csv and txt file Tokenizer works fine with \n or \r\n or \r.
Few minutes ago, I found solution for a txt delimited file adding a
charset=iso-8859-1 option to uri of file component avoiding converterBodyTo
and as token \n.
Thank you very much for your help/suggestions. The memory usage is
significantly improved.
Now, I'll try to improve ActiveMq performance because the speed of process
on Producer side (Read-Split-StoreInQueue) is very slow (Put in queue
~2msg/s in some process). My goal is to store the whole file splitted line
by line in queue as soon as possible to increase reliability.
Thank you a lot again.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781298.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
If that doesn't work for you there is another way I use when I have to read
complex files that aren't simple one line = one record but it isn't really
necessary if you are simply reading in a CSV or fixed width file with only
single lines. I don't know what your record format or Beanio looks like
so can't really tell. But if it is set up to marshal/unmarshal single
lines then you simply have to tell the tokenizer what those lines look like
and what the line ending delimiters are.
On Mon, Apr 18, 2016 at 11:17 AM, Brad Johnson <brad.johnson@mediadriver.com
> wrote:
> The tokenization may require a different line ending - \r or \n or \r\n
> for example. The file reader has to understand what it is parsing. I take
> it when you use that splitter it with the tokenization it was reading the
> whole file in one big slurp and never finding a line ending and you ended
> up doing a convertBody to string in order to convert the whole file into a
> String? Obviously that's not what you're after. What are the line endings?
>
> On Mon, Apr 18, 2016 at 3:26 AM, Michele <
> michele.mazzilli@finconsgroup.com> wrote:
>
>> Hi Brad,
>>
>> first of all thank you very much for the time you dedicate me.
>>
>> Are you getting the entire file in memory? I think so. I thought BeanIO
>> worked in lazy mode...
>>
>> A question, I noticed that in some file Splitter doesn't work with
>> <from uri="file:inbox"/>
>> <split streaming="true">
>> <tokenize token="\n" />
>> <to uri="activemq:queue:store"/>
>> </split>
>>
>> but It required a convertBodyTo type="java.lang.String" before split, but
>> we
>> don't want this (entire file in memory). How can I split file without
>> converter?
>>
>> I tried to introduce some changes at Route as you suggested with a good
>> use
>> of memory avoiding OOM.
>>
>> Current Route:
>>
>> <route id="FileRetriever_Route">
>> <from
>>
>> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_SAP{{uri.inbound.options}}"
>> />
>> <split streaming="true" executorServiceRef="ThreadPoolExecutor">
>> <tokenize token="\n" />
>> <choice>
>> <when>
>> <simple></simple>
>> <log message="Store Message in Queue
>> ${in.header.CamelSplitIndex} -
>> ${body}" loggingLevel="DEBUG" />
>> <to uri="activemq:queue:SAP.Product" />
>> </when>
>> <otherwise>
>> <log message="Invalid Row
>> ${in.header.CamelSplitIndex} - Discarded:
>> ${body}" />
>> </otherwise>
>> </choice>
>> </split>
>> </route>
>> <route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
>> <from
>> uri="activemq:queue:SAP.Product?destination.consumer.prefetchSize=1" />
>>
>> <throttle timePeriodMillis="1000"
>> asyncDelayed="true">
>> <constant>3</constant>
>>
>> <log message="START -
>> ProcessMessageData_Route" />
>>
>>
>> <unmarshal ref="RecordParser" />
>>
>> <setBody><simple>${body[0]}</simple></setBody>
>> <to
>>
>> uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=java.util.LinkedHashMap"
>> />
>> <marshal ref="Gson" />
>>
>> <enrich uri="direct:crm-login"
>> strategyRef="OAuthStrategy" />
>>
>> <recipientList>
>>
>> <header>CompleteActionPath</header>
>> </recipientList>
>>
>> </throttle>
>> </route>
>>
>> Thanks a lot again
>>
>> Best Regards
>>
>> Michele
>>
>>
>>
>>
>> --
>> View this message in context:
>> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781251.html
>> Sent from the Camel - Users mailing list archive at Nabble.com.
>>
>
>
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
The tokenization may require a different line ending - \r or \n or \r\n
for example. The file reader has to understand what it is parsing. I take
it when you use that splitter it with the tokenization it was reading the
whole file in one big slurp and never finding a line ending and you ended
up doing a convertBody to string in order to convert the whole file into a
String? Obviously that's not what you're after. What are the line endings?
On Mon, Apr 18, 2016 at 3:26 AM, Michele <mi...@finconsgroup.com>
wrote:
> Hi Brad,
>
> first of all thank you very much for the time you dedicate me.
>
> Are you getting the entire file in memory? I think so. I thought BeanIO
> worked in lazy mode...
>
> A question, I noticed that in some file Splitter doesn't work with
> <from uri="file:inbox"/>
> <split streaming="true">
> <tokenize token="\n" />
> <to uri="activemq:queue:store"/>
> </split>
>
> but It required a convertBodyTo type="java.lang.String" before split, but
> we
> don't want this (entire file in memory). How can I split file without
> converter?
>
> I tried to introduce some changes at Route as you suggested with a good use
> of memory avoiding OOM.
>
> Current Route:
>
> <route id="FileRetriever_Route">
> <from
>
> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_SAP{{uri.inbound.options}}"
> />
> <split streaming="true" executorServiceRef="ThreadPoolExecutor">
> <tokenize token="\n" />
> <choice>
> <when>
> <simple></simple>
> <log message="Store Message in Queue
> ${in.header.CamelSplitIndex} -
> ${body}" loggingLevel="DEBUG" />
> <to uri="activemq:queue:SAP.Product" />
> </when>
> <otherwise>
> <log message="Invalid Row
> ${in.header.CamelSplitIndex} - Discarded:
> ${body}" />
> </otherwise>
> </choice>
> </split>
> </route>
> <route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
> <from
> uri="activemq:queue:SAP.Product?destination.consumer.prefetchSize=1" />
>
> <throttle timePeriodMillis="1000"
> asyncDelayed="true">
> <constant>3</constant>
>
> <log message="START -
> ProcessMessageData_Route" />
>
>
> <unmarshal ref="RecordParser" />
>
> <setBody><simple>${body[0]}</simple></setBody>
> <to
>
> uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=java.util.LinkedHashMap"
> />
> <marshal ref="Gson" />
>
> <enrich uri="direct:crm-login"
> strategyRef="OAuthStrategy" />
>
> <recipientList>
> <header>CompleteActionPath</header>
> </recipientList>
>
> </throttle>
> </route>
>
> Thanks a lot again
>
> Best Regards
>
> Michele
>
>
>
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781251.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi Brad,
first of all thank you very much for the time you dedicate me.
Are you getting the entire file in memory? I think so. I thought BeanIO
worked in lazy mode...
A question, I noticed that in some file Splitter doesn't work with
<from uri="file:inbox"/>
<split streaming="true">
<tokenize token="\n" />
<to uri="activemq:queue:store"/>
</split>
but It required a convertBodyTo type="java.lang.String" before split, but we
don't want this (entire file in memory). How can I split file without
converter?
I tried to introduce some changes at Route as you suggested with a good use
of memory avoiding OOM.
Current Route:
<route id="FileRetriever_Route">
<from
uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_SAP{{uri.inbound.options}}"
/>
<split streaming="true" executorServiceRef="ThreadPoolExecutor">
<tokenize token="\n" />
<choice>
<when>
<simple></simple>
<log message="Store Message in Queue ${in.header.CamelSplitIndex} -
${body}" loggingLevel="DEBUG" />
<to uri="activemq:queue:SAP.Product" />
</when>
<otherwise>
<log message="Invalid Row ${in.header.CamelSplitIndex} - Discarded:
${body}" />
</otherwise>
</choice>
</split>
</route>
<route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
<from
uri="activemq:queue:SAP.Product?destination.consumer.prefetchSize=1" />
<throttle timePeriodMillis="1000" asyncDelayed="true">
<constant>3</constant>
<log message="START - ProcessMessageData_Route" />
<unmarshal ref="RecordParser" />
<setBody><simple>${body[0]}</simple></setBody>
<to
uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=java.util.LinkedHashMap"
/>
<marshal ref="Gson" />
<enrich uri="direct:crm-login" strategyRef="OAuthStrategy" />
<recipientList>
<header>CompleteActionPath</header>
</recipientList>
</throttle>
</route>
Thanks a lot again
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781251.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
At this point in your code:
<from uri="seda:processAndStoreInQueue" /
<unmarshal ref="myBeanio" />
Are you getting the entire file in memory? That's going to be big so maybe
not what you want. It may be livable however.
ActiveMQ can handle a lot of throughput but remember that if you are using
persistent queues you necessarily lose several orders of magnitude in speed
to persist and restore the data and you are both marshaling and
unmarshaling the data. KahaDB and LevelDB are both journaling databases
and are very fast but any time you go from nanosecond RAM speeds to
milliscecond hard disk speeds you are going to slow down. And that's times
two - down to the disk and back up into memory. Memory consumption is also
going to go up. It still shouldn't take 6 hours to process.
Something like this could be used to put individual lines from a file onto
your queue and you can even specify it in blocks if you like. But I don't
know the form of your data.
<route>
<from uri="file:inbox"/>
<split streaming="true">
<tokenize token="\n" />
<to uri="activemq:queue:store"/>
</split>
</route>
In that case it starts reading individual lines from the file, splits them
on the lines, and drops each line onto the queue. Simultaneously you can
start reading from that queue using the multithreaded consumer that you
have. In this way you won't be reading the entire file into memory in one
big chunk but will stream it in a line at a time or you can add the "group"
to the tokenize line to specify how many lines should be read in at a
time.
I'd just take that and try it with your seda queue structure first to see
if you can get it to work right for you. When you read off the queue you
can then unmarshal it via Beanio and do transactions. At that point you
could specify min/max threads for reading off the queue.
This should also let the JVM reclaim memory as it goes along since the
strings read in and off the queue, transformed and then processed are all
ready for garbage collection. If the entire file is held in memory and
being passed from queue to queue that won't get released.
On Sat, Apr 16, 2016 at 8:04 AM, Michele <mi...@finconsgroup.com>
wrote:
> Hi,
>
> as Ranx suggested, I tried with chain routes and attached screenshot of
> memory usage seda-memory-usage.png
> <http://camel.465427.n5.nabble.com/file/n5781206/seda-memory-usage.png> .
> it is evident that there is a net improvement, on average 600 MB memory
> usage.
>
> The chain of routes configured is:
>
> <route id="FileRetriever_Route">
> <from
>
> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
> />
> <to uri="seda:processAndStoreInQueue" />
> </route>
>
> <route id="Splitter_Route">
> <from uri="seda:processAndStoreInQueue" />
>
> <unmarshal ref="myBeanio" />
>
> <split streaming="true" executorServiceRef="threadPoolExecutor" >
> <simple>${body}</simple>
> <choice>
> <when>
> <simple></simple>
> <setHeader
>
> headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
> <to uri="seda:process.queue" />
> </when>
> <otherwise>
> <log message="Message discarded
> ${in.header.CamelSplitIndex} - ${body}"
> />
> </otherwise>
> </choice>
> </split>
> </route>
>
> <route id="ProcessAndStoreInSedaQueue">
> <from uri="seda:process.queue?concurrentConsumers=10" />
> <process ref="BodyEnricherProcessor" />
> <to
>
> uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=com.fincons.ingenico.crt2.cbikit.inbound.model.SerialNumber"
> />
> <marshal ref="Gson" />
> <log message="Store in SEDA Queue: ${in.header.CamelSplitIndex} -
> ${body}"
> />
> <to uri="seda:readyToSendRS.queue" />
> </route>
>
> <route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
> <from uri="seda:readyToSendRS.queue?concurrentConsumers=10" />
> <throttle timePeriodMillis="1000" asyncDelayed="true">
> <constant>3</constant>
>
> <enrich uri="direct:crm-login" strategyRef="OAuthStrategy"
> />
>
> <recipientList>
> <header>CompleteActionPath</header>
> </recipientList>
>
> <unmarshal ref="Gson" />
>
> <choice>
> <when>
> <simple>${in.header.Result} ==
> 'ERROR'</simple>
> <log message="CRM Response: ${body}"
> loggingLevel="ERROR"/>
> </when>
> </choice>
> </throttle>
> </route>
>
> Logs:
> Store in SEDA Queue: 36913 -
>
> {"data_caricamento_cbkit_c":"2016-04-05T08:17:51+02:00","data_import_cbikit_c":"2016-04-16T13:41:01+02:00","nome_file_cbikit_c":"20160405074559..csv","serialnumber_c":"12154WL60756738"}
> and Throttling Policy in action:
> INFO | d #46 - Throttle | CRMLogin_Route ...
>
> In addition, from my JUnit Test Camel is fastest (less than 1 minute) in
> the
> processing Read-Unmarshal-Split-Process-MockResult.
>
> So, when I repalced seda:readyToSendRS.queue with
> activemq:queue:readyToSendRS, the memory usage triples.
>
> I can not use SEDA Component because in according to documentation:
> this component does not implement any kind of persistence or recovery, if
> the VM terminates while messages are yet to be processed. If you need
> persistence, reliability or distributed SEDA, try using either JMS or
> ActiveMQ.
>
> So, Can ActiveMQ manage a large number of messages in few seconds or
> minutes? Or Do I need to defined a load balancer on Broker?
>
> Thanks a lot for your support.
>
> Best Regards
>
> Michele
>
>
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781206.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
At this point in your code:
<from uri="seda:processAndStoreInQueue" /
<unmarshal ref="myBeanio" />
Are you getting the entire file in memory? That's going to be big so maybe
not what you want. It may be livable however.
ActiveMQ can handle a lot of throughput but remember that if you are using
persistent queues you necessarily lose several orders of magnitude in speed
to persist and restore the data and you are both marshaling and
unmarshaling the data. KahaDB and LevelDB are both journaling databases
and are very fast but any time you go from nanosecond RAM speeds to
milliscecond hard disk speeds you are going to slow down. And that's times
two - down to the disk and back up into memory. Memory consumption is also
going to go up. It still shouldn't take 6 hours to process.
Something like this could be used to put individual lines from a file onto
your queue and you can even specify it in blocks if you like. But I don't
know the form of your data.
<route>
<from uri="file:inbox"/>
<split streaming="true">
<tokenize token="\n" />
<to uri="activemq:queue:store"/>
</split>
</route>
In that case it starts reading individual lines from the file, splits them
on the lines, and drops each line onto the queue. Simultaneously you can
start reading from that queue using the multithreaded consumer that you
have. In this way you won't be reading the entire file into memory in one
big chunk but will stream it in a line at a time or you can add the "group"
to the tokenize line to specify how many lines should be read in at a
time.
I'd just take that and try it with your seda queue structure first to see
if you can get it to work right for you. When you read off the queue you
can then unmarshal it via Beanio and do transactions. At that point you
could specify min/max threads for reading off the queue.
This should also let the JVM reclaim memory as it goes along since the
strings read in and off the queue, transformed and then processed are all
ready for garbage collection. It will also kick off your pocessing right
away instead of waiting until the entire file is read in.
Brad
On Sat, Apr 16, 2016 at 8:04 AM, Michele <mi...@finconsgroup.com>
wrote:
> Hi,
>
> as Ranx suggested, I tried with chain routes and attached screenshot of
> memory usage seda-memory-usage.png
> <http://camel.465427.n5.nabble.com/file/n5781206/seda-memory-usage.png> .
> it is evident that there is a net improvement, on average 600 MB memory
> usage.
>
> The chain of routes configured is:
>
> <route id="FileRetriever_Route">
> <from
>
> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
> />
> <to uri="seda:processAndStoreInQueue" />
> </route>
>
> <route id="Splitter_Route">
> <from uri="seda:processAndStoreInQueue" />
>
> <unmarshal ref="myBeanio" />
>
> <split streaming="true" executorServiceRef="threadPoolExecutor" >
> <simple>${body}</simple>
> <choice>
> <when>
> <simple></simple>
> <setHeader
>
> headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
> <to uri="seda:process.queue" />
> </when>
> <otherwise>
> <log message="Message discarded
> ${in.header.CamelSplitIndex} - ${body}"
> />
> </otherwise>
> </choice>
> </split>
> </route>
>
> <route id="ProcessAndStoreInSedaQueue">
> <from uri="seda:process.queue?concurrentConsumers=10" />
> <process ref="BodyEnricherProcessor" />
> <to
>
> uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=com.fincons.ingenico.crt2.cbikit.inbound.model.SerialNumber"
> />
> <marshal ref="Gson" />
> <log message="Store in SEDA Queue: ${in.header.CamelSplitIndex} -
> ${body}"
> />
> <to uri="seda:readyToSendRS.queue" />
> </route>
>
> <route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
> <from uri="seda:readyToSendRS.queue?concurrentConsumers=10" />
> <throttle timePeriodMillis="1000" asyncDelayed="true">
> <constant>3</constant>
>
> <enrich uri="direct:crm-login" strategyRef="OAuthStrategy"
> />
>
> <recipientList>
> <header>CompleteActionPath</header>
> </recipientList>
>
> <unmarshal ref="Gson" />
>
> <choice>
> <when>
> <simple>${in.header.Result} ==
> 'ERROR'</simple>
> <log message="CRM Response: ${body}"
> loggingLevel="ERROR"/>
> </when>
> </choice>
> </throttle>
> </route>
>
> Logs:
> Store in SEDA Queue: 36913 -
>
> {"data_caricamento_cbkit_c":"2016-04-05T08:17:51+02:00","data_import_cbikit_c":"2016-04-16T13:41:01+02:00","nome_file_cbikit_c":"20160405074559..csv","serialnumber_c":"12154WL60756738"}
> and Throttling Policy in action:
> INFO | d #46 - Throttle | CRMLogin_Route ...
>
> In addition, from my JUnit Test Camel is fastest (less than 1 minute) in
> the
> processing Read-Unmarshal-Split-Process-MockResult.
>
> So, when I repalced seda:readyToSendRS.queue with
> activemq:queue:readyToSendRS, the memory usage triples.
>
> I can not use SEDA Component because in according to documentation:
> this component does not implement any kind of persistence or recovery, if
> the VM terminates while messages are yet to be processed. If you need
> persistence, reliability or distributed SEDA, try using either JMS or
> ActiveMQ.
>
> So, Can ActiveMQ manage a large number of messages in few seconds or
> minutes? Or Do I need to defined a load balancer on Broker?
>
> Thanks a lot for your support.
>
> Best Regards
>
> Michele
>
>
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781206.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
as Ranx suggested, I tried with chain routes and attached screenshot of
memory usage seda-memory-usage.png
<http://camel.465427.n5.nabble.com/file/n5781206/seda-memory-usage.png> .
it is evident that there is a net improvement, on average 600 MB memory
usage.
The chain of routes configured is:
<route id="FileRetriever_Route">
<from
uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
/>
<to uri="seda:processAndStoreInQueue" />
</route>
<route id="Splitter_Route">
<from uri="seda:processAndStoreInQueue" />
<unmarshal ref="myBeanio" />
<split streaming="true" executorServiceRef="threadPoolExecutor" >
<simple>${body}</simple>
<choice>
<when>
<simple></simple>
<setHeader
headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
<to uri="seda:process.queue" />
</when>
<otherwise>
<log message="Message discarded ${in.header.CamelSplitIndex} - ${body}"
/>
</otherwise>
</choice>
</split>
</route>
<route id="ProcessAndStoreInSedaQueue">
<from uri="seda:process.queue?concurrentConsumers=10" />
<process ref="BodyEnricherProcessor" />
<to
uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=com.fincons.ingenico.crt2.cbikit.inbound.model.SerialNumber"
/>
<marshal ref="Gson" />
<log message="Store in SEDA Queue: ${in.header.CamelSplitIndex} - ${body}"
/>
<to uri="seda:readyToSendRS.queue" />
</route>
<route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
<from uri="seda:readyToSendRS.queue?concurrentConsumers=10" />
<throttle timePeriodMillis="1000" asyncDelayed="true">
<constant>3</constant>
<enrich uri="direct:crm-login" strategyRef="OAuthStrategy" />
<recipientList>
<header>CompleteActionPath</header>
</recipientList>
<unmarshal ref="Gson" />
<choice>
<when>
<simple>${in.header.Result} == 'ERROR'</simple>
<log message="CRM Response: ${body}" loggingLevel="ERROR"/>
</when>
</choice>
</throttle>
</route>
Logs:
Store in SEDA Queue: 36913 -
{"data_caricamento_cbkit_c":"2016-04-05T08:17:51+02:00","data_import_cbikit_c":"2016-04-16T13:41:01+02:00","nome_file_cbikit_c":"20160405074559..csv","serialnumber_c":"12154WL60756738"}
and Throttling Policy in action:
INFO | d #46 - Throttle | CRMLogin_Route ...
In addition, from my JUnit Test Camel is fastest (less than 1 minute) in the
processing Read-Unmarshal-Split-Process-MockResult.
So, when I repalced seda:readyToSendRS.queue with
activemq:queue:readyToSendRS, the memory usage triples.
I can not use SEDA Component because in according to documentation:
this component does not implement any kind of persistence or recovery, if
the VM terminates while messages are yet to be processed. If you need
persistence, reliability or distributed SEDA, try using either JMS or
ActiveMQ.
So, Can ActiveMQ manage a large number of messages in few seconds or
minutes? Or Do I need to defined a load balancer on Broker?
Thanks a lot for your support.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781206.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
I suspect the biggest problem you may be having here is that the file isn't
truly streaming but getting slurped into memory as a whole. I don't know
that for certain but that test of two queues should show you that.
That might be throwaway code or it might be the basis for a next
implementation. But it will at least let you see what you're dealing
with. I wouldn't go too far past just reading/streaming the file in and
passing it on to the next queue so you can verify that you are getting
individual data records.
On Fri, Apr 15, 2016 at 9:58 AM, Michele <mi...@finconsgroup.com>
wrote:
> How big is that object on the queue?
> As optimization, I changed generic LinkedHashMap in a bean SerialNumber
> that
> has this properties
>
> public class SerialNumber implements Serializable {
>
> private static final long serialVersionUID = -9067015746823050409L;
>
> private Date arg1;
> private Date arg2;
> private String arg3;
> private String arg4;
>
> // getter setter
> }
>
> Is it only a single line of data? Yes
>
> I try with seda:processAndStoreInQueue?concurrentConsumers=30 but the
> result
> is the same.
> I attached scrennshots with memory usage and live threads.
> memory-usage.png
> <http://camel.465427.n5.nabble.com/file/n5781179/memory-usage.png>
> live-threads.png
> <http://camel.465427.n5.nabble.com/file/n5781179/live-threads.png>
>
> I will try decompose the problem into a number of separate seda:queues to
> verify performance and memory usage and let you know.
>
> Thanks a lot again for your support.
>
> Best Regards
>
> Michele
>
>
>
>
>
>
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781179.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
How big is that object on the queue?
As optimization, I changed generic LinkedHashMap in a bean SerialNumber that
has this properties
public class SerialNumber implements Serializable {
private static final long serialVersionUID = -9067015746823050409L;
private Date arg1;
private Date arg2;
private String arg3;
private String arg4;
// getter setter
}
Is it only a single line of data? Yes
I try with seda:processAndStoreInQueue?concurrentConsumers=30 but the result
is the same.
I attached scrennshots with memory usage and live threads.
memory-usage.png
<http://camel.465427.n5.nabble.com/file/n5781179/memory-usage.png>
live-threads.png
<http://camel.465427.n5.nabble.com/file/n5781179/live-threads.png>
I will try decompose the problem into a number of separate seda:queues to
verify performance and memory usage and let you know.
Thanks a lot again for your support.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781179.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
The easiest way to think about SEDA queues is to recognize why they were
created and I think that will help you decompose such problems. Staged
Event Driven Architecture. Initially they were invented for auto
management of threads on queues where one stage that is getting too few
threads could steal some from another that has too many and the whole
system would degrade gracefully under high load. But in our current
context you might think of it as an easy way to decouple processing stages
and assign threading.
Just decompose the problem into a number of separate seda:queues and chain
the calls together with a log statement in each so you can make sure you
aren't accidentally reading an entire file in and inadvertently picking up
a huge lump of data.
<route id="stream.file.in">
<from uri="file:/inbox"/>
<unmarshal...."/>
<log message="${body}"/>
<to uri="seda:process.queue"/>
</route>
<route id="process.records">
<from uri="seda:process.queue?concurrentConsumers=10"/>
<log message="${body}"/>
//etc.
</route>
Make sure you are only seeing the rows come through. Obviously I left out
a lot of the data streaming/unmarshalling logic there. But a quick simple
route like that with the appropriate "streaming" and splitting as necessary
should result in that second stage getting on single lines or single
records (if they are multi-line).
On Fri, Apr 15, 2016 at 9:01 AM, Brad Johnson <br...@mediadriver.com>
wrote:
> 6 hours? I'm currently streaming a 50k line file through Beanio with seda
> queues in and doing similar processing in under a minute.
>
> You have a queue here tht appears to have something on it that isbeing
> unmarashalled and then split. How big is that object on the queue? Is it
> only a single line of data? Try this.
>
> <route id="ProcessAndStoreInQueue_Route">
> <from
> uri="seda:processAndStoreInQueue?concurrentConsumers=30" />
>
>
>
> On Fri, Apr 15, 2016 at 8:25 AM, Jens Breitenstein <ma...@j-b-s.de>
> wrote:
>
>> Hi Michele
>>
>> Reading a CSV with 40k lines using camel in streaming takes a view
>> seconds. As you limit the queue-size to avoid OOM the entire performance
>> depends how fast you can empty the queue.
>> How long does processing of ONE message take in average? To me it looks
>> like approximately 1.6 secs (35000/6/60/60). The processes responsible for
>> reading the queue is single-threaded??
>>
>> Jens
>>
>>
>> Am 15/04/16 um 14:59 schrieb Michele:
>>
>> Hi,
>>>
>>> I spent a bit of time reading different topics on this issue, and I
>>> changed
>>> my route like this reducing the memory usage of about 300Mb:
>>>
>>> <route id="FileRetriever_Route">
>>> <from
>>>
>>> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
>>> />
>>> <setHeader
>>>
>>> headerName="ImportDateTime"><simple>${date:now:yyyyMMdd-HHmmss}</simple></setHeader>
>>> <setHeader
>>>
>>> headerName="MsgCorrelationId"><simple>CBIKIT_INBOUND_${in.header.ImportDateTime}</simple></setHeader>
>>> <setHeader headerName="breadcrumbId">
>>>
>>>
>>> <simple>Import-${in.header.CamelFileName}-${in.header.ImportDateTime}-${in.header.breadcrumbId}</simple>
>>> </setHeader>
>>> <to uri="seda:processAndStoreInQueue" />
>>> <log message="END - FileRetriever_Route" />
>>> </route>
>>>
>>> <route id="ProcessAndStoreInQueue_Route">
>>> <from uri="seda:processAndStoreInQueue" />
>>> <unmarshal>
>>> <bindy type="Csv"
>>> classType="com.fincons.ingenico.crt2.cbikit.inbound.model.RowData"/>
>>> </unmarshal>
>>>
>>> <split streaming="true"
>>> executorServiceRef="myThreadPoolExecutor" >
>>> <simple>${body}</simple>
>>> <choice>
>>> <when>
>>> <simple></simple>
>>> <setHeader
>>>
>>> headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
>>> <process
>>> ref="BodyEnricherProcessor" />
>>> <to
>>>
>>> uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=com.fincons.ingenico.crt2.cbikit.inbound.model.SerialNumber"
>>> />
>>> <marshal ref="Gson" />
>>> <to
>>> uri="activemq:queue:CBIKIT" />
>>> </when>
>>> <otherwise>
>>> <log message="Message
>>> discarded ${in.header.CamelSplitIndex} -
>>> ${body}" />
>>> </otherwise>
>>> </choice>
>>> </split>
>>> </route>
>>>
>>> The last test processed 35000 lines of CSV file in about 6h with an
>>> average
>>> memory usage 1400Mb successful. But, Can I improve further processing
>>> performance?
>>>
>>> In addition, I noticed that Queue Size of Queue is low. Why? (Producer is
>>> slower than Consumer?)
>>>
>>> Thanks in advance.
>>>
>>> Best Regards
>>>
>>> Michele
>>>
>>>
>>>
>>> --
>>> View this message in context:
>>> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781168.html
>>> Sent from the Camel - Users mailing list archive at Nabble.com.
>>>
>>
>>
>
Re: Best Strategy to process a large number of rows in File
Posted by Brad Johnson <br...@mediadriver.com>.
6 hours? I'm currently streaming a 50k line file through Beanio with seda
queues in and doing similar processing in under a minute.
You have a queue here tht appears to have something on it that isbeing
unmarashalled and then split. How big is that object on the queue? Is it
only a single line of data? Try this.
<route id="ProcessAndStoreInQueue_Route">
<from
uri="seda:processAndStoreInQueue?concurrentConsumers=30" />
On Fri, Apr 15, 2016 at 8:25 AM, Jens Breitenstein <ma...@j-b-s.de>
wrote:
> Hi Michele
>
> Reading a CSV with 40k lines using camel in streaming takes a view
> seconds. As you limit the queue-size to avoid OOM the entire performance
> depends how fast you can empty the queue.
> How long does processing of ONE message take in average? To me it looks
> like approximately 1.6 secs (35000/6/60/60). The processes responsible for
> reading the queue is single-threaded??
>
> Jens
>
>
> Am 15/04/16 um 14:59 schrieb Michele:
>
> Hi,
>>
>> I spent a bit of time reading different topics on this issue, and I
>> changed
>> my route like this reducing the memory usage of about 300Mb:
>>
>> <route id="FileRetriever_Route">
>> <from
>>
>> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
>> />
>> <setHeader
>>
>> headerName="ImportDateTime"><simple>${date:now:yyyyMMdd-HHmmss}</simple></setHeader>
>> <setHeader
>>
>> headerName="MsgCorrelationId"><simple>CBIKIT_INBOUND_${in.header.ImportDateTime}</simple></setHeader>
>> <setHeader headerName="breadcrumbId">
>>
>>
>> <simple>Import-${in.header.CamelFileName}-${in.header.ImportDateTime}-${in.header.breadcrumbId}</simple>
>> </setHeader>
>> <to uri="seda:processAndStoreInQueue" />
>> <log message="END - FileRetriever_Route" />
>> </route>
>>
>> <route id="ProcessAndStoreInQueue_Route">
>> <from uri="seda:processAndStoreInQueue" />
>> <unmarshal>
>> <bindy type="Csv"
>> classType="com.fincons.ingenico.crt2.cbikit.inbound.model.RowData"/>
>> </unmarshal>
>>
>> <split streaming="true"
>> executorServiceRef="myThreadPoolExecutor" >
>> <simple>${body}</simple>
>> <choice>
>> <when>
>> <simple></simple>
>> <setHeader
>>
>> headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
>> <process
>> ref="BodyEnricherProcessor" />
>> <to
>>
>> uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=com.fincons.ingenico.crt2.cbikit.inbound.model.SerialNumber"
>> />
>> <marshal ref="Gson" />
>> <to
>> uri="activemq:queue:CBIKIT" />
>> </when>
>> <otherwise>
>> <log message="Message
>> discarded ${in.header.CamelSplitIndex} -
>> ${body}" />
>> </otherwise>
>> </choice>
>> </split>
>> </route>
>>
>> The last test processed 35000 lines of CSV file in about 6h with an
>> average
>> memory usage 1400Mb successful. But, Can I improve further processing
>> performance?
>>
>> In addition, I noticed that Queue Size of Queue is low. Why? (Producer is
>> slower than Consumer?)
>>
>> Thanks in advance.
>>
>> Best Regards
>>
>> Michele
>>
>>
>>
>> --
>> View this message in context:
>> http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781168.html
>> Sent from the Camel - Users mailing list archive at Nabble.com.
>>
>
>
Re: Best Strategy to process a large number of rows in File
Posted by Jens Breitenstein <ma...@j-b-s.de>.
Hi Michele
Reading a CSV with 40k lines using camel in streaming takes a view
seconds. As you limit the queue-size to avoid OOM the entire performance
depends how fast you can empty the queue.
How long does processing of ONE message take in average? To me it looks
like approximately 1.6 secs (35000/6/60/60). The processes responsible
for reading the queue is single-threaded??
Jens
Am 15/04/16 um 14:59 schrieb Michele:
> Hi,
>
> I spent a bit of time reading different topics on this issue, and I changed
> my route like this reducing the memory usage of about 300Mb:
>
> <route id="FileRetriever_Route">
> <from
> uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
> />
> <setHeader
> headerName="ImportDateTime"><simple>${date:now:yyyyMMdd-HHmmss}</simple></setHeader>
> <setHeader
> headerName="MsgCorrelationId"><simple>CBIKIT_INBOUND_${in.header.ImportDateTime}</simple></setHeader>
> <setHeader headerName="breadcrumbId">
>
> <simple>Import-${in.header.CamelFileName}-${in.header.ImportDateTime}-${in.header.breadcrumbId}</simple>
> </setHeader>
> <to uri="seda:processAndStoreInQueue" />
> <log message="END - FileRetriever_Route" />
> </route>
>
> <route id="ProcessAndStoreInQueue_Route">
> <from uri="seda:processAndStoreInQueue" />
> <unmarshal>
> <bindy type="Csv"
> classType="com.fincons.ingenico.crt2.cbikit.inbound.model.RowData"/>
> </unmarshal>
>
> <split streaming="true" executorServiceRef="myThreadPoolExecutor" >
> <simple>${body}</simple>
> <choice>
> <when>
> <simple></simple>
> <setHeader
> headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
> <process ref="BodyEnricherProcessor" />
> <to
> uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=com.fincons.ingenico.crt2.cbikit.inbound.model.SerialNumber"
> />
> <marshal ref="Gson" />
> <to uri="activemq:queue:CBIKIT" />
> </when>
> <otherwise>
> <log message="Message discarded ${in.header.CamelSplitIndex} -
> ${body}" />
> </otherwise>
> </choice>
> </split>
> </route>
>
> The last test processed 35000 lines of CSV file in about 6h with an average
> memory usage 1400Mb successful. But, Can I improve further processing
> performance?
>
> In addition, I noticed that Queue Size of Queue is low. Why? (Producer is
> slower than Consumer?)
>
> Thanks in advance.
>
> Best Regards
>
> Michele
>
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781168.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
RE: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
I spent a bit of time reading different topics on this issue, and I changed
my route like this reducing the memory usage of about 300Mb:
<route id="FileRetriever_Route">
<from
uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
/>
<setHeader
headerName="ImportDateTime"><simple>${date:now:yyyyMMdd-HHmmss}</simple></setHeader>
<setHeader
headerName="MsgCorrelationId"><simple>CBIKIT_INBOUND_${in.header.ImportDateTime}</simple></setHeader>
<setHeader headerName="breadcrumbId">
<simple>Import-${in.header.CamelFileName}-${in.header.ImportDateTime}-${in.header.breadcrumbId}</simple>
</setHeader>
<to uri="seda:processAndStoreInQueue" />
<log message="END - FileRetriever_Route" />
</route>
<route id="ProcessAndStoreInQueue_Route">
<from uri="seda:processAndStoreInQueue" />
<unmarshal>
<bindy type="Csv"
classType="com.fincons.ingenico.crt2.cbikit.inbound.model.RowData"/>
</unmarshal>
<split streaming="true" executorServiceRef="myThreadPoolExecutor" >
<simple>${body}</simple>
<choice>
<when>
<simple></simple>
<setHeader
headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
<process ref="BodyEnricherProcessor" />
<to
uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=com.fincons.ingenico.crt2.cbikit.inbound.model.SerialNumber"
/>
<marshal ref="Gson" />
<to uri="activemq:queue:CBIKIT" />
</when>
<otherwise>
<log message="Message discarded ${in.header.CamelSplitIndex} -
${body}" />
</otherwise>
</choice>
</split>
</route>
The last test processed 35000 lines of CSV file in about 6h with an average
memory usage 1400Mb successful. But, Can I improve further processing
performance?
In addition, I noticed that Queue Size of Queue is low. Why? (Producer is
slower than Consumer?)
Thanks in advance.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781168.html
Sent from the Camel - Users mailing list archive at Nabble.com.
RE: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
With Memory Analyzer but also with Visual VM, problems suspect found:
The thread java.lang.Thread @ 0x783481d90 Camel (IF_CBIKIT-Inbound-Context)
thread #28 - seda://processAndStoreInQueue keeps local variables with total
size 54.615.096 (11,93%) byte
1.212 instances of "byte[]", loaded by "<system class loader>" occupy
167.709.528 (36,65%) bytes.
(As you can said... your 20 MB file is taking up 40 MB as a String. The List
will take up a little more than another 40 MB (so you will need 80 MB during
the conversion). This lists lives throughout your split process...)
6 instances of "org.apache.activemq.command.ActiveMQObjectMessage", loaded
by "org.apache.felix.framework.BundleWiringImpl$BundleClassLoaderJava5 @
0x781b26308" occupy 104.612.744 (22,86%) bytes.
This result obtained introducing lazyloat on csv.
Do I need to change my route?
Thanks a lot for your support.
Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781014.html
Sent from the Camel - Users mailing list archive at Nabble.com.
RE: Best Strategy to process a large number of rows in File
Posted by "Siano, Stephan" <st...@sap.com>.
Hi,
The camel-csv component converts the file into a List,
Therefore your 20 MB file is taking up 40 MB as a String. The List will take up a little more than another 40 MB (so you will need 80 MB during the conversion). This lists lives throughout your split process...
What you could do to reduce memory footprint:
1. Avoid the string conversion. If you need to read your streams multiple times, a steam cache might help you.
2. Use streaming for the csv conversion (set lazyLoad="true").
If that doesn't help, you should analyze the heap dump for the out of memory error to see where the memory is consumed (if you are using eclipse there is a Memory Analyzer plugin available).
Best regards
Stephan
-----Original Message-----
From: Michele [mailto:michele.mazzilli@finconsgroup.com]
Sent: Mittwoch, 13. April 2016 14:34
To: users@camel.apache.org
Subject: RE: Best Strategy to process a large number of rows in File
Hi,
In this case file size is 20.424KB and it contains about 40000 lines.
I used the convertTo to avoid Rollback on rename file
(GenericFileOperationFailedException - only Windows).
The File is a cvs and so I defined CSV Data format like this (Transforms
each line in a map key-value)
<csv id="ReaderDataFormat" delimiter="{{in.file.delimiter}}"
skipHeaderRecord="true" ignoreEmptyLines="true" useMaps="true">
<header>Key</header>
<header>Date</header>
<header>Time</header>
<header>User1</header>
<header>User2</header>
<header>User3</header>
<header>Status</header>
<header>CertificateNumber</header>
<header>CA_KEY_INDEX</header>
<header>INJ_CARD_SN</header>
<header>TERM_PN</header>
<header>TERM_SN</header>
<header>Certificate</header>
</csv>
I used CSV DF to handle CSV payloads.
Active MQ configuration is based on PooledConnection and to limit the amount
of memory (Per-Destination) I set:
<policyEntry queue="IF_CBIKIT" producerFlowControl="true"
optimizedDispatch="true" reduceMemoryFootprint="true" memoryLimit="10Mb" >
</policyEntry>
But, after about 1h the system goes in OutOfMemory.
Thanks you so much for your help.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781003.html
Sent from the Camel - Users mailing list archive at Nabble.com.
RE: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
In this case file size is 20.424KB and it contains about 40000 lines.
I used the convertTo to avoid Rollback on rename file
(GenericFileOperationFailedException - only Windows).
The File is a cvs and so I defined CSV Data format like this (Transforms
each line in a map key-value)
<csv id="ReaderDataFormat" delimiter="{{in.file.delimiter}}"
skipHeaderRecord="true" ignoreEmptyLines="true" useMaps="true">
<header>Key</header>
<header>Date</header>
<header>Time</header>
<header>User1</header>
<header>User2</header>
<header>User3</header>
<header>Status</header>
<header>CertificateNumber</header>
<header>CA_KEY_INDEX</header>
<header>INJ_CARD_SN</header>
<header>TERM_PN</header>
<header>TERM_SN</header>
<header>Certificate</header>
</csv>
I used CSV DF to handle CSV payloads.
Active MQ configuration is based on PooledConnection and to limit the amount
of memory (Per-Destination) I set:
<policyEntry queue="IF_CBIKIT" producerFlowControl="true"
optimizedDispatch="true" reduceMemoryFootprint="true" memoryLimit="10Mb" >
</policyEntry>
But, after about 1h the system goes in OutOfMemory.
Thanks you so much for your help.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781003.html
Sent from the Camel - Users mailing list archive at Nabble.com.
RE: Best Strategy to process a large number of rows in File
Posted by "Siano, Stephan" <st...@sap.com>.
Hi,
How big is your document?
You are converting that thing into a String, which is generally unadvisable for large content (as it will consume twice the file size in memory). The second question is what your ReaderDataFormat does with the data. Does it copy it again or does it support streaming?
Best regards
Stephan
-----Original Message-----
From: Michele [mailto:michele.mazzilli@finconsgroup.com]
Sent: Mittwoch, 13. April 2016 11:06
To: users@camel.apache.org
Subject: Re: Best Strategy to process a large number of rows in File
Hi,
I'm here again because I don't resolved my problem.
After several checks, I noticed this:
memory-usage.png
<http://camel.465427.n5.nabble.com/file/n5780965/memory-usage.png>
Why does thread related to seda://processAndStoreInQueue consume much
memory? How to optimize memory usage?
This is my route to manage file (in this case is a csv with 40000 rows) with
large number of rows:
<route id="FileRetriever_Route">
<from
uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
/>
<setHeader
headerName="ImportDateTime"><simple>${date:now:yyyyMMdd-HHmmss}</simple></setHeader>
<setHeader
headerName="MsgCorrelationId"><simple>CBIKIT_INBOUND_${in.header.ImportDateTime}</simple></setHeader>
<setHeader headerName="breadcrumbId">
<simple>Import-${in.header.CamelFileName}-${in.header.ImportDateTime}-${in.header.breadcrumbId}</simple>
</setHeader>
<log message="START - FileRetriever_Route - Found file
${in.header.CamelFileName}" />
<to uri="seda:processAndStoreInQueue" />
<log message="END - FileRetriever_Route" />
</route>
<route id="ProcessAndStoreInQueue_Route" >
<from uri="seda:processAndStoreInQueue" />
<convertBodyTo type="java.lang.String" charset="UTF-8"/>
<unmarshal ref="ReaderDataFormat" />
<log message="Content File size received ${body.size}" />
<split streaming="true" parallelProcessing="false">
<simple>${body}</simple>
<choice>
<when>
<simple></simple>
<setHeader
headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
<process ref="BodyEnricherProcessor" />
<to
uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=java.util.LinkedHashMap"
/>
<marshal ref="Gson" />
<log message="Message transformed ${in.header.CamelSplitIndex} -
${body}" />
<to uri="activemq:queue:IF_CBIKIT" />
</when>
<otherwise>
<log message="Message discarded ${in.header.CamelSplitIndex} -
${body}" />
</otherwise>
</choice>
</split>
</route>
Now Out Of Memory is generated by:
Heap dumped on OutOfMemoryError exception
Thread causing OutOfMemoryError exception: Camel
(IF_CBIKIT-Inbound-Context) thread #18 - seda://processAndStoreInQueue
Please help me!
Thanks a lot in advance
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780965.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
I'm here again because I don't resolved my problem.
After several checks, I noticed this:
memory-usage.png
<http://camel.465427.n5.nabble.com/file/n5780965/memory-usage.png>
Why does thread related to seda://processAndStoreInQueue consume much
memory? How to optimize memory usage?
This is my route to manage file (in this case is a csv with 40000 rows) with
large number of rows:
<route id="FileRetriever_Route">
<from
uri="{{uri.inbound}}?scheduler=quartz2&scheduler.cron={{poll.consumer.scheduler}}&scheduler.triggerId=FileRetriever&scheduler.triggerGroup=IF_CBIKIT{{uri.inbound.options}}"
/>
<setHeader
headerName="ImportDateTime"><simple>${date:now:yyyyMMdd-HHmmss}</simple></setHeader>
<setHeader
headerName="MsgCorrelationId"><simple>CBIKIT_INBOUND_${in.header.ImportDateTime}</simple></setHeader>
<setHeader headerName="breadcrumbId">
<simple>Import-${in.header.CamelFileName}-${in.header.ImportDateTime}-${in.header.breadcrumbId}</simple>
</setHeader>
<log message="START - FileRetriever_Route - Found file
${in.header.CamelFileName}" />
<to uri="seda:processAndStoreInQueue" />
<log message="END - FileRetriever_Route" />
</route>
<route id="ProcessAndStoreInQueue_Route" >
<from uri="seda:processAndStoreInQueue" />
<convertBodyTo type="java.lang.String" charset="UTF-8"/>
<unmarshal ref="ReaderDataFormat" />
<log message="Content File size received ${body.size}" />
<split streaming="true" parallelProcessing="false">
<simple>${body}</simple>
<choice>
<when>
<simple></simple>
<setHeader
headerName="CamelSplitIndex"><simple>${in.header.CamelSplitIndex}</simple></setHeader>
<process ref="BodyEnricherProcessor" />
<to
uri="dozer:transform?mappingFile=file:{{crt2.apps.home}}{{dozer.mapping.path}}&targetModel=java.util.LinkedHashMap"
/>
<marshal ref="Gson" />
<log message="Message transformed ${in.header.CamelSplitIndex} -
${body}" />
<to uri="activemq:queue:IF_CBIKIT" />
</when>
<otherwise>
<log message="Message discarded ${in.header.CamelSplitIndex} -
${body}" />
</otherwise>
</choice>
</split>
</route>
Now Out Of Memory is generated by:
Heap dumped on OutOfMemoryError exception
Thread causing OutOfMemoryError exception: Camel
(IF_CBIKIT-Inbound-Context) thread #18 - seda://processAndStoreInQueue
Please help me!
Thanks a lot in advance
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780965.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Ranx <br...@mediadriver.com>.
Cool. It's good to see you're able to chip away at the problem. This is
another section you may want to look at as well. Christian has good
explanation of it: https://dzone.com/articles/activemq-understanding-memory
This is why when I'm starting on proof of concepts, spikes or protos I tend
to stay with SEDA or direct until I know that I need the persistence or
spooling. The tempUsage is where AMQ defines how much memory to use when
writing non-persistent messages to hard disk cache. Sounds strange, I know,
but it's just using it as an extension of RAM really.
When you start paging or caching objects you're going to lose several orders
of magnitude in speed. The objects get taken from nanosecond RAM speeds to
write out at millisecond hard disk speeds and then it reads it again from
millisecond hard disk speeds back up into RAM again.
Here's another good link:
http://activemq.apache.org/javalangoutofmemory.html
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage percentOfJvmHeap="70" />
</memoryUsage>
<storeUsage>
<storeUsage limit="100 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="50 gb"/>
</tempUsage>
</systemUsage>
</systemUsage>
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780169.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
yesterday I worked on JMS Configuration and following your suggestion I
changed it like this avoiding OOM
<policyEntry queue=">" producerFlowControl="true" memoryLimit="1024mb"
optimizedDispatch="true">
<pendingQueuePolicy>
<vmQueueCursor />
</pendingQueuePolicy>
<pendingMessageLimitStrategy>
<constantPendingMessageLimitStrategy limit="1000"/>
</pendingMessageLimitStrategy>
</policyEntry>
and on tag systemUsage I set sendFailIfNoSpace="false".
I'll work on this way to improve performance.
Anyway, thanks a lot for your support.
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780132.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Ranx <br...@mediadriver.com>.
Are you eventually going to separate these queues in different locations?
One thing that's happening when you use JMS this way is you read the records
in, and then you immediately write them back out over a socket and then read
them back in. Those all involve duplication of memory.
Since you may be doing it for transactional purposes then you need to make
sure your LevelDB is set up to handle the data you send to it. I've used
KahaDB with AMQ but not LevelDB though they are for the same purposes. In
this case you'll need to specify the maximum number of in memory objects and
then set policies for physical storage of the elements on disk.
http://activemq.apache.org/xml-configuration.html
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780053.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
yes... it is clear ;)!
I changed configuration of Heap and Perm progressively as you suggested like
this:
-server -Xms1024M -Xmx2048M -XX:PermSize=256M -XX:MaxPermSize=512M -Xss512M
-XX:+HeapDumpOnOutOfMemoryError -verbose:gc -Xloggc:gc.log ...
The number of rows processed has increased, unfortunately the result is the
same OOM.
With VisualVM, I have traced Threads and Memory and the biggest object found
is
org.apache.activemq.broker.region.Queue#1 Retained Size Memory
1.449.034.641 B
I attached images of heap dump of import result for further information
biggest-objects.png
<http://camel.465427.n5.nabble.com/file/n5780047/biggest-objects.png>
memory-threads.png
<http://camel.465427.n5.nabble.com/file/n5780047/memory-threads.png>
And then in log of gc, i found this
"ActiveMQ Transport: tcp:///127.0.0.1:63664@61616" daemon prio=5 tid=330
RUNNABLE
at java.lang.OutOfMemoryError.<init>(OutOfMemoryError.java:48)
at
org.apache.activemq.util.DataByteArrayOutputStream.ensureEnoughBuffer(DataByteArrayOutputStream.java:249)
at
org.apache.activemq.util.DataByteArrayOutputStream.writeBoolean(DataByteArrayOutputStream.java:140)
at
org.apache.activemq.openwire.v6.BaseDataStreamMarshaller.looseMarshalByteSequence(BaseDataStreamMarshaller.java:628)
Local Variable: org.apache.activemq.util.ByteSequence#52
at
org.apache.activemq.openwire.v6.MessageMarshaller.looseMarshal(MessageMarshaller.java:296)
at
org.apache.activemq.openwire.v6.ActiveMQMessageMarshaller.looseMarshal(ActiveMQMessageMarshaller.java:111)
at
org.apache.activemq.openwire.v6.ActiveMQObjectMessageMarshaller.looseMarshal(ActiveMQObjectMessageMarshaller.java:111)
Local Variable: org.apache.activemq.util.DataByteArrayOutputStream#2
at
org.apache.activemq.openwire.OpenWireFormat.marshal(OpenWireFormat.java:155)
Local Variable:
org.apache.activemq.openwire.v6.ActiveMQObjectMessageMarshaller#1
Local Variable: org.apache.activemq.openwire.OpenWireFormat#2
at
org.apache.activemq.leveldb.DelayableUOW.create_message_record$1(DBManager.scala:299)
at org.apache.activemq.leveldb.DelayableUOW.enqueue(DBManager.scala:311)
Local Variable: org.apache.activemq.command.MessageId#42
at
org.apache.activemq.leveldb.LevelDBStore$LevelDBMessageStore.doAdd(LevelDBStore.scala:773)
Local Variable: java.util.concurrent.atomic.AtomicLong#133
Local Variable: java.lang.Long#47
at
org.apache.activemq.leveldb.LevelDBStore$LevelDBMessageStore$$anonfun$asyncAddQueueMessage$1.apply(LevelDBStore.scala:785)
at
org.apache.activemq.leveldb.LevelDBStore$LevelDBMessageStore$$anonfun$asyncAddQueueMessage$1.apply(LevelDBStore.scala:784)
at org.apache.activemq.leveldb.LevelDBStore.withUow(LevelDBStore.scala:735)
Local Variable: org.apache.activemq.leveldb.DelayableUOW#1
Local Variable:
org.apache.activemq.leveldb.LevelDBStore$LevelDBMessageStore$$anonfun$asyncAddQueueMessage$1#1
Local Variable:
org.apache.activemq.store.leveldb.LevelDBPersistenceAdapter#1
at
org.apache.activemq.leveldb.LevelDBStore$LevelDBMessageStore.asyncAddQueueMessage(LevelDBStore.scala:784)
Local Variable:
org.apache.activemq.leveldb.LevelDBStore$LevelDBMessageStore#1
at org.apache.activemq.broker.region.Queue.doMessageSend(Queue.java:829)
at org.apache.activemq.broker.region.Queue.send(Queue.java:723)
Local Variable: org.apache.activemq.broker.region.Queue#1
Local Variable: org.apache.activemq.state.ProducerState#2
Local Variable: org.apache.activemq.command.ProducerInfo#3
at
org.apache.activemq.broker.region.DestinationFilter.send(DestinationFilter.java:132)
Local Variable:
org.apache.activemq.security.AuthorizationDestinationFilter#1
at
org.apache.activemq.broker.region.AbstractRegion.send(AbstractRegion.java:419)
Local Variable: org.apache.activemq.broker.jmx.ManagedQueueRegion#1
at
org.apache.activemq.broker.region.RegionBroker.send(RegionBroker.java:468)
at
org.apache.activemq.broker.jmx.ManagedRegionBroker.send(ManagedRegionBroker.java:297)
Local Variable: org.apache.activemq.broker.jmx.ManagedRegionBroker#1
at org.apache.activemq.broker.BrokerFilter.send(BrokerFilter.java:152)
Local Variable: org.apache.activemq.advisory.AdvisoryBroker#1
at
org.apache.activemq.broker.CompositeDestinationBroker.send(CompositeDestinationBroker.java:96)
Local Variable: org.apache.activemq.command.ActiveMQQueue#27
Local Variable: org.apache.activemq.broker.CompositeDestinationBroker#1
at
org.apache.activemq.broker.TransactionBroker.send(TransactionBroker.java:293)
Local Variable: org.apache.activemq.broker.ConnectionContext#3
Local Variable: org.apache.activemq.broker.TransactionBroker#1
at org.apache.activemq.broker.BrokerFilter.send(BrokerFilter.java:152)
Local Variable: org.apache.activemq.security.JaasAuthenticationBroker#1
at org.apache.activemq.broker.BrokerFilter.send(BrokerFilter.java:152)
at
org.apache.activemq.security.AuthorizationBroker.send(AuthorizationBroker.java:223)
Local Variable:
org.apache.activemq.security.JaasAuthenticationBroker$JaasSecurityContext#2
Local Variable: org.apache.activemq.security.AuthorizationBroker#1
at
org.apache.activemq.broker.MutableBrokerFilter.send(MutableBrokerFilter.java:157)
Local Variable: org.apache.activemq.broker.BrokerService$5#1
at
org.apache.activemq.broker.TransportConnection.processMessage(TransportConnection.java:542)
Local Variable: org.apache.activemq.broker.ProducerBrokerExchange#2
Local Variable: org.apache.activemq.command.ProducerId#39
at
org.apache.activemq.command.ActiveMQMessage.visit(ActiveMQMessage.java:768)
at
org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:334)
Local Variable:
org.apache.activemq.broker.jmx.ManagedTransportConnection#2
at
org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:188)
Local Variable: org.apache.activemq.broker.TransportConnection$1#2
at
org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
Local Variable: org.apache.activemq.transport.MutexTransport#4
at
org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:113)
Local Variable: org.apache.activemq.transport.WireFormatNegotiator#4
at
org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:300)
Local Variable: org.apache.activemq.transport.InactivityMonitor#4
at
org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
at
org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:214)
Local Variable: org.apache.activemq.command.ActiveMQObjectMessage#27
at
org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:196)
Local Variable: org.apache.activemq.transport.tcp.TcpTransport#4
at java.lang.Thread.run(Thread.java:745)
How can I improve the Store and forward?
Thanks a log again for your support.
Greetings
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780047.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by "mailinglist@j-b-s.de" <ma...@j-b-s.de>.
Hi Michele!
Your mem settings look odd to mee.
> -server -Xms256M -Xss512M -Xmx512M -XX:+UnlockDiagnosticVMOptions
> -XX:+UnsyncloadClass -XX:PermSize=512M -XX:MaxPermSize=1024M
Your Perm Space is greater than your heap? Sure? I doubt you need 1G set your heap to 1G and Perm to 256m as a first try abd monitor your app with jconsole. Java 8 does not use perm settings any more.
OOM: The outofmem error occurs because your mem is exhausted and is far away from "normal". Nevertheless when this happens the currently visible stacktrace just tells you where it fails but not why or what the root cause is (what creates all the objects, why are they not garbage collectable aso) and this is not necessarily activemq.
If you open your refrigirator and its totally empty, you don't know either who to blame for grabbing the last beer. So activemq is you, the refrigirator is your VM and the beer is your heap... See what I mean?
Jens
Von meinem iPhone gesendet
> Am 30.03.2016 um 10:12 schrieb Michele <mi...@finconsgroup.com>:
>
> Hi all,
>
> first of all, thanks so much for your support.
>
> I configured routing with SEDA Component configured like this:
>
> I declared Bean with queueSize
> <bean id="myseda" class="org.apache.camel.component.seda.SedaComponent">
> <property name="queueSize" value="150" />
> </bean>
>
> and then
>
> <route id="FileReader_Route">
> <from uri="file:incoming?....." />
> <split streaming="true" parallelProcessing="true">
> <tokenize token="\n" />
> <unmarshal ref="IncomingCSVFileDataFormat" />
> <process ref="DataProcessor" />
> <marshal ref="Gson" />
> <to uri="myseda:IF_INGESTATE_Inbound?blockWhenFull=true" />
> </split>
> </route>
>
> <route id="ProcessTicket_Route">
> <from uri="myseda:IF_INGESTATE_Inbound" />
> <throttle timePeriodMillis="10000" asyncDelayed="true">
> <constant>5</constant>
> <to
> uri="jetty:http://host/rs/v1.0/ticket?jettyHttpBindingRef=CustomJettyHttpBinding"
> />
> </throttle>
> </route>
>
> and everything worked fine and very fast.
>
> What is the meaning of "But now, I have an error java.lang.OutOfMemoryError:
> Java heap space on
> ActiveMQ"?
>
> Sorry, but I'm new in using Apache Camel and ActiveMQ... Is it normal that
> Active MQ works OutOfMemory?
>
> Here JMS queue configuration:
>
> <bean id="jmsConnectionFactory"
> class="org.apache.activemq.ActiveMQConnectionFactory">
> <property name="brokerURL" value="tcp://localhost:61616" />
> <property name="userName" value="admin" />
> <property name="password" value="admin" />
> <property name="producerWindowSize" value="1024000" />
> </bean>
>
> <bean id="pooledConnectionFactory"
> class="org.apache.activemq.pool.PooledConnectionFactory"
> init-method="start" destroy-method="stop">
> <property name="maxConnections" value="5" />
> <property name="connectionFactory" ref="jmsConnectionFactory" />
> </bean>
>
> <bean id="jmsConfigNoTx"
> class="org.apache.camel.component.jms.JmsConfiguration">
> <property name="connectionFactory" ref="pooledConnectionFactory" />
> <property name="concurrentConsumers" value="3" />
> <property name="transferExchange" value="true" />
> </bean>
>
>
> <bean id="activemq"
> class="org.apache.activemq.camel.component.ActiveMQComponent">
> <property name="configuration" ref="jmsConfigNoTx" />
> </bean>
>
> and then (following these topics
> http://activemq.apache.org/javalangoutofmemory.html,
> http://activemq.apache.org/how-do-i-configure-10s-of-1000s-of-queues-in-a-single-broker-.html)
> I tried to change file activemq.xml in ${JBossFuseInstDir}/etc like this
>
> <destinationPolicy>
> <policyMap>
> <policyEntries>
> <policyEntry topic=">" producerFlowControl="true">
> <pendingMessageLimitStrategy>
> <constantPendingMessageLimitStrategy limit="1000"/>
> </pendingMessageLimitStrategy>
> </policyEntry>
> <policyEntry queue=">" producerFlowControl="true"
> memoryLimit="1024mb" optimizedDispatch="true">
>
> <pendingQueuePolicy>
> <storeCursor />
> </pendingQueuePolicy>
> </policyEntry>
> </policyEntries>
> </policyMap>
> </destinationPolicy>
>
> ....
> <persistenceAdapter>
>
> <levelDB directory="${data}/leveldb" />
> </persistenceAdapter>
>
> <systemUsage>
> <systemUsage>
> <memoryUsage>
> <memoryUsage percentOfJvmHeap="80"/>
> </memoryUsage>
> <storeUsage>
> <storeUsage limit="100 gb"/>
> </storeUsage>
> <tempUsage>
> <tempUsage limit="50 gb"/>
> </tempUsage>
> </systemUsage>
> </systemUsage>
>
> and I launch JB Fuse with VM Arguments
> -server -Xms256M -Xss512M -Xmx512M -XX:+UnlockDiagnosticVMOptions
> -XX:+UnsyncloadClass -XX:PermSize=512M -XX:MaxPermSize=1024M
> -Dorg.apache.activemq.UseDedicatedTaskRunner=false
>
> That's all...
>
> I tried the same process done with Seda with Active MQ and soon I have seen
> in the log file
>
> 09:23:58,948 | ERROR | Inbound_Worker-2 | DefaultErrorHandler |
> 198 - org.apache.camel.camel-core - 2.15.1.redhat-620133 -
> IF_INGESTATE-Inbound-Context - IMP-IF-Ingestate-20160330-090900 | Failed
> delivery for (MessageId: ID-FGBAL201530-54640-1459321665233-0-2684 on
> ExchangeId: ID-FGBAL201530-54640-1459321665233-0-2685). Exhausted after
> delivery attempt: 1 caught: org.apache.camel.TypeConversionException: Error
> during type conversion from type: java.lang.String to the required type:
> byte[] with value [Body is instance of java.io.InputStream] due
> java.lang.OutOfMemoryError: Java heap space
>
> and then LevelDB Stop and the process ends...
>
> I attach first OOM excpetion occurred during process oof.txt
> <http://camel.465427.n5.nabble.com/file/n5780003/oof.txt>
>
> Thanks for all
>
> Best regards
>
> Michele
>
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780003.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
I forgot
this link http://activemq.apache.org/what-is-the-prefetch-limit-for.html and
so I changed the consumer endpoint like this
<route id="ProcessTicket_Route">
<from
uri="activemq:queue:IF-INGESTATE_INBOUND?destination.consumer.prefetchSize=0"/>
....
</route>
Thanks again.
Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780004.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi all,
first of all, thanks so much for your support.
I configured routing with SEDA Component configured like this:
I declared Bean with queueSize
<bean id="myseda" class="org.apache.camel.component.seda.SedaComponent">
<property name="queueSize" value="150" />
</bean>
and then
<route id="FileReader_Route">
<from uri="file:incoming?....." />
<split streaming="true" parallelProcessing="true">
<tokenize token="\n" />
<unmarshal ref="IncomingCSVFileDataFormat" />
<process ref="DataProcessor" />
<marshal ref="Gson" />
<to uri="myseda:IF_INGESTATE_Inbound?blockWhenFull=true" />
</split>
</route>
<route id="ProcessTicket_Route">
<from uri="myseda:IF_INGESTATE_Inbound" />
<throttle timePeriodMillis="10000" asyncDelayed="true">
<constant>5</constant>
<to
uri="jetty:http://host/rs/v1.0/ticket?jettyHttpBindingRef=CustomJettyHttpBinding"
/>
</throttle>
</route>
and everything worked fine and very fast.
What is the meaning of "But now, I have an error java.lang.OutOfMemoryError:
Java heap space on
ActiveMQ"?
Sorry, but I'm new in using Apache Camel and ActiveMQ... Is it normal that
Active MQ works OutOfMemory?
Here JMS queue configuration:
<bean id="jmsConnectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
<property name="userName" value="admin" />
<property name="password" value="admin" />
<property name="producerWindowSize" value="1024000" />
</bean>
<bean id="pooledConnectionFactory"
class="org.apache.activemq.pool.PooledConnectionFactory"
init-method="start" destroy-method="stop">
<property name="maxConnections" value="5" />
<property name="connectionFactory" ref="jmsConnectionFactory" />
</bean>
<bean id="jmsConfigNoTx"
class="org.apache.camel.component.jms.JmsConfiguration">
<property name="connectionFactory" ref="pooledConnectionFactory" />
<property name="concurrentConsumers" value="3" />
<property name="transferExchange" value="true" />
</bean>
<bean id="activemq"
class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="configuration" ref="jmsConfigNoTx" />
</bean>
and then (following these topics
http://activemq.apache.org/javalangoutofmemory.html,
http://activemq.apache.org/how-do-i-configure-10s-of-1000s-of-queues-in-a-single-broker-.html)
I tried to change file activemq.xml in ${JBossFuseInstDir}/etc like this
<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry topic=">" producerFlowControl="true">
<pendingMessageLimitStrategy>
<constantPendingMessageLimitStrategy limit="1000"/>
</pendingMessageLimitStrategy>
</policyEntry>
<policyEntry queue=">" producerFlowControl="true"
memoryLimit="1024mb" optimizedDispatch="true">
<pendingQueuePolicy>
<storeCursor />
</pendingQueuePolicy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>
....
<persistenceAdapter>
<levelDB directory="${data}/leveldb" />
</persistenceAdapter>
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage percentOfJvmHeap="80"/>
</memoryUsage>
<storeUsage>
<storeUsage limit="100 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="50 gb"/>
</tempUsage>
</systemUsage>
</systemUsage>
and I launch JB Fuse with VM Arguments
-server -Xms256M -Xss512M -Xmx512M -XX:+UnlockDiagnosticVMOptions
-XX:+UnsyncloadClass -XX:PermSize=512M -XX:MaxPermSize=1024M
-Dorg.apache.activemq.UseDedicatedTaskRunner=false
That's all...
I tried the same process done with Seda with Active MQ and soon I have seen
in the log file
09:23:58,948 | ERROR | Inbound_Worker-2 | DefaultErrorHandler |
198 - org.apache.camel.camel-core - 2.15.1.redhat-620133 -
IF_INGESTATE-Inbound-Context - IMP-IF-Ingestate-20160330-090900 | Failed
delivery for (MessageId: ID-FGBAL201530-54640-1459321665233-0-2684 on
ExchangeId: ID-FGBAL201530-54640-1459321665233-0-2685). Exhausted after
delivery attempt: 1 caught: org.apache.camel.TypeConversionException: Error
during type conversion from type: java.lang.String to the required type:
byte[] with value [Body is instance of java.io.InputStream] due
java.lang.OutOfMemoryError: Java heap space
and then LevelDB Stop and the process ends...
I attach first OOM excpetion occurred during process oof.txt
<http://camel.465427.n5.nabble.com/file/n5780003/oof.txt>
Thanks for all
Best regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780003.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Ranx <br...@mediadriver.com>.
I think you're hitting a lot of good points there. I'm not used to CSVs with
100 columns of data but can see why that could get huge.
If she starts with SEDA and just sets a queue size of something like 100 or
200 and then sets blockWhenFull to true her streaming will halt until the
queue can take more. So she won't have all the file in memory that way.
She didn't post what the JMS queue configuration looks like so it's hard to
tell what caused the OOM. Personally that's why I like to start with the
simple SEDA even if I know I'm not going to end up there. Everything is in
memory so there shouldn't be replication. Threading is fairly simple and she
has a bottleneck at her REST endpoint. If all the marshalling and data
conversion is done before the objects are put on the queue then she has some
nice granularity to be able to determine exactly what her throughput at that
point is. In other words, figuring out how many threads she needs to
communicate with the REST endpoint doesn't have confounding factors like the
threads doing processing and GSon conversion. It is just about how fast can
I take a GSon model off the queue, invoke the REST call and get the data
back.
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5779994.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Jens Breitenstein <ma...@j-b-s.de>.
Sorry Ranx, missed your previous post.
From our experience: We ran into OOM troubles with CVS files when
converting a row to Map<String, String> for example.
Depending on the number of columns (we have aprox 100+) this can quite
easily eat up the entire memory (sure you can always provide more), but
maps are not really memory friendly.
And streaming does not necessarily imply memory is immediately freed
after the row has been read, camel needs to acknowledge a message
(containing a unmarshalled row) has been processed.
We are using streaming of csv's parallelized, with SEDA and a limited
amount of rows in mem (less than 100) and everything works fast with
really low mem consumption.
And to be honest: we have the CSV row as string, the unmarshalled camel
message (maybe using Map<String, String>) and the serialized JMS
representation in memory at the same time for each csv-row, data is
already duplicated, no way to prevent this.
There is one thing left which puzzles me: what is the meaning of "But
now, I have an error java.lang.OutOfMemoryError: Java heap space on
ActiveMQ"? This is just where the OOM happens,
but not necessarily the root cause. Maybe JConsole will be of help to
track down memory usage and number of objects next
Jens
Am 29/03/16 um 20:11 schrieb Ranx:
> Jens,
>
> That's why I suggested setting the limit on the queue size. She has
> streaming turned on already so I believe that will block when the queue
> (SEDA or JMS) gets full. But 50,000 objects isn't usually that much memory
> so there may be something else in the JMS settings that is actually
> marshalling and unmarshalling all the data. It may not be a local in memory
> queue. That would end up duplicating everything. And it appears that the
> out of memory is associated with AMQ itself. Without digging into that can
> of worms immediately it is easy enough to switch over to SEDA just to get
> the functionality working and throughput parameters established and then
> take a look at what might be happening with JMS.
>
> The SEDA queue has a "blockWhenFull" setting that should make that
> relatively easy. It's hard to say whether the overhead of JMS is necessary
> in this case without knowing the actual business case and transactional
> integrity requirements.
>
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5779989.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Ranx <br...@mediadriver.com>.
Jens,
That's why I suggested setting the limit on the queue size. She has
streaming turned on already so I believe that will block when the queue
(SEDA or JMS) gets full. But 50,000 objects isn't usually that much memory
so there may be something else in the JMS settings that is actually
marshalling and unmarshalling all the data. It may not be a local in memory
queue. That would end up duplicating everything. And it appears that the
out of memory is associated with AMQ itself. Without digging into that can
of worms immediately it is easy enough to switch over to SEDA just to get
the functionality working and throughput parameters established and then
take a look at what might be happening with JMS.
The SEDA queue has a "blockWhenFull" setting that should make that
relatively easy. It's hard to say whether the overhead of JMS is necessary
in this case without knowing the actual business case and transactional
integrity requirements.
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5779989.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Jens Breitenstein <ma...@j-b-s.de>.
To me it looks like the entire CSV file is converted to activemq
messages which finally causes the OOM.
The same problem exists with Camel SEDA queues as the messages need to
be kept in memory until handled.
I guess you need to set a queue size limit which blocks your csv sending
process if no space is left.
see: http://activemq.apache.org/producer-flow-control.html
Jens
Am 29/03/16 um 10:12 schrieb Michele:
> Hi,
>
> thanks a lot for your reply.
>
> I changed my route introducing Throttler Pattern like this:
>
> <route id="FileReader_Route">
> <from uri="file:incoming?....." />
> <split streaming="true" parallelProcessing="true">
> <tokenize token="\n" />
> <unmarshal ref="IncomingCSVFileDataFormat" />
> .....
> <to uri="activemq:queue:incomingTickets" />
> </split>
> </route>
>
> <route id="ProcessTicket_Route">
> <from uri="activemq:queue:incomingTickets" />
> <throttle timePeriodMillis="10000" asyncDelayed="true">
> <constant>5</constant>
> <process ref="DataProcessor" />
> <marshal ref="Gson" />
> <to
> uri="jetty:http://host/rs/v1.0/ticket?jettyHttpBindingRef=CustomJettyHttpBinding"
> />
> </throttle>
> </route>
>
> But now, I have an error java.lang.OutOfMemoryError: Java heap space on
> ActiveMQ... I defined a PooledConnectionFactory with concurrentConsumers
> equals 5 and maxConnections equals 3
>
> .....
> Message History
> ---------------------------------------------------------------------------------------------------------------------------------------
> RouteId ProcessorId Processor
> Elapsed (ms)
> [FileRetriever_Rout] [FileRetriever_Rout]
> [file://C:/inbox?include=%5EEdenred_%5B0-9%5D%7B8%7D.csv&s] [ 46034]
> [FileRetriever_Rout] [unmarshal3 ]
> [unmarshal[ref:IncomingFileDataFormat]
> ] [ 0]
> [FileRetriever_Rout] [setHeader17 ] [setHeader[CamelSplitIndex]
> ] [ 0]
> [FileRetriever_Rout] [setBody1 ] [setBody[simple{${body[0]}}]
> ] [ 10]
> [FileRetriever_Rout] [choice2 ]
> [when[simple{${body[INGENICO_OPERATION_ID]} regex '[0-9]+'}]choice[]
> ] [ 1634]
> [FileRetriever_Rout] [log15 ] [log
> ] [ 0]
> [FileRetriever_Rout] [to6 ]
> [activemq:queue:IF_INGESTATE_Inbound?destination.consumer.dispatchAsync=true&de]
> [ 1634]
>
> Exchange
> ---------------------------------------------------------------------------------------------------------------------------------------
> Exchange[
> Id ID-FGBAL201530-50191-1459236743101-1-259
> ExchangePattern InOnly
> Headers {breadcrumbId=IMP-IF-Ingestate-20160329-093400,
> CamelFileAbsolute=true,
> CamelFileAbsolutePath=C:\CRT-2.0\IF-Ingestate\inbox\Edenred_15092015.csv,
> CamelFileContentType=application/vnd.ms-excel,
> CamelFileLastModified=1458774180380, CamelFileLength=6961789,
> CamelFileName=Edenred_15092015.csv,
> CamelFileNameConsumed=Edenred_15092015.csv,
> CamelFileNameOnly=Edenred_15092015.csv,
> CamelFileParent=C:\CRT-2.0\IF-Ingestate\inbox,
> CamelFilePath=C:\CRT-2.0\IF-Ingestate\inbox\Edenred_15092015.csv,
> CamelFileRelativePath=Edenred_15092015.csv, CamelRedelivered=false,
> CamelRedeliveryCounter=0, CamelSplitIndex=59,
> CrmRSActionPath=/tk_rt_ticket/ingestate/maintenance,
> ImportDateTime=20160329-093400,
> MsgCorrelationId=Inbound_INGESTATE_20160329-093400}
> BodyType java.util.HashMap
> Body {TICKET_ID=00108560, INGENICO_OPERATION_ID=1721710,
> TERMINAL_NUMBEROFCONTACTS=1, TICKET_STATUS=1,
> TERMINAL_LAST_CONTACT_DATE=2012-11-03 01:57:26.360,
> TERMINAL_TECHNOLOGY=TELIUM, TERMINAL_SN=0000107370643471,
> TERMINAL_INGESTATE_ID=ICT-TICKET-RESTAURANT:91215343,
> TICKET_REGISTRATION_DATE=2012-11-02 10:08:00.000, TERMINAL_PN=M40,
> TERMINAL_FIRST_CONTACT_DATE=2012-11-03 01:56:08.343, TERMINAL_ID=91215343,
> RECORDER=EDENRED}
> ]
>
> Stacktrace
> ---------------------------------------------------------------------------------------------------------------------------------------
> org.apache.camel.CamelExecutionException: Exception occurred during
> execution on the exchange: Exchange[Message: {TICKET_ID=00108560,
> INGENICO_OPERATION_ID=1721710, TERMINAL_NUMBEROFCONTACTS=1, TICKET_STATUS=1,
> TERMINAL_LAST_CONTACT_DATE=2012-11-03 01:57:26.360,
> TERMINAL_TECHNOLOGY=TELIUM, TERMINAL_SN=0000107370643471,
> TERMINAL_INGESTATE_ID=ICT-TICKET-RESTAURANT:91215343,
> TICKET_REGISTRATION_DATE=2012-11-02 10:08:00.000, TERMINAL_PN=M40,
> TERMINAL_FIRST_CONTACT_DATE=2012-11-03 01:56:08.343, TERMINAL_ID=91215343,
> RECORDER=EDENRED}]
> at
> org.apache.camel.util.ObjectHelper.wrapCamelExecutionException(ObjectHelper.java:1635)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.impl.DefaultExchange.setException(DefaultExchange.java:308)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.component.jms.JmsProducer.process(JmsProducer.java:158)[209:org.apache.camel.camel-jms:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.SendProcessor.process(SendProcessor.java:129)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:118)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:80)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.ChoiceProcessor.process(ChoiceProcessor.java:111)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:118)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:80)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:652)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:580)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.MulticastProcessor.process(MulticastProcessor.java:227)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.Splitter.process(Splitter.java:104)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:118)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:80)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:435)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:211)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:175)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:174)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:101)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
> at
> org.apache.camel.pollconsumer.quartz2.QuartzScheduledPollConsumerJob.execute(QuartzScheduledPollConsumerJob.java:59)[279:org.apache.camel.camel-quartz2:2.15.1.redhat-620133]
> at
> org.quartz.core.JobRunShell.run(JobRunShell.java:202)[278:org.quartz-scheduler.quartz:2.2.1]
> at
> org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:573)[278:org.quartz-scheduler.quartz:2.2.1]
> Caused by: java.lang.OutOfMemoryError: Java heap space
>
>
> Any idea?
>
> How configure active mq component in order to manage the load in JBoss Fuse
> 6.2 context? Or Is there something wrong in my system routing?
>
> Thanks in advance
>
> Best Regards
> Michele
>
>
>
>
>
>
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5779935.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Ranx <br...@mediadriver.com>.
There are a number of answers to that question but this should be relatively
easy to fix. Since you are running out of memory you should probably bump
the max memory in the karaf startup batch file. But you will also want to
limit the number of rows that you bring into memory. You already have the
consumer set to streaming which is what you want.
Since you are using the ActiveMQ component you can set the maximum number of
items on the queue. You could also use the SEDA or disruptor queues and set
the size from 100 to 1,000 limiting the number of in memory objects. You
may want to try that just for testing purposes and then switch back to JMS.
You don't appear to be using JMS here for its distributed capabilities nor
for its ability to have durable subscribers or persistence mechanics on the
queue. The SEDA component is much simpler in that it is just a List/queue
in memory with configurable numbers of threads associated with it. Doing
that will let you see if you still have a memory problem.
Later, if you decide that you want JMS in the mix then you can easily swap
that back in but then only deal with the problems associated with the queue
and AMQ configuration. One item you could then look at is having the queue
set up to limit the number of in memory objects but have the queue itself
backed by the persistent store (kahadb). That's very simple to do and ships
as a standard out of the box configuration.
You might also want to consider putting the processor and Gson marshalling
prior to the queue if the REST endpoint is the true bottleneck. The
incoming data can be unmarshalled, processed, converted via Gson and then
put on the queue. When the consumer takes it off the queue it does exactly
one thing with it - invokes the REST endpoint.
How big are these objects. You'd mentioned that there are only 50,000 rows
in the file so that isn't much.
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5779973.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Michele <mi...@finconsgroup.com>.
Hi,
thanks a lot for your reply.
I changed my route introducing Throttler Pattern like this:
<route id="FileReader_Route">
<from uri="file:incoming?....." />
<split streaming="true" parallelProcessing="true">
<tokenize token="\n" />
<unmarshal ref="IncomingCSVFileDataFormat" />
.....
<to uri="activemq:queue:incomingTickets" />
</split>
</route>
<route id="ProcessTicket_Route">
<from uri="activemq:queue:incomingTickets" />
<throttle timePeriodMillis="10000" asyncDelayed="true">
<constant>5</constant>
<process ref="DataProcessor" />
<marshal ref="Gson" />
<to
uri="jetty:http://host/rs/v1.0/ticket?jettyHttpBindingRef=CustomJettyHttpBinding"
/>
</throttle>
</route>
But now, I have an error java.lang.OutOfMemoryError: Java heap space on
ActiveMQ... I defined a PooledConnectionFactory with concurrentConsumers
equals 5 and maxConnections equals 3
.....
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor
Elapsed (ms)
[FileRetriever_Rout] [FileRetriever_Rout]
[file://C:/inbox?include=%5EEdenred_%5B0-9%5D%7B8%7D.csv&s] [ 46034]
[FileRetriever_Rout] [unmarshal3 ]
[unmarshal[ref:IncomingFileDataFormat]
] [ 0]
[FileRetriever_Rout] [setHeader17 ] [setHeader[CamelSplitIndex]
] [ 0]
[FileRetriever_Rout] [setBody1 ] [setBody[simple{${body[0]}}]
] [ 10]
[FileRetriever_Rout] [choice2 ]
[when[simple{${body[INGENICO_OPERATION_ID]} regex '[0-9]+'}]choice[]
] [ 1634]
[FileRetriever_Rout] [log15 ] [log
] [ 0]
[FileRetriever_Rout] [to6 ]
[activemq:queue:IF_INGESTATE_Inbound?destination.consumer.dispatchAsync=true&de]
[ 1634]
Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
Id ID-FGBAL201530-50191-1459236743101-1-259
ExchangePattern InOnly
Headers {breadcrumbId=IMP-IF-Ingestate-20160329-093400,
CamelFileAbsolute=true,
CamelFileAbsolutePath=C:\CRT-2.0\IF-Ingestate\inbox\Edenred_15092015.csv,
CamelFileContentType=application/vnd.ms-excel,
CamelFileLastModified=1458774180380, CamelFileLength=6961789,
CamelFileName=Edenred_15092015.csv,
CamelFileNameConsumed=Edenred_15092015.csv,
CamelFileNameOnly=Edenred_15092015.csv,
CamelFileParent=C:\CRT-2.0\IF-Ingestate\inbox,
CamelFilePath=C:\CRT-2.0\IF-Ingestate\inbox\Edenred_15092015.csv,
CamelFileRelativePath=Edenred_15092015.csv, CamelRedelivered=false,
CamelRedeliveryCounter=0, CamelSplitIndex=59,
CrmRSActionPath=/tk_rt_ticket/ingestate/maintenance,
ImportDateTime=20160329-093400,
MsgCorrelationId=Inbound_INGESTATE_20160329-093400}
BodyType java.util.HashMap
Body {TICKET_ID=00108560, INGENICO_OPERATION_ID=1721710,
TERMINAL_NUMBEROFCONTACTS=1, TICKET_STATUS=1,
TERMINAL_LAST_CONTACT_DATE=2012-11-03 01:57:26.360,
TERMINAL_TECHNOLOGY=TELIUM, TERMINAL_SN=0000107370643471,
TERMINAL_INGESTATE_ID=ICT-TICKET-RESTAURANT:91215343,
TICKET_REGISTRATION_DATE=2012-11-02 10:08:00.000, TERMINAL_PN=M40,
TERMINAL_FIRST_CONTACT_DATE=2012-11-03 01:56:08.343, TERMINAL_ID=91215343,
RECORDER=EDENRED}
]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.CamelExecutionException: Exception occurred during
execution on the exchange: Exchange[Message: {TICKET_ID=00108560,
INGENICO_OPERATION_ID=1721710, TERMINAL_NUMBEROFCONTACTS=1, TICKET_STATUS=1,
TERMINAL_LAST_CONTACT_DATE=2012-11-03 01:57:26.360,
TERMINAL_TECHNOLOGY=TELIUM, TERMINAL_SN=0000107370643471,
TERMINAL_INGESTATE_ID=ICT-TICKET-RESTAURANT:91215343,
TICKET_REGISTRATION_DATE=2012-11-02 10:08:00.000, TERMINAL_PN=M40,
TERMINAL_FIRST_CONTACT_DATE=2012-11-03 01:56:08.343, TERMINAL_ID=91215343,
RECORDER=EDENRED}]
at
org.apache.camel.util.ObjectHelper.wrapCamelExecutionException(ObjectHelper.java:1635)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.impl.DefaultExchange.setException(DefaultExchange.java:308)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.component.jms.JmsProducer.process(JmsProducer.java:158)[209:org.apache.camel.camel-jms:2.15.1.redhat-620133]
at
org.apache.camel.processor.SendProcessor.process(SendProcessor.java:129)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.Pipeline.process(Pipeline.java:118)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.Pipeline.process(Pipeline.java:80)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.ChoiceProcessor.process(ChoiceProcessor.java:111)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.Pipeline.process(Pipeline.java:118)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.Pipeline.process(Pipeline.java:80)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:652)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:580)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.MulticastProcessor.process(MulticastProcessor.java:227)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.Splitter.process(Splitter.java:104)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.Pipeline.process(Pipeline.java:118)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.Pipeline.process(Pipeline.java:80)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:435)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:211)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:175)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:174)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:101)[198:org.apache.camel.camel-core:2.15.1.redhat-620133]
at
org.apache.camel.pollconsumer.quartz2.QuartzScheduledPollConsumerJob.execute(QuartzScheduledPollConsumerJob.java:59)[279:org.apache.camel.camel-quartz2:2.15.1.redhat-620133]
at
org.quartz.core.JobRunShell.run(JobRunShell.java:202)[278:org.quartz-scheduler.quartz:2.2.1]
at
org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:573)[278:org.quartz-scheduler.quartz:2.2.1]
Caused by: java.lang.OutOfMemoryError: Java heap space
Any idea?
How configure active mq component in order to manage the load in JBoss Fuse
6.2 context? Or Is there something wrong in my system routing?
Thanks in advance
Best Regards
Michele
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5779935.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Best Strategy to process a large number of rows in File
Posted by Ranx <br...@mediadriver.com>.
Michelle,
There are a number of ways you can do that and it will depend on what is
constraining your REST API. Is it limited on the number of concurrent
connections? Is it limited to the number of transactions/minute? There are
at least two components you'll want after the JMS queue. One will be a
thread pool limiting/defining the number of worker threads. The second will
be a Throttler EIP.
http://camel.apache.org/throttler.html
--
View this message in context: http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5779901.html
Sent from the Camel - Users mailing list archive at Nabble.com.