Flink not reading from Kafka
I have this code trying to read from a topic however the flink process comes up and waits forever even though there is data in the topic. Not sure why? Has anyone else seen this problem? StreamExecutionEnvironment env = StreamExecutionEnvironment .*createLocalEnvironment*(); Properties properties = *new* Properties(); properties.setProperty("bootstrap.servers", ":9092"); properties.setProperty("group.id", "test1"); properties.setProperty("auto.offset.reset", "earliest"); FlatMapFunction> flatMapper = //something DataStream stream = env .addSource(*new* FlinkKafkaConsumer010<>("test", *new* SimpleStringSchema(), properties)); stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( *new* TypeHint>() { }).print(); JobExecutionResult *res* = env.execute(); 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: Unnamed(4/4) switched to RUNNING 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: Unnamed(1/4) switched to RUNNING 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: Unnamed(2/4) switched to RUNNING
Re: Flink not reading from Kafka
Interestingly enough same job runs ok on Linux but not on windows On Fri, Feb 17, 2017 at 4:54 PM, Mohit Anchlia wrote: > I have this code trying to read from a topic however the flink process > comes up and waits forever even though there is data in the topic. Not sure > why? Has anyone else seen this problem? > > StreamExecutionEnvironment env = StreamExecutionEnvironment > > .*createLocalEnvironment*(); > > Properties properties = *new* Properties(); > > properties.setProperty("bootstrap.servers", ":9092"); > > properties.setProperty("group.id", "test1"); > > properties.setProperty("auto.offset.reset", "earliest"); > > FlatMapFunction> flatMapper = > //something > > > > DataStream stream = env > > .addSource(*new* FlinkKafkaConsumer010<>("test", *new* > SimpleStringSchema(), properties)); > > stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( > > *new* TypeHint>() { > > }).print(); > > JobExecutionResult *res* = env.execute(); > > > > 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: > Unnamed(4/4) switched to RUNNING > > > 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: > Unnamed(1/4) switched to RUNNING > > > 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: > Unnamed(2/4) switched to RUNNING > > >
Re: Flink not reading from Kafka
Hi Mohit, is there new data being produced into the topic? The properties.setProperty("auto.offset.reset", "earliest"); setting only applies if you haven't consumed anything in this consumer group. So if you have read all the data in the topic before, you won't see anything new showing up. On Sat, Feb 18, 2017 at 2:09 AM, Mohit Anchlia wrote: > Interestingly enough same job runs ok on Linux but not on windows > > On Fri, Feb 17, 2017 at 4:54 PM, Mohit Anchlia > wrote: > >> I have this code trying to read from a topic however the flink process >> comes up and waits forever even though there is data in the topic. Not sure >> why? Has anyone else seen this problem? >> >> StreamExecutionEnvironment env = StreamExecutionEnvironment >> >> .*createLocalEnvironment*(); >> >> Properties properties = *new* Properties(); >> >> properties.setProperty("bootstrap.servers", ":9092"); >> >> properties.setProperty("group.id", "test1"); >> >> properties.setProperty("auto.offset.reset", "earliest"); >> >> FlatMapFunction> flatMapper = >> //something >> >> >> >> DataStream stream = env >> >> .addSource(*new* FlinkKafkaConsumer010<>("test", *new* >> SimpleStringSchema(), properties)); >> >> stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( >> >> *new* TypeHint>() { >> >> }).print(); >> >> JobExecutionResult *res* = env.execute(); >> >> >> >> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >> Unnamed(4/4) switched to RUNNING >> >> >> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >> Unnamed(1/4) switched to RUNNING >> >> >> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >> Unnamed(2/4) switched to RUNNING >> >> >> >
Re: Flink not reading from Kafka
I was facing a similar problem yesterday. In my case print() was not working. Try adding a Sink and write the output to another Kafka topic. Something like https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka/WriteIntoKafka.java#L71 .. It worked for me. Is the stdout disabled somehow by default ? regards. On Thu, Feb 23, 2017 at 9:42 PM, Robert Metzger wrote: > Hi Mohit, > > is there new data being produced into the topic? > The properties.setProperty("auto.offset.reset", "earliest"); setting only > applies if you haven't consumed anything in this consumer group. > So if you have read all the data in the topic before, you won't see > anything new showing up. > > On Sat, Feb 18, 2017 at 2:09 AM, Mohit Anchlia > wrote: > >> Interestingly enough same job runs ok on Linux but not on windows >> >> On Fri, Feb 17, 2017 at 4:54 PM, Mohit Anchlia >> wrote: >> >>> I have this code trying to read from a topic however the flink process >>> comes up and waits forever even though there is data in the topic. Not sure >>> why? Has anyone else seen this problem? >>> >>> StreamExecutionEnvironment env = StreamExecutionEnvironment >>> >>> .*createLocalEnvironment*(); >>> >>> Properties properties = *new* Properties(); >>> >>> properties.setProperty("bootstrap.servers", ":9092"); >>> >>> properties.setProperty("group.id", "test1"); >>> >>> properties.setProperty("auto.offset.reset", "earliest"); >>> >>> FlatMapFunction> flatMapper = >>> //something >>> >>> >>> >>> DataStream stream = env >>> >>> .addSource(*new* FlinkKafkaConsumer010<>("test", *new* >>> SimpleStringSchema(), properties)); >>> >>> stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( >>> >>> *new* TypeHint>() { >>> >>> }).print(); >>> >>> JobExecutionResult *res* = env.execute(); >>> >>> >>> >>> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >>> Unnamed(4/4) switched to RUNNING >>> >>> >>> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >>> Unnamed(1/4) switched to RUNNING >>> >>> >>> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >>> Unnamed(2/4) switched to RUNNING >>> >>> >>> >> > -- Debasish Ghosh http://manning.com/ghosh2 http://manning.com/ghosh Twttr: @debasishg Blog: http://debasishg.blogspot.com Code: http://github.com/debasishg
Re: Flink not reading from Kafka
Hi Debashish, did you execute Flink in a distributed setting? print() will output the stream contents on stdout on the respective worker node (taskmanager), not on the machine that submitted the job. On Thu, Feb 23, 2017 at 5:41 PM, Debasish Ghosh wrote: > I was facing a similar problem yesterday. In my case print() was not > working. Try adding a Sink and write the output to another Kafka topic. > Something like https://github.com/apache/flink/blob/master/flink- > examples/flink-examples-streaming/src/main/java/org/ > apache/flink/streaming/examples/kafka/WriteIntoKafka.java#L71 .. > > It worked for me. Is the stdout disabled somehow by default ? > > regards. > > On Thu, Feb 23, 2017 at 9:42 PM, Robert Metzger > wrote: > >> Hi Mohit, >> >> is there new data being produced into the topic? >> The properties.setProperty("auto.offset.reset", "earliest"); setting >> only applies if you haven't consumed anything in this consumer group. >> So if you have read all the data in the topic before, you won't see >> anything new showing up. >> >> On Sat, Feb 18, 2017 at 2:09 AM, Mohit Anchlia >> wrote: >> >>> Interestingly enough same job runs ok on Linux but not on windows >>> >>> On Fri, Feb 17, 2017 at 4:54 PM, Mohit Anchlia >>> wrote: >>> I have this code trying to read from a topic however the flink process comes up and waits forever even though there is data in the topic. Not sure why? Has anyone else seen this problem? StreamExecutionEnvironment env = StreamExecutionEnvironment .*createLocalEnvironment*(); Properties properties = *new* Properties(); properties.setProperty("bootstrap.servers", ":9092"); properties.setProperty("group.id", "test1"); properties.setProperty("auto.offset.reset", "earliest"); FlatMapFunction> flatMapper = //something DataStream stream = env .addSource(*new* FlinkKafkaConsumer010<>("test", *new* SimpleStringSchema(), properties)); stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( *new* TypeHint>() { }).print(); JobExecutionResult *res* = env.execute(); 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: Unnamed(4/4) switched to RUNNING 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: Unnamed(1/4) switched to RUNNING 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: Unnamed(2/4) switched to RUNNING >>> >> > > > -- > Debasish Ghosh > http://manning.com/ghosh2 > http://manning.com/ghosh > > Twttr: @debasishg > Blog: http://debasishg.blogspot.com > Code: http://github.com/debasishg >
Re: Flink not reading from Kafka
Yes .. I was running Flink on a DC/OS cluster. AFAIR I checked the taskmanager log from the Flink UI in Mesos. It said stdout was not available. But this may be due to the fact that Flink on DC/OS is not yet very stable .. regards. On Fri, Feb 24, 2017 at 1:41 AM, Robert Metzger wrote: > Hi Debashish, > > did you execute Flink in a distributed setting? print() will output the > stream contents on stdout on the respective worker node (taskmanager), not > on the machine that submitted the job. > > On Thu, Feb 23, 2017 at 5:41 PM, Debasish Ghosh > wrote: > >> I was facing a similar problem yesterday. In my case print() was not >> working. Try adding a Sink and write the output to another Kafka topic. >> Something like https://github.com/apache/flin >> k/blob/master/flink-examples/flink-examples-streaming/src/ >> main/java/org/apache/flink/streaming/examples/kafka/ >> WriteIntoKafka.java#L71 .. >> >> It worked for me. Is the stdout disabled somehow by default ? >> >> regards. >> >> On Thu, Feb 23, 2017 at 9:42 PM, Robert Metzger >> wrote: >> >>> Hi Mohit, >>> >>> is there new data being produced into the topic? >>> The properties.setProperty("auto.offset.reset", "earliest"); setting >>> only applies if you haven't consumed anything in this consumer group. >>> So if you have read all the data in the topic before, you won't see >>> anything new showing up. >>> >>> On Sat, Feb 18, 2017 at 2:09 AM, Mohit Anchlia >>> wrote: >>> Interestingly enough same job runs ok on Linux but not on windows On Fri, Feb 17, 2017 at 4:54 PM, Mohit Anchlia wrote: > I have this code trying to read from a topic however the flink process > comes up and waits forever even though there is data in the topic. Not > sure > why? Has anyone else seen this problem? > > StreamExecutionEnvironment env = StreamExecutionEnvironment > > .*createLocalEnvironment*(); > > Properties properties = *new* Properties(); > > properties.setProperty("bootstrap.servers", ":9092"); > > properties.setProperty("group.id", "test1"); > > properties.setProperty("auto.offset.reset", "earliest"); > > FlatMapFunction> flatMapper = > //something > > > > DataStream stream = env > > .addSource(*new* FlinkKafkaConsumer010<>("test", *new* > SimpleStringSchema(), properties)); > > stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( > > *new* TypeHint>() { > > }).print(); > > JobExecutionResult *res* = env.execute(); > > > > 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: > Unnamed(4/4) switched to RUNNING > > > 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: > Unnamed(1/4) switched to RUNNING > > > 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: > Unnamed(2/4) switched to RUNNING > > > >>> >> >> >> -- >> Debasish Ghosh >> http://manning.com/ghosh2 >> http://manning.com/ghosh >> >> Twttr: @debasishg >> Blog: http://debasishg.blogspot.com >> Code: http://github.com/debasishg >> > > -- Debasish Ghosh http://manning.com/ghosh2 http://manning.com/ghosh Twttr: @debasishg Blog: http://debasishg.blogspot.com Code: http://github.com/debasishg
Re: Flink not reading from Kafka
Hi, It is possible that the stdout file is not properly available in the taskmanager UI. I guess if you log into the machine directly to get the stout file, you'll find the output. On Thu, Feb 23, 2017 at 9:24 PM, Debasish Ghosh wrote: > Yes .. I was running Flink on a DC/OS cluster. > > AFAIR I checked the taskmanager log from the Flink UI in Mesos. It said > stdout was not available. But this may be due to the fact that Flink on > DC/OS is not yet very stable .. > > regards. > > On Fri, Feb 24, 2017 at 1:41 AM, Robert Metzger > wrote: > >> Hi Debashish, >> >> did you execute Flink in a distributed setting? print() will output the >> stream contents on stdout on the respective worker node (taskmanager), not >> on the machine that submitted the job. >> >> On Thu, Feb 23, 2017 at 5:41 PM, Debasish Ghosh > > wrote: >> >>> I was facing a similar problem yesterday. In my case print() was not >>> working. Try adding a Sink and write the output to another Kafka topic. >>> Something like https://github.com/apache/flin >>> k/blob/master/flink-examples/flink-examples-streaming/src/ma >>> in/java/org/apache/flink/streaming/examples/kafka/WriteIntoK >>> afka.java#L71 .. >>> >>> It worked for me. Is the stdout disabled somehow by default ? >>> >>> regards. >>> >>> On Thu, Feb 23, 2017 at 9:42 PM, Robert Metzger >>> wrote: >>> Hi Mohit, is there new data being produced into the topic? The properties.setProperty("auto.offset.reset", "earliest"); setting only applies if you haven't consumed anything in this consumer group. So if you have read all the data in the topic before, you won't see anything new showing up. On Sat, Feb 18, 2017 at 2:09 AM, Mohit Anchlia wrote: > Interestingly enough same job runs ok on Linux but not on windows > > On Fri, Feb 17, 2017 at 4:54 PM, Mohit Anchlia > wrote: > >> I have this code trying to read from a topic however the flink >> process comes up and waits forever even though there is data in the >> topic. >> Not sure why? Has anyone else seen this problem? >> >> StreamExecutionEnvironment env = StreamExecutionEnvironment >> >> .*createLocalEnvironment*(); >> >> Properties properties = *new* Properties(); >> >> properties.setProperty("bootstrap.servers", ":9092"); >> >> properties.setProperty("group.id", "test1"); >> >> properties.setProperty("auto.offset.reset", "earliest"); >> >> FlatMapFunction> flatMapper = >> //something >> >> >> >> DataStream stream = env >> >> .addSource(*new* FlinkKafkaConsumer010<>("test", *new* >> SimpleStringSchema(), properties)); >> >> stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( >> >> *new* TypeHint>() { >> >> }).print(); >> >> JobExecutionResult *res* = env.execute(); >> >> >> >> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >> Unnamed(4/4) switched to RUNNING >> >> >> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >> Unnamed(1/4) switched to RUNNING >> >> >> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> Sink: >> Unnamed(2/4) switched to RUNNING >> >> >> > >>> >>> >>> -- >>> Debasish Ghosh >>> http://manning.com/ghosh2 >>> http://manning.com/ghosh >>> >>> Twttr: @debasishg >>> Blog: http://debasishg.blogspot.com >>> Code: http://github.com/debasishg >>> >> >> > > > -- > Debasish Ghosh > http://manning.com/ghosh2 > http://manning.com/ghosh > > Twttr: @debasishg > Blog: http://debasishg.blogspot.com > Code: http://github.com/debasishg >
Re: Flink not reading from Kafka
May be .. I will try to log in to the machine directly and see .. regards. On Fri, Feb 24, 2017 at 2:05 AM, Robert Metzger wrote: > Hi, > > It is possible that the stdout file is not properly available in the > taskmanager UI. > I guess if you log into the machine directly to get the stout file, you'll > find the output. > > On Thu, Feb 23, 2017 at 9:24 PM, Debasish Ghosh > wrote: > >> Yes .. I was running Flink on a DC/OS cluster. >> >> AFAIR I checked the taskmanager log from the Flink UI in Mesos. It said >> stdout was not available. But this may be due to the fact that Flink on >> DC/OS is not yet very stable .. >> >> regards. >> >> On Fri, Feb 24, 2017 at 1:41 AM, Robert Metzger >> wrote: >> >>> Hi Debashish, >>> >>> did you execute Flink in a distributed setting? print() will output the >>> stream contents on stdout on the respective worker node (taskmanager), not >>> on the machine that submitted the job. >>> >>> On Thu, Feb 23, 2017 at 5:41 PM, Debasish Ghosh < >>> ghosh.debas...@gmail.com> wrote: >>> I was facing a similar problem yesterday. In my case print() was not working. Try adding a Sink and write the output to another Kafka topic. Something like https://github.com/apache/flin k/blob/master/flink-examples/flink-examples-streaming/src/ma in/java/org/apache/flink/streaming/examples/kafka/WriteIntoK afka.java#L71 .. It worked for me. Is the stdout disabled somehow by default ? regards. On Thu, Feb 23, 2017 at 9:42 PM, Robert Metzger wrote: > Hi Mohit, > > is there new data being produced into the topic? > The properties.setProperty("auto.offset.reset", "earliest"); setting > only applies if you haven't consumed anything in this consumer group. > So if you have read all the data in the topic before, you won't see > anything new showing up. > > On Sat, Feb 18, 2017 at 2:09 AM, Mohit Anchlia > wrote: > >> Interestingly enough same job runs ok on Linux but not on windows >> >> On Fri, Feb 17, 2017 at 4:54 PM, Mohit Anchlia < >> mohitanch...@gmail.com> wrote: >> >>> I have this code trying to read from a topic however the flink >>> process comes up and waits forever even though there is data in the >>> topic. >>> Not sure why? Has anyone else seen this problem? >>> >>> StreamExecutionEnvironment env = StreamExecutionEnvironment >>> >>> .*createLocalEnvironment*(); >>> >>> Properties properties = *new* Properties(); >>> >>> properties.setProperty("bootstrap.servers", ":9092"); >>> >>> properties.setProperty("group.id", "test1"); >>> >>> properties.setProperty("auto.offset.reset", "earliest"); >>> >>> FlatMapFunction> flatMapper = >>> //something >>> >>> >>> >>> DataStream stream = env >>> >>> .addSource(*new* FlinkKafkaConsumer010<>("test", *new* >>> SimpleStringSchema(), properties)); >>> >>> stream.map(s -> Integer.*valueOf*(s)).flatMap(flatMapper).returns( >>> >>> *new* TypeHint>() { >>> >>> }).print(); >>> >>> JobExecutionResult *res* = env.execute(); >>> >>> >>> >>> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> >>> Sink: Unnamed(4/4) switched to RUNNING >>> >>> >>> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> >>> Sink: Unnamed(1/4) switched to RUNNING >>> >>> >>> 02/17/2017 16:50:25 Source: Custom Source -> Map -> Flat Map -> >>> Sink: Unnamed(2/4) switched to RUNNING >>> >>> >>> >> > -- Debasish Ghosh http://manning.com/ghosh2 http://manning.com/ghosh Twttr: @debasishg Blog: http://debasishg.blogspot.com Code: http://github.com/debasishg >>> >>> >> >> >> -- >> Debasish Ghosh >> http://manning.com/ghosh2 >> http://manning.com/ghosh >> >> Twttr: @debasishg >> Blog: http://debasishg.blogspot.com >> Code: http://github.com/debasishg >> > > -- Debasish Ghosh http://manning.com/ghosh2 http://manning.com/ghosh Twttr: @debasishg Blog: http://debasishg.blogspot.com Code: http://github.com/debasishg