You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Vinti Maheshwari <vi...@gmail.com> on 2016/03/07 23:52:57 UTC
Spark Streaming, very slow processing and increasing scheduling delay
of kafka input stream
Hi,
My spark-streaming program seems very slow. I am using Ambari for cluster
setup and i am using Kafka for data input.
I tried to use batch size 2 secs and check pointing duration 10 secs. But
as i was seeing scheduling delay was keep increasing so i tried increasing
the batch size to 5 and then 10 secs. But it seems noting changed in
respect of performance.
*My program is doing two tasks:*
1) Data aggregation
2) Data insertion into Hbase
Action which took maximum time, when i called foreachRDD on Dstream object
(state).
*state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
*Program sample input coming from kafka:*
test_id, file1, 1,1,1,1,1
*Code snippets:*
val parsedStream = inputStream
.map(line => {
val splitLines = line.split(",")
(splitLines(1), splitLines.slice(2, splitLines.length).map((_.trim.toLong)))
})
val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
(current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
prev.map(_ +: current).orElse(Some(current))
.flatMap(as => Try(as.map(BDV(_)).reduce(_ + _).toArray).toOption)
})
*state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
object Blaher {
def blah(tup: (String, Array[Long])) {
val hConf = HBaseConfiguration.create()
------
val hTable = new HTable(hConf, tableName)
val thePut = new Put(Bytes.toBytes("file_data"))
thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1),
Bytes.toBytes(tup._2.toList.toString))
new ImmutableBytesWritable(Bytes.toBytes("file_data"))
hTable.put(thePut)
}
}
*My Cluster Specifications:*
16 executors ( 1 core each and 2g memory)
I have attached some screenshots of running execution.
Anyone has idea what changes should i do to speedup the processing?
Thanks & Regards,
Vinti
Re: Spark Streaming, very slow processing and increasing scheduling
delay of kafka input stream
Posted by Vinti Maheshwari <vi...@gmail.com>.
Hi Todd,
Thanks for replying on my email. Thanks for sharing link i will read that
as well.
The info you ask is as follows:
It was running on YARN. Overall memory is 256 GB ( 2 nodes, 128 GB each)
I have total 16 cores (8 cores on each node) So i was running 16 executors
with one core each. I was not sure how much memory, i should allocate to
each executor. I gave 6 GB to each executor while running these tests.
One more thing, i wanted to mention is when i was running with small data
sets it was finished very quickly but when i am sending more data then the
performance start decreasing.
I enabled backpressure on for kafka input. So as scheduling delay was
increasing, then backpressure automatically reduced the number of events
from 4000 records/ sec to 160 records/sec.
I have attached two more screenshots, which might explain this behaviour
further.
Thanks & Regards,
Vinti
On Thu, Mar 10, 2016 at 6:03 AM, Todd Nist <ts...@gmail.com> wrote:
> Hi Vinti,
>
> All of your tasks are failing based on the screen shots provided.
>
> I think a few more details would be helpful. Is this YARN or a Standalone
> cluster? How much overall memory is on your cluster? On each machine
> where workers and executors are running? Are you using the Direct
> (KafkaUtils.createDirectStream) or Receiver (KafkaUtils.createStream)?
>
> You may find this discussion of value on SO:
>
> http://stackoverflow.com/questions/28901123/org-apache-spark-shuffle-metadatafetchfailedexception-missing-an-output-locatio
>
> -Todd
>
> On Mon, Mar 7, 2016 at 5:52 PM, Vinti Maheshwari <vi...@gmail.com>
> wrote:
>
>> Hi,
>>
>> My spark-streaming program seems very slow. I am using Ambari for cluster
>> setup and i am using Kafka for data input.
>> I tried to use batch size 2 secs and check pointing duration 10 secs. But
>> as i was seeing scheduling delay was keep increasing so i tried increasing
>> the batch size to 5 and then 10 secs. But it seems noting changed in
>> respect of performance.
>>
>> *My program is doing two tasks:*
>>
>> 1) Data aggregation
>>
>> 2) Data insertion into Hbase
>>
>> Action which took maximum time, when i called foreachRDD on Dstream
>> object (state).
>>
>> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>>
>>
>>
>>
>> *Program sample input coming from kafka:*
>> test_id, file1, 1,1,1,1,1
>>
>> *Code snippets:*
>>
>> val parsedStream = inputStream
>> .map(line => {
>> val splitLines = line.split(",")
>> (splitLines(1), splitLines.slice(2, splitLines.length).map((_.trim.toLong)))
>> })
>> val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
>> (current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
>> prev.map(_ +: current).orElse(Some(current))
>> .flatMap(as => Try(as.map(BDV(_)).reduce(_ + _).toArray).toOption)
>> })
>> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>>
>>
>>
>> object Blaher {
>> def blah(tup: (String, Array[Long])) {
>> val hConf = HBaseConfiguration.create()
>> ------
>> val hTable = new HTable(hConf, tableName)
>> val thePut = new Put(Bytes.toBytes("file_data"))
>> thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1), Bytes.toBytes(tup._2.toList.toString))
>> new ImmutableBytesWritable(Bytes.toBytes("file_data"))
>>
>> hTable.put(thePut)
>> }
>> }
>>
>>
>> *My Cluster Specifications:*
>> 16 executors ( 1 core each and 2g memory)
>>
>> I have attached some screenshots of running execution.
>>
>> Anyone has idea what changes should i do to speedup the processing?
>>
>> Thanks & Regards,
>>
>> Vinti
>>
>>
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>> For additional commands, e-mail: user-help@spark.apache.org
>>
>
>
Re: Spark Streaming, very slow processing and increasing scheduling
delay of kafka input stream
Posted by Andy Davidson <An...@SantaCruzIntegration.com>.
Hi Vinti
You have stages with durations of 10 sec .
Your screen shots look different from from mine. I use spark-1.6. My urls
ends with ompute.amazonaws.com:4040/stages/
Notice the completed stages section shows stage duration and discretion show
java file and line number
The +details link is very helpful
From: Vinti Maheshwari <vi...@gmail.com>
Date: Thursday, March 10, 2016 at 2:55 PM
To: Andrew Davidson <An...@SantaCruzIntegration.com>
Cc: Todd Nist <ts...@gmail.com>, "user @spark" <us...@spark.apache.org>
Subject: Re: Spark Streaming, very slow processing and increasing
scheduling delay of kafka input stream
> Hi Andy, thanks much for your inputs.
>
> Adding stage tab, screenshots also.
>
> Regards,
> Vinti
>
>
> On Thu, Mar 10, 2016 at 2:40 PM, Andy Davidson <An...@santacruzintegration.com>
> wrote:
>> In my experience I would try the following
>>
>>
>> I use the standalone cluster manager. Each app gets it own performance web
>> page . The streaming tab is really helpful. If processing time is greater
>> than then your mini batch length you are going to run into performance
>> problems
>>
>> Use the ³stages² tab to figure out where the bottleneck is
>>
>> In our experience we where working with large # of empty partitions.
>> Reparation solved our s3 output problem
>>
>> Your milage may vary
>>
>> andy
>>
>> From: Todd Nist <ts...@gmail.com>
>> Date: Thursday, March 10, 2016 at 6:03 AM
>> To: Vinti Maheshwari <vi...@gmail.com>
>> Cc: "user @spark" <us...@spark.apache.org>
>> Subject: Re: Spark Streaming, very slow processing and increasing scheduling
>> delay of kafka input stream
>>
>>> Hi Vinti,
>>>
>>> All of your tasks are failing based on the screen shots provided.
>>>
>>> I think a few more details would be helpful. Is this YARN or a Standalone
>>> cluster? How much overall memory is on your cluster? On each machine where
>>> workers and executors are running? Are you using the Direct
>>> (KafkaUtils.createDirectStream) or Receiver (KafkaUtils.createStream)?
>>>
>>> You may find this discussion of value on SO:
>>> http://stackoverflow.com/questions/28901123/org-apache-spark-shuffle-metadat
>>> afetchfailedexception-missing-an-output-locatio
>>>
>>> -Todd
>>>
>>> On Mon, Mar 7, 2016 at 5:52 PM, Vinti Maheshwari <vi...@gmail.com>
>>> wrote:
>>>> Hi,
>>>>
>>>> My spark-streaming program seems very slow. I am using Ambari for cluster
>>>> setup and i am using Kafka for data input.
>>>> I tried to use batch size 2 secs and check pointing duration 10 secs. But
>>>> as i was seeing scheduling delay was keep increasing so i tried increasing
>>>> the batch size to 5 and then 10 secs. But it seems noting changed in
>>>> respect of performance.
>>>>
>>>> My program is doing two tasks:
>>>>
>>>> 1) Data aggregation
>>>>
>>>> 2) Data insertion into Hbase
>>>>
>>>> Action which took maximum time, when i called foreachRDD on Dstream object
>>>> (state).
>>>> state.foreachRDD(rdd => rdd.foreach(Blaher.blah))
>>>>
>>>>
>>>> Program sample input coming from kafka:
>>>> test_id, file1, 1,1,1,1,1
>>>>
>>>> Code snippets:
>>>> val parsedStream = inputStream
>>>> .map(line => {
>>>> val splitLines = line.split(",")
>>>> (splitLines(1), splitLines.slice(2,
>>>> splitLines.length).map((_.trim.toLong)))
>>>> })
>>>> val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
>>>> (current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
>>>> prev.map(_ +: current).orElse(Some(current))
>>>> .flatMap(as => Try(as.map(BDV(_)).reduce(_ +
>>>> _).toArray).toOption)
>>>> })
>>>> state.foreachRDD(rdd => rdd.foreach(Blaher.blah))
>>>>
>>>>
>>>>
>>>> object Blaher {
>>>> def blah(tup: (String, Array[Long])) {
>>>> val hConf = HBaseConfiguration.create()
>>>> ------
>>>> val hTable = new HTable(hConf, tableName)
>>>> val thePut = new Put(Bytes.toBytes("file_data"))
>>>> thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1),
>>>> Bytes.toBytes(tup._2.toList.toString))
>>>> new ImmutableBytesWritable(Bytes.toBytes("file_data"))
>>>>
>>>> hTable.put(thePut)
>>>> }
>>>> }
>>>>
>>>>
>>>> My Cluster Specifications:
>>>> 16 executors ( 1 core each and 2g memory)
>>>> I have attached some screenshots of running execution.
>>>>
>>>> Anyone has idea what changes should i do to speedup the processing?
>>>>
>>>> Thanks & Regards,
>>>> Vinti
>>>>
>>>>
>>>> ---------------------------------------------------------------------
>>>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>>>> For additional commands, e-mail: user-help@spark.apache.org
>>>
>
Re: Spark Streaming, very slow processing and increasing scheduling
delay of kafka input stream
Posted by Vinti Maheshwari <vi...@gmail.com>.
Hi Andy, thanks much for your inputs.
Adding stage tab, screenshots also.
Regards,
Vinti
On Thu, Mar 10, 2016 at 2:40 PM, Andy Davidson <
Andy@santacruzintegration.com> wrote:
> In my experience I would try the following
>
>
> I use the standalone cluster manager. Each app gets it own performance web
> page . The streaming tab is really helpful. If processing time is greater
> than then your mini batch length you are going to run into performance
> problems
>
> Use the “stages” tab to figure out where the bottleneck is
>
> In our experience we where working with large # of empty partitions.
> Reparation solved our s3 output problem
>
> Your milage may vary
>
> andy
>
> From: Todd Nist <ts...@gmail.com>
> Date: Thursday, March 10, 2016 at 6:03 AM
> To: Vinti Maheshwari <vi...@gmail.com>
> Cc: "user @spark" <us...@spark.apache.org>
> Subject: Re: Spark Streaming, very slow processing and increasing
> scheduling delay of kafka input stream
>
> Hi Vinti,
>
> All of your tasks are failing based on the screen shots provided.
>
> I think a few more details would be helpful. Is this YARN or a Standalone
> cluster? How much overall memory is on your cluster? On each machine
> where workers and executors are running? Are you using the Direct
> (KafkaUtils.createDirectStream) or Receiver (KafkaUtils.createStream)?
>
> You may find this discussion of value on SO:
>
> http://stackoverflow.com/questions/28901123/org-apache-spark-shuffle-metadatafetchfailedexception-missing-an-output-locatio
>
> -Todd
>
> On Mon, Mar 7, 2016 at 5:52 PM, Vinti Maheshwari <vi...@gmail.com>
> wrote:
>
>> Hi,
>>
>> My spark-streaming program seems very slow. I am using Ambari for cluster
>> setup and i am using Kafka for data input.
>> I tried to use batch size 2 secs and check pointing duration 10 secs. But
>> as i was seeing scheduling delay was keep increasing so i tried increasing
>> the batch size to 5 and then 10 secs. But it seems noting changed in
>> respect of performance.
>>
>> *My program is doing two tasks:*
>>
>> 1) Data aggregation
>>
>> 2) Data insertion into Hbase
>>
>> Action which took maximum time, when i called foreachRDD on Dstream
>> object (state).
>>
>> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>>
>>
>>
>>
>> *Program sample input coming from kafka:*
>> test_id, file1, 1,1,1,1,1
>>
>> *Code snippets:*
>>
>> val parsedStream = inputStream
>> .map(line => {
>> val splitLines = line.split(",")
>> (splitLines(1), splitLines.slice(2, splitLines.length).map((_.trim.toLong)))
>> })
>> val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
>> (current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
>> prev.map(_ +: current).orElse(Some(current))
>> .flatMap(as => Try(as.map(BDV(_)).reduce(_ + _).toArray).toOption)
>> })
>> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>>
>>
>>
>> object Blaher {
>> def blah(tup: (String, Array[Long])) {
>> val hConf = HBaseConfiguration.create()
>> ------
>> val hTable = new HTable(hConf, tableName)
>> val thePut = new Put(Bytes.toBytes("file_data"))
>> thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1), Bytes.toBytes(tup._2.toList.toString))
>> new ImmutableBytesWritable(Bytes.toBytes("file_data"))
>>
>> hTable.put(thePut)
>> }
>> }
>>
>>
>> *My Cluster Specifications:*
>> 16 executors ( 1 core each and 2g memory)
>>
>> I have attached some screenshots of running execution.
>>
>> Anyone has idea what changes should i do to speedup the processing?
>>
>> Thanks & Regards,
>>
>> Vinti
>>
>>
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>> For additional commands, e-mail: user-help@spark.apache.org
>>
>
>
Re: Spark Streaming, very slow processing and increasing scheduling
delay of kafka input stream
Posted by Andy Davidson <An...@SantaCruzIntegration.com>.
In my experience I would try the following
I use the standalone cluster manager. Each app gets it own performance web
page . The streaming tab is really helpful. If processing time is greater
than then your mini batch length you are going to run into performance
problems
Use the ³stages² tab to figure out where the bottleneck is
In our experience we where working with large # of empty partitions.
Reparation solved our s3 output problem
Your milage may vary
andy
From: Todd Nist <ts...@gmail.com>
Date: Thursday, March 10, 2016 at 6:03 AM
To: Vinti Maheshwari <vi...@gmail.com>
Cc: "user @spark" <us...@spark.apache.org>
Subject: Re: Spark Streaming, very slow processing and increasing
scheduling delay of kafka input stream
> Hi Vinti,
>
> All of your tasks are failing based on the screen shots provided.
>
> I think a few more details would be helpful. Is this YARN or a Standalone
> cluster? How much overall memory is on your cluster? On each machine where
> workers and executors are running? Are you using the Direct
> (KafkaUtils.createDirectStream) or Receiver (KafkaUtils.createStream)?
>
> You may find this discussion of value on SO:
> http://stackoverflow.com/questions/28901123/org-apache-spark-shuffle-metadataf
> etchfailedexception-missing-an-output-locatio
>
> -Todd
>
> On Mon, Mar 7, 2016 at 5:52 PM, Vinti Maheshwari <vi...@gmail.com> wrote:
>> Hi,
>>
>> My spark-streaming program seems very slow. I am using Ambari for cluster
>> setup and i am using Kafka for data input.
>> I tried to use batch size 2 secs and check pointing duration 10 secs. But as
>> i was seeing scheduling delay was keep increasing so i tried increasing the
>> batch size to 5 and then 10 secs. But it seems noting changed in respect of
>> performance.
>>
>> My program is doing two tasks:
>>
>> 1) Data aggregation
>>
>> 2) Data insertion into Hbase
>>
>> Action which took maximum time, when i called foreachRDD on Dstream object
>> (state).
>> state.foreachRDD(rdd => rdd.foreach(Blaher.blah))
>>
>>
>> Program sample input coming from kafka:
>> test_id, file1, 1,1,1,1,1
>>
>> Code snippets:
>> val parsedStream = inputStream
>> .map(line => {
>> val splitLines = line.split(",")
>> (splitLines(1), splitLines.slice(2,
>> splitLines.length).map((_.trim.toLong)))
>> })
>> val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
>> (current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
>> prev.map(_ +: current).orElse(Some(current))
>> .flatMap(as => Try(as.map(BDV(_)).reduce(_ +
>> _).toArray).toOption)
>> })
>> state.foreachRDD(rdd => rdd.foreach(Blaher.blah))
>>
>>
>>
>> object Blaher {
>> def blah(tup: (String, Array[Long])) {
>> val hConf = HBaseConfiguration.create()
>> ------
>> val hTable = new HTable(hConf, tableName)
>> val thePut = new Put(Bytes.toBytes("file_data"))
>> thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1),
>> Bytes.toBytes(tup._2.toList.toString))
>> new ImmutableBytesWritable(Bytes.toBytes("file_data"))
>>
>> hTable.put(thePut)
>> }
>> }
>>
>>
>> My Cluster Specifications:
>> 16 executors ( 1 core each and 2g memory)
>> I have attached some screenshots of running execution.
>>
>> Anyone has idea what changes should i do to speedup the processing?
>>
>> Thanks & Regards,
>> Vinti
>>
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>> For additional commands, e-mail: user-help@spark.apache.org
>
Re: Spark Streaming, very slow processing and increasing scheduling
delay of kafka input stream
Posted by Todd Nist <ts...@gmail.com>.
Hi Vinti,
All of your tasks are failing based on the screen shots provided.
I think a few more details would be helpful. Is this YARN or a Standalone
cluster? How much overall memory is on your cluster? On each machine
where workers and executors are running? Are you using the Direct
(KafkaUtils.createDirectStream) or Receiver (KafkaUtils.createStream)?
You may find this discussion of value on SO:
http://stackoverflow.com/questions/28901123/org-apache-spark-shuffle-metadatafetchfailedexception-missing-an-output-locatio
-Todd
On Mon, Mar 7, 2016 at 5:52 PM, Vinti Maheshwari <vi...@gmail.com>
wrote:
> Hi,
>
> My spark-streaming program seems very slow. I am using Ambari for cluster
> setup and i am using Kafka for data input.
> I tried to use batch size 2 secs and check pointing duration 10 secs. But
> as i was seeing scheduling delay was keep increasing so i tried increasing
> the batch size to 5 and then 10 secs. But it seems noting changed in
> respect of performance.
>
> *My program is doing two tasks:*
>
> 1) Data aggregation
>
> 2) Data insertion into Hbase
>
> Action which took maximum time, when i called foreachRDD on Dstream object
> (state).
>
> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>
>
>
>
> *Program sample input coming from kafka:*
> test_id, file1, 1,1,1,1,1
>
> *Code snippets:*
>
> val parsedStream = inputStream
> .map(line => {
> val splitLines = line.split(",")
> (splitLines(1), splitLines.slice(2, splitLines.length).map((_.trim.toLong)))
> })
> val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
> (current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
> prev.map(_ +: current).orElse(Some(current))
> .flatMap(as => Try(as.map(BDV(_)).reduce(_ + _).toArray).toOption)
> })
> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>
>
>
> object Blaher {
> def blah(tup: (String, Array[Long])) {
> val hConf = HBaseConfiguration.create()
> ------
> val hTable = new HTable(hConf, tableName)
> val thePut = new Put(Bytes.toBytes("file_data"))
> thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1), Bytes.toBytes(tup._2.toList.toString))
> new ImmutableBytesWritable(Bytes.toBytes("file_data"))
>
> hTable.put(thePut)
> }
> }
>
>
> *My Cluster Specifications:*
> 16 executors ( 1 core each and 2g memory)
>
> I have attached some screenshots of running execution.
>
> Anyone has idea what changes should i do to speedup the processing?
>
> Thanks & Regards,
>
> Vinti
>
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
Re: Spark Streaming, very slow processing and increasing scheduling
delay of kafka input stream
Posted by Andy Davidson <An...@SantaCruzIntegration.com>.
Hi Vinti
I use the stand alone cluster. I the mgmt console provides a link to an app
UI. It has all sorts of performance info. There should be a tab ¹stages¹.
You can use it to find bottlenecks
Note the links in the mgmt console do not seem to work. The app UI runs on
the same machine as the drivers. The first app will run on port 4040, the
second on 4041,
Good luck
andy
From: Vinti Maheshwari <vi...@gmail.com>
Date: Monday, March 7, 2016 at 3:01 PM
To: "user @spark" <us...@spark.apache.org>
Subject: Re: Spark Streaming, very slow processing and increasing
scheduling delay of kafka input stream
> Adding one more screenshots of running spark jobs, it's showing lots of
> skipped tasks. I am not sure why it's showing so many skipped tasks.
>
> Regards
> ~Vinti
>
> On Mon, Mar 7, 2016 at 2:52 PM, Vinti Maheshwari <vi...@gmail.com> wrote:
>> Hi,
>>
>> My spark-streaming program seems very slow. I am using Ambari for cluster
>> setup and i am using Kafka for data input.
>> I tried to use batch size 2 secs and check pointing duration 10 secs. But as
>> i was seeing scheduling delay was keep increasing so i tried increasing the
>> batch size to 5 and then 10 secs. But it seems noting changed in respect of
>> performance.
>>
>> My program is doing two tasks:
>>
>> 1) Data aggregation
>>
>> 2) Data insertion into Hbase
>>
>> Action which took maximum time, when i called foreachRDD on Dstream object
>> (state).
>> state.foreachRDD(rdd => rdd.foreach(Blaher.blah))
>>
>>
>> Program sample input coming from kafka:
>> test_id, file1, 1,1,1,1,1
>>
>> Code snippets:
>> val parsedStream = inputStream
>> .map(line => {
>> val splitLines = line.split(",")
>> (splitLines(1), splitLines.slice(2,
>> splitLines.length).map((_.trim.toLong)))
>> })
>> val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
>> (current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
>> prev.map(_ +: current).orElse(Some(current))
>> .flatMap(as => Try(as.map(BDV(_)).reduce(_ +
>> _).toArray).toOption)
>> })
>> state.foreachRDD(rdd => rdd.foreach(Blaher.blah))
>>
>>
>>
>> object Blaher {
>> def blah(tup: (String, Array[Long])) {
>> val hConf = HBaseConfiguration.create()
>> ------
>> val hTable = new HTable(hConf, tableName)
>> val thePut = new Put(Bytes.toBytes("file_data"))
>> thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1),
>> Bytes.toBytes(tup._2.toList.toString))
>> new ImmutableBytesWritable(Bytes.toBytes("file_data"))
>>
>> hTable.put(thePut)
>> }
>> }
>>
>>
>> My Cluster Specifications:
>> 16 executors ( 1 core each and 2g memory)
>> I have attached some screenshots of running execution.
>>
>> Anyone has idea what changes should i do to speedup the processing?
>>
>> Thanks & Regards,
>> Vinti
>
> --------------------------------------------------------------------- To
> unsubscribe, e-mail: user-unsubscribe@spark.apache.org For additional
> commands, e-mail: user-help@spark.apache.org
Re: Spark Streaming, very slow processing and increasing scheduling
delay of kafka input stream
Posted by Vinti Maheshwari <vi...@gmail.com>.
Adding one more screenshots of running spark jobs, it's showing lots of
skipped tasks. I am not sure why it's showing so many skipped tasks.
Regards
~Vinti
On Mon, Mar 7, 2016 at 2:52 PM, Vinti Maheshwari <vi...@gmail.com>
wrote:
> Hi,
>
> My spark-streaming program seems very slow. I am using Ambari for cluster
> setup and i am using Kafka for data input.
> I tried to use batch size 2 secs and check pointing duration 10 secs. But
> as i was seeing scheduling delay was keep increasing so i tried increasing
> the batch size to 5 and then 10 secs. But it seems noting changed in
> respect of performance.
>
> *My program is doing two tasks:*
>
> 1) Data aggregation
>
> 2) Data insertion into Hbase
>
> Action which took maximum time, when i called foreachRDD on Dstream object
> (state).
>
> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>
>
>
>
> *Program sample input coming from kafka:*
> test_id, file1, 1,1,1,1,1
>
> *Code snippets:*
>
> val parsedStream = inputStream
> .map(line => {
> val splitLines = line.split(",")
> (splitLines(1), splitLines.slice(2, splitLines.length).map((_.trim.toLong)))
> })
> val state: DStream[(String, Array[Long])] = parsedStream.updateStateByKey(
> (current: Seq[Array[Long]], prev: Option[Array[Long]]) => {
> prev.map(_ +: current).orElse(Some(current))
> .flatMap(as => Try(as.map(BDV(_)).reduce(_ + _).toArray).toOption)
> })
> *state.foreachRDD(rdd => rdd.foreach(Blaher.blah))*
>
>
>
> object Blaher {
> def blah(tup: (String, Array[Long])) {
> val hConf = HBaseConfiguration.create()
> ------
> val hTable = new HTable(hConf, tableName)
> val thePut = new Put(Bytes.toBytes("file_data"))
> thePut.add(Bytes.toBytes("file_counts"), Bytes.toBytes(tup._1), Bytes.toBytes(tup._2.toList.toString))
> new ImmutableBytesWritable(Bytes.toBytes("file_data"))
>
> hTable.put(thePut)
> }
> }
>
>
> *My Cluster Specifications:*
> 16 executors ( 1 core each and 2g memory)
>
> I have attached some screenshots of running execution.
>
> Anyone has idea what changes should i do to speedup the processing?
>
> Thanks & Regards,
>
> Vinti
>
>