[jira] [Comment Edited] (SPARK-4960) Interceptor pattern in receivers

["Saisai Shao (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-4960?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14258994#comment-14258994 ] 

Saisai Shao edited comment on SPARK-4960 at 12/26/14 8:47 AM:
--------------------------------------------------------------

What confused me a lot is the type inference, as the interceptor added, say M => Iterable\[T\], so the type of DStream\[M\] need to be changed into DStream\[T\], while actually the type of input DStream is M, like:

{code}
DStream[T] => InputDStream[T] => ReceiverInputDStream[T] => SocketInputDStream[M]
{code}

Assuming I add the interceptor before the ReceiverInputDStream.

 I'm not sure there's way to do it? A possible way is changing the interceptor to Any => Iterable\[T\], but how to reflect the received data type is also a problem, still function like bytesToObjects: InputStream => Iterator\[T\] is hard to work.

Another way is to change the signature of SocketInputDStream, like:

{code}
DStream[T] => InputDStream[T] => ReceiverInputDStream[T] => SocketInputDStream[M, T]
{code}

But this will require related socketStream API change to add another ClassTag.

To keep without changing the current API, while get the correct type information seems a little hard to design. Any suggestion would be greatly appreciated.


was (Author: jerryshao):
What confused me a lot is the type inference, as the interceptor add, say M => Iterable\[T\], so the type of DStream\[M\] need to be changed into DStream\[T\], while actually the type of input DStream is M, like:

{code}
DStream[T] => InputDStream[T] => ReceiverInputDStream[T] => SocketInputDStream[M]
{code}

Assuming I add the interceptor before the ReceiverInputDStream.

 I'm not sure there's way to do it? A possible way is changing the interceptor to Any => Iterable\[T\], but how to reflect the received data type is also a problem, still function like bytesToObjects: InputStream => Iterator\[T\] is hard to work.

Another way is to change the signature of SocketInputDStream, like:

{code}
DStream[T] => InputDStream[T] => ReceiverInputDStream[T] => SocketInputDStream[M, T]
{code}

But this will require related socketStream API change to add another ClassTag.

To keep without changing the current API, while get the correct type information seems a little hard to design. Any suggestion would be greatly appreciated.

> Interceptor pattern in receivers
> --------------------------------
>
>                 Key: SPARK-4960
>                 URL: https://issues.apache.org/jira/browse/SPARK-4960
>             Project: Spark
>          Issue Type: New Feature
>          Components: Streaming
>            Reporter: Tathagata Das
>
> Sometimes it is good to intercept a message received through a receiver and modify / do something with the message before it is stored into Spark. This is often referred to as the interceptor pattern. There should be general way to specify an interceptor function that gets applied to all receivers. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org