You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@storm.apache.org by Chen Wang <ch...@gmail.com> on 2014/02/17 20:49:22 UTC
Trident not working on streaming spout
Hi,
I might be missing something apparent. In my spout, next tuple method, I
have an infinite loop to read from our socket server:
@Override
public void nextTuple() {
try {
String line;
while ((line = socket.readLine()) != null) {
System.out.println("emitting new entries");
_collector.emit(new Values(line));
}
} catch (Exception e) {
// TODO
// log the exception with Storm
}
}
Then in my trident topology, i am doing something like:(psedu code)
topology.newStream(
myspout).
each(filter logic)
.groupBy( mykey)
.persistentAggregate(
When I run it in local mode, it seems that the trident will not catch any
tuple emitted from my spout. The topology will just spin on the spout
emitting tuples, while the rest of the topology never get executed. How
would I fix that?
Using normal bolt/topology will be able to receive the tuple.
Thanks,
Chen
Re: Trident not working on streaming spout
Posted by bijoy deb <bi...@gmail.com>.
Hi Chen,
You should remove the while() loop from nextTuple(),initialise the socket
opening code in your Spout's prepare() method(which gets called only once)
and in your nextTuple() you should have something like this:
String line;
if(line = socket.readLine()) != null) {
_collector.emit(new Values(line));
}
....
..
And,if you want to batch the tuples before emitting,you can maintain a
counter at class level,which gets incremented on every line!=null check in
nextTuple() and you emit only when that count is = BATCH_SIZE; e.g.
String line;
if(line = socket.readLine()) != null) {
.......Add the *line *to a List,etc.
lineCount++;
if(lineCount==BATCH_SIZE){
_collector.emit(new Values(line));
lineCount=0;
}
}
Hope this helps.
On Tue, Feb 18, 2014 at 1:19 AM, Chen Wang <ch...@gmail.com>wrote:
> Hi,
> I might be missing something apparent. In my spout, next tuple method, I
> have an infinite loop to read from our socket server:
>
> @Override
>
> public void nextTuple() {
>
> try {
>
> String line;
>
> while ((line = socket.readLine()) != null) {
>
> System.out.println("emitting new entries");
>
> _collector.emit(new Values(line));
>
> }
>
> } catch (Exception e) {
>
> // TODO
>
> // log the exception with Storm
>
> }
>
> }
>
>
> Then in my trident topology, i am doing something like:(psedu code)
>
> topology.newStream(
>
> myspout).
>
> each(filter logic)
>
> .groupBy( mykey)
>
> .persistentAggregate(
>
>
> When I run it in local mode, it seems that the trident will not catch any
> tuple emitted from my spout. The topology will just spin on the spout
> emitting tuples, while the rest of the topology never get executed. How
> would I fix that?
>
>
> Using normal bolt/topology will be able to receive the tuple.
>
> Thanks,
>
> Chen
>
Re: Trident not working on streaming spout
Posted by Nathan Marz <na...@nathanmarz.com>.
That's not true, it won't make a difference performance-wise whether you
emit one tuple or many in nextTuple. The batching happens automatically
after that.
On Mon, Feb 17, 2014 at 3:51 PM, Danijel Schiavuzzi <ds...@gmail.com>wrote:
> Also, Trident is batch-based, so in nextTuple() you should really emit
> batches of tuples, not single tuples, for maximum performance.
>
>
> On Tue, Feb 18, 2014 at 12:47 AM, Danijel Schiavuzzi <ds...@gmail.com>wrote:
>
>> Exactly, you must never block in nextTuple().
>>
>> Read your socket asynchronously (in a separate thread) and fill the data
>> into a queue. Then in nextTuple() emit data from that queue.
>>
>> On Tue, Feb 18, 2014 at 12:05 AM, Chen Wang <ch...@gmail.com>wrote:
>>
>>> If I remove the while loop in my spout, then everything works fine.(well
>>> except data loss in the spout..)
>>>
>>>
>>> On Mon, Feb 17, 2014 at 2:46 PM, Chen Wang <ch...@gmail.com>wrote:
>>>
>>>> Its just a BaseRichSpout.
>>>>
>>>>
>>>> On Mon, Feb 17, 2014 at 1:06 PM, Danijel Schiavuzzi <dschiavu@gmail.com
>>>> > wrote:
>>>>
>>>>> What interface does your spout implement? Is it IBatchSpout?
>>>>>
>>>>>
>>>>> On Mon, Feb 17, 2014 at 8:49 PM, Chen Wang <chen.apache.solr@gmail.com
>>>>> > wrote:
>>>>>
>>>>>> Hi,
>>>>>> I might be missing something apparent. In my spout, next tuple
>>>>>> method, I have an infinite loop to read from our socket server:
>>>>>>
>>>>>> @Override
>>>>>>
>>>>>> public void nextTuple() {
>>>>>>
>>>>>> try {
>>>>>>
>>>>>> String line;
>>>>>>
>>>>>> while ((line = socket.readLine()) != null) {
>>>>>>
>>>>>> System.out.println("emitting new entries");
>>>>>>
>>>>>> _collector.emit(new Values(line));
>>>>>>
>>>>>> }
>>>>>>
>>>>>> } catch (Exception e) {
>>>>>>
>>>>>> // TODO
>>>>>>
>>>>>> // log the exception with Storm
>>>>>>
>>>>>> }
>>>>>>
>>>>>> }
>>>>>>
>>>>>>
>>>>>> Then in my trident topology, i am doing something like:(psedu code)
>>>>>>
>>>>>> topology.newStream(
>>>>>>
>>>>>> myspout).
>>>>>>
>>>>>> each(filter logic)
>>>>>>
>>>>>> .groupBy( mykey)
>>>>>>
>>>>>> .persistentAggregate(
>>>>>>
>>>>>>
>>>>>> When I run it in local mode, it seems that the trident will not catch
>>>>>> any tuple emitted from my spout. The topology will just spin on the spout
>>>>>> emitting tuples, while the rest of the topology never get executed. How
>>>>>> would I fix that?
>>>>>>
>>>>>>
>>>>>> Using normal bolt/topology will be able to receive the tuple.
>>>>>>
>>>>>> Thanks,
>>>>>>
>>>>>> Chen
>>>>>>
>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> Danijel Schiavuzzi
>>>>>
>>>>
>>>>
>>>
>>
>>
>> --
>> Danijel Schiavuzzi
>>
>
>
>
> --
> Danijel Schiavuzzi
>
--
Twitter: @nathanmarz
http://nathanmarz.com
Re: Trident not working on streaming spout
Posted by Danijel Schiavuzzi <ds...@gmail.com>.
Also, Trident is batch-based, so in nextTuple() you should really emit
batches of tuples, not single tuples, for maximum performance.
On Tue, Feb 18, 2014 at 12:47 AM, Danijel Schiavuzzi <ds...@gmail.com>wrote:
> Exactly, you must never block in nextTuple().
>
> Read your socket asynchronously (in a separate thread) and fill the data
> into a queue. Then in nextTuple() emit data from that queue.
>
> On Tue, Feb 18, 2014 at 12:05 AM, Chen Wang <ch...@gmail.com>wrote:
>
>> If I remove the while loop in my spout, then everything works fine.(well
>> except data loss in the spout..)
>>
>>
>> On Mon, Feb 17, 2014 at 2:46 PM, Chen Wang <ch...@gmail.com>wrote:
>>
>>> Its just a BaseRichSpout.
>>>
>>>
>>> On Mon, Feb 17, 2014 at 1:06 PM, Danijel Schiavuzzi <ds...@gmail.com>wrote:
>>>
>>>> What interface does your spout implement? Is it IBatchSpout?
>>>>
>>>>
>>>> On Mon, Feb 17, 2014 at 8:49 PM, Chen Wang <ch...@gmail.com>wrote:
>>>>
>>>>> Hi,
>>>>> I might be missing something apparent. In my spout, next tuple method,
>>>>> I have an infinite loop to read from our socket server:
>>>>>
>>>>> @Override
>>>>>
>>>>> public void nextTuple() {
>>>>>
>>>>> try {
>>>>>
>>>>> String line;
>>>>>
>>>>> while ((line = socket.readLine()) != null) {
>>>>>
>>>>> System.out.println("emitting new entries");
>>>>>
>>>>> _collector.emit(new Values(line));
>>>>>
>>>>> }
>>>>>
>>>>> } catch (Exception e) {
>>>>>
>>>>> // TODO
>>>>>
>>>>> // log the exception with Storm
>>>>>
>>>>> }
>>>>>
>>>>> }
>>>>>
>>>>>
>>>>> Then in my trident topology, i am doing something like:(psedu code)
>>>>>
>>>>> topology.newStream(
>>>>>
>>>>> myspout).
>>>>>
>>>>> each(filter logic)
>>>>>
>>>>> .groupBy( mykey)
>>>>>
>>>>> .persistentAggregate(
>>>>>
>>>>>
>>>>> When I run it in local mode, it seems that the trident will not catch
>>>>> any tuple emitted from my spout. The topology will just spin on the spout
>>>>> emitting tuples, while the rest of the topology never get executed. How
>>>>> would I fix that?
>>>>>
>>>>>
>>>>> Using normal bolt/topology will be able to receive the tuple.
>>>>>
>>>>> Thanks,
>>>>>
>>>>> Chen
>>>>>
>>>>
>>>>
>>>>
>>>> --
>>>> Danijel Schiavuzzi
>>>>
>>>
>>>
>>
>
>
> --
> Danijel Schiavuzzi
>
--
Danijel Schiavuzzi
Re: Trident not working on streaming spout
Posted by Danijel Schiavuzzi <ds...@gmail.com>.
Exactly, you must never block in nextTuple().
Read your socket asynchronously (in a separate thread) and fill the data
into a queue. Then in nextTuple() emit data from that queue.
On Tue, Feb 18, 2014 at 12:05 AM, Chen Wang <ch...@gmail.com>wrote:
> If I remove the while loop in my spout, then everything works fine.(well
> except data loss in the spout..)
>
>
> On Mon, Feb 17, 2014 at 2:46 PM, Chen Wang <ch...@gmail.com>wrote:
>
>> Its just a BaseRichSpout.
>>
>>
>> On Mon, Feb 17, 2014 at 1:06 PM, Danijel Schiavuzzi <ds...@gmail.com>wrote:
>>
>>> What interface does your spout implement? Is it IBatchSpout?
>>>
>>>
>>> On Mon, Feb 17, 2014 at 8:49 PM, Chen Wang <ch...@gmail.com>wrote:
>>>
>>>> Hi,
>>>> I might be missing something apparent. In my spout, next tuple method,
>>>> I have an infinite loop to read from our socket server:
>>>>
>>>> @Override
>>>>
>>>> public void nextTuple() {
>>>>
>>>> try {
>>>>
>>>> String line;
>>>>
>>>> while ((line = socket.readLine()) != null) {
>>>>
>>>> System.out.println("emitting new entries");
>>>>
>>>> _collector.emit(new Values(line));
>>>>
>>>> }
>>>>
>>>> } catch (Exception e) {
>>>>
>>>> // TODO
>>>>
>>>> // log the exception with Storm
>>>>
>>>> }
>>>>
>>>> }
>>>>
>>>>
>>>> Then in my trident topology, i am doing something like:(psedu code)
>>>>
>>>> topology.newStream(
>>>>
>>>> myspout).
>>>>
>>>> each(filter logic)
>>>>
>>>> .groupBy( mykey)
>>>>
>>>> .persistentAggregate(
>>>>
>>>>
>>>> When I run it in local mode, it seems that the trident will not catch
>>>> any tuple emitted from my spout. The topology will just spin on the spout
>>>> emitting tuples, while the rest of the topology never get executed. How
>>>> would I fix that?
>>>>
>>>>
>>>> Using normal bolt/topology will be able to receive the tuple.
>>>>
>>>> Thanks,
>>>>
>>>> Chen
>>>>
>>>
>>>
>>>
>>> --
>>> Danijel Schiavuzzi
>>>
>>
>>
>
--
Danijel Schiavuzzi
Re: Trident not working on streaming spout
Posted by Chen Wang <ch...@gmail.com>.
If I remove the while loop in my spout, then everything works fine.(well
except data loss in the spout..)
On Mon, Feb 17, 2014 at 2:46 PM, Chen Wang <ch...@gmail.com>wrote:
> Its just a BaseRichSpout.
>
>
> On Mon, Feb 17, 2014 at 1:06 PM, Danijel Schiavuzzi <ds...@gmail.com>wrote:
>
>> What interface does your spout implement? Is it IBatchSpout?
>>
>>
>> On Mon, Feb 17, 2014 at 8:49 PM, Chen Wang <ch...@gmail.com>wrote:
>>
>>> Hi,
>>> I might be missing something apparent. In my spout, next tuple method, I
>>> have an infinite loop to read from our socket server:
>>>
>>> @Override
>>>
>>> public void nextTuple() {
>>>
>>> try {
>>>
>>> String line;
>>>
>>> while ((line = socket.readLine()) != null) {
>>>
>>> System.out.println("emitting new entries");
>>>
>>> _collector.emit(new Values(line));
>>>
>>> }
>>>
>>> } catch (Exception e) {
>>>
>>> // TODO
>>>
>>> // log the exception with Storm
>>>
>>> }
>>>
>>> }
>>>
>>>
>>> Then in my trident topology, i am doing something like:(psedu code)
>>>
>>> topology.newStream(
>>>
>>> myspout).
>>>
>>> each(filter logic)
>>>
>>> .groupBy( mykey)
>>>
>>> .persistentAggregate(
>>>
>>>
>>> When I run it in local mode, it seems that the trident will not catch
>>> any tuple emitted from my spout. The topology will just spin on the spout
>>> emitting tuples, while the rest of the topology never get executed. How
>>> would I fix that?
>>>
>>>
>>> Using normal bolt/topology will be able to receive the tuple.
>>>
>>> Thanks,
>>>
>>> Chen
>>>
>>
>>
>>
>> --
>> Danijel Schiavuzzi
>>
>
>
Re: Trident not working on streaming spout
Posted by Chen Wang <ch...@gmail.com>.
Its just a BaseRichSpout.
On Mon, Feb 17, 2014 at 1:06 PM, Danijel Schiavuzzi <ds...@gmail.com>wrote:
> What interface does your spout implement? Is it IBatchSpout?
>
>
> On Mon, Feb 17, 2014 at 8:49 PM, Chen Wang <ch...@gmail.com>wrote:
>
>> Hi,
>> I might be missing something apparent. In my spout, next tuple method, I
>> have an infinite loop to read from our socket server:
>>
>> @Override
>>
>> public void nextTuple() {
>>
>> try {
>>
>> String line;
>>
>> while ((line = socket.readLine()) != null) {
>>
>> System.out.println("emitting new entries");
>>
>> _collector.emit(new Values(line));
>>
>> }
>>
>> } catch (Exception e) {
>>
>> // TODO
>>
>> // log the exception with Storm
>>
>> }
>>
>> }
>>
>>
>> Then in my trident topology, i am doing something like:(psedu code)
>>
>> topology.newStream(
>>
>> myspout).
>>
>> each(filter logic)
>>
>> .groupBy( mykey)
>>
>> .persistentAggregate(
>>
>>
>> When I run it in local mode, it seems that the trident will not catch any
>> tuple emitted from my spout. The topology will just spin on the spout
>> emitting tuples, while the rest of the topology never get executed. How
>> would I fix that?
>>
>>
>> Using normal bolt/topology will be able to receive the tuple.
>>
>> Thanks,
>>
>> Chen
>>
>
>
>
> --
> Danijel Schiavuzzi
>
Re: Trident not working on streaming spout
Posted by Danijel Schiavuzzi <ds...@gmail.com>.
What interface does your spout implement? Is it IBatchSpout?
On Mon, Feb 17, 2014 at 8:49 PM, Chen Wang <ch...@gmail.com>wrote:
> Hi,
> I might be missing something apparent. In my spout, next tuple method, I
> have an infinite loop to read from our socket server:
>
> @Override
>
> public void nextTuple() {
>
> try {
>
> String line;
>
> while ((line = socket.readLine()) != null) {
>
> System.out.println("emitting new entries");
>
> _collector.emit(new Values(line));
>
> }
>
> } catch (Exception e) {
>
> // TODO
>
> // log the exception with Storm
>
> }
>
> }
>
>
> Then in my trident topology, i am doing something like:(psedu code)
>
> topology.newStream(
>
> myspout).
>
> each(filter logic)
>
> .groupBy( mykey)
>
> .persistentAggregate(
>
>
> When I run it in local mode, it seems that the trident will not catch any
> tuple emitted from my spout. The topology will just spin on the spout
> emitting tuples, while the rest of the topology never get executed. How
> would I fix that?
>
>
> Using normal bolt/topology will be able to receive the tuple.
>
> Thanks,
>
> Chen
>
--
Danijel Schiavuzzi