flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Suneel Marthi <suneel.mar...@gmail.com>
Subject Re: setSlotSharing NPE: Starting a stream consumer in a thread
Date Sun, 04 Oct 2015 04:01:51 GMT
While on that Marton,  would it make sense to have a
dataStream.writeAsJson() method?

On Sat, Oct 3, 2015 at 11:54 PM, Márton Balassi <balassi.marton@gmail.com>
wrote:

> Hi Jay,
>
> As for the NPE: the file monitoring function throws it when the location
> is empty. Try running the datagenerator first! :) This behaviour is
> unwanted though, I am adding a JIRA ticket for it.
>
> Best,
>
> Marton
>
> On Sun, Oct 4, 2015 at 5:28 AM, Márton Balassi <balassi.marton@gmail.com>
> wrote:
>
>> Hi Jay,
>>
>> Creating a batch and a streaming environment in a single Java source file
>> is fine, they just run separately. (If you run it from an IDE locally they
>> might conflict as the second one would try to launch a local executor on a
>> port that is most likely already taken by the first one.) I would suggest
>> to have these jobs in separate files currently, exactly for the previous
>> reason.
>>
>> Looking at your
>> code ExecutionEnvironment.getExecutionEnvironment().registerTypeWithKryoSerializer(com.github.rnowling.bps.datagenerator.datamodels.Product.class,
>> new FlinkBPSGenerator.ProductSerializer()); does not do much good for you.
>> You need to register your serializers to the environment to which you are
>> using. Currently you would need to register it to the streaming env
>> variable. If you would like to also assemble a batch job you need to add
>> them there too.
>>
>> As for the streaming job I assume that you are using Flink version 0.9.1
>> and checking out the problem shortly.
>>
>> Best,
>>
>> Marton
>>
>> On Sun, Oct 4, 2015 at 3:37 AM, jay vyas <jayunit100.apache@gmail.com>
>> wrote:
>>
>>> Here is a distilled example of the issue, should be easier to debug for
>>> folks interested... :)
>>>
>>> public static void main(String[] args) {
>>>
>>>
>>>     ExecutionEnvironment.getExecutionEnvironment().registerTypeWithKryoSerializer(com.github.rnowling.bps.datagenerator.datamodels.Product.class,
new FlinkBPSGenerator.ProductSerializer());
>>>     ExecutionEnvironment.getExecutionEnvironment().registerTypeWithKryoSerializer(com.github.rnowling.bps.datagenerator.datamodels.Transaction.class,
new FlinkBPSGenerator.TransactionSerializer());
>>>
>>>     final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>>>
>>>     //when running "env.execute" this stream should start consuming...
>>>     DataStream<String> dataStream = env.readFileStream("/tmp/a", 1000,
FileMonitoringFunction.WatchType.ONLY_NEW_FILES);
>>>     dataStream.iterate().map(new MapFunction<String, String>() {
>>>         public String map(String value) throws Exception {
>>>             System.out.println(value);
>>>             return ">>> > > > > " + value + " < <
< <  <<<";
>>>         }
>>>     });
>>>     try {
>>>         env.execute();
>>>     }
>>>     catch(Exception e){
>>>         e.printStackTrace();
>>>     }
>>> }
>>>
>>>
>>>
>>> On Sat, Oct 3, 2015 at 9:08 PM, jay vyas <jayunit100.apache@gmail.com>
>>> wrote:
>>>
>>>> Hi flink !
>>>>
>>>> Looks like "setSlotSharing" is throwing an NPE when I try to start a
>>>> Thread  which runs a streaming job.
>>>>
>>>> I'm trying to do this by creating a dataStream from env.readFileStream,
>>>> and then later starting a job which writes files out ...
>>>>
>>>> However, I get
>>>>
>>>> Exception in thread "main" java.lang.NullPointerException
>>>>     at
>>>> org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.setSlotSharing(StreamingJobGraphGenerator.java:361)
>>>>
>>>> What is the right way to create a stream and batch job all in one
>>>> environment?
>>>>
>>>> For reference, here is a gist of the code
>>>> https://gist.github.com/jayunit100/c7ab61d1833708d290df, and the
>>>> offending line is the
>>>>
>>>> DataStream<String> dataStream = env.readFileStream("/tmp/a",1000, FileMonitoringFunction.WatchType.ONLY_NEW_FILES);
>>>>
>>>> line.
>>>>
>>>> Thanks again !
>>>>
>>>> --
>>>> jay vyas
>>>>
>>>
>>>
>>>
>>> --
>>> jay vyas
>>>
>>
>>
>

Mime
View raw message