Re: Flink keyed stream windows
Clarification: Its 30 Seconds not 30 minutes. On Mon, Aug 13, 2018 at 3:20 PM Garvit Sharma wrote: > Hi, > > I am working on a use case where I have a stream of users active locations > and I want to store(by hitting an HTTP API) the latest active location for > each of the unique users every 30 minutes. > > Since I have a lot of unique users(rpm 1.5 million), how to use Flink's > timed windows on keyed stream to solve this problem. > > Please help! > > Thanks, > > -- > > Garvit Sharma > github.com/garvitlnmiit/ > > No Body is a Scholar by birth, its only hard work and strong determination > that makes him master. > -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Flink keyed stream windows
Hi, I am working on a use case where I have a stream of users active locations and I want to store(by hitting an HTTP API) the latest active location for each of the unique users every 30 minutes. Since I have a lot of unique users(rpm 1.5 million), how to use Flink's timed windows on keyed stream to solve this problem. Please help! Thanks, -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Re: Cleaning of state snapshot in state backend(HDFS)
Thank you for the clarification.
On Thu, Jun 21, 2018 at 1:36 PM sihua zhou wrote:
> Yes, you can clear the state for a key(the currently active key), if you
> clear it, it means that you have also cleaned it from the state backend,
> and the future checpoints won't contains the key anymore unless you add it
> again.
>
> Best, Sihua
>
>
> On 06/21/2018 16:04，Garvit Sharma
> wrote：
>
> Now, after clearing state for a key, I don't want that redundant data in
> the state backend. This is my concern.
>
> Please let me know if there are any gaps.
>
> Thanks,
>
> On Thu, Jun 21, 2018 at 1:31 PM Garvit Sharma wrote:
>
>> I am maintaining state data for a key in ValueState. As per [0] I can
>> clear() state for that key.
>>
>> [0]
>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/state.html
>>
>> Please let me know.
>>
>> Thanks,
>>
>>
>> On Thu, Jun 21, 2018 at 1:19 PM sihua zhou wrote:
>>
>>> Hi Garvit,
>>>
>>> Let's say you clearing the state at timestamp t1, then the checkpoints
>>> completed before t1 will still contains the data you cleared. But the
>>> future checkpoints won't contain the cleared data again. But I'm not sure
>>> what you meaning by the cleared state, you can only clear a key-value pair
>>> of the state currently, you can't cleared the whole state currently.
>>>
>>> Best, Sihua
>>>
>>> On 06/21/2018 15:41，Garvit Sharma
>>> wrote：
>>>
>>> So, would it delete all the files in HDFS associated with the cleared
>>> state?
>>>
>>> On Thu, Jun 21, 2018 at 12:58 PM sihua zhou wrote:
>>>
>>>> Hi Garvit,
>>>>
>>>> > Now, let's say, we clear the state. Would the state data be removed
>>>> from HDFS too?
>>>>
>>>> The state data would not be removed from HDFS immediately, if you clear
>>>> the state in your job. But after you clearing the state in your job, the
>>>> later completed checkpoint won't contain the state any more.
>>>>
>>>> > How does Flink manage to clear the state data from state backend on
>>>> clearing the keyed state?
>>>>
>>>> 1. you can use the {{tate.checkpoints.num-retained}} to set the number
>>>> of the completed checkpoint maintanced on HDFS.
>>>> 2. If you set {{
>>>> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
>>>> DELETE_ON_CANCELLATION)}} then the checkpoints on HDFS will be removed
>>>> once your job is finished(or cancled). And if you set {{
>>>> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
>>>> RETAIN_ON_CANCELLATION)}} then the checkpoints will be remained.
>>>>
>>>> Please refer to
>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/ops/state/checkpoints.html
>>>> to
>>>> find more information.
>>>>
>>>>
>>>> Additional, I'd like to give a bref info of the checkpoint on HDFS. In
>>>> a nutshell, what ever you did with the state in your running job, they only
>>>> effect the content on the state backend locally. When checkpointing, flink
>>>> takes a snapshot of the local state backend, and send it to the checkpoint
>>>> target directory(in your case, the HDFS). The checkpoints on the HDFS looks
>>>> like the periodic snapshot of the state backend of your job, they can be
>>>> created or deleted but never be changed. Maybe Stefan(cc) could give you
>>>> more professional information and plz correct me if I'm incorrect.
>>>>
>>>> Best, Sihua
>>>> On 06/21/2018 14:40，Garvit Sharma
>>>> wrote：
>>>>
>>>> Hi,
>>>>
>>>> Consider a managed keyed state backed by HDFS with checkpointing
>>>> enabled. Now, as the state grows the state data will be saved on HDFS.
>>>>
>>>> Now, let's say, we clear the state. Would the state data be removed
>>>> from HDFS too?
>>>>
>>>> How does Flink manage to clear the state data from state backend on
>>>> clearing the keyed state?
>>>>
>>>> --
>>>>
>>>> Garvit Sharma
>>>> github.com/garvitlnmiit/
>>>>
>>>> No Body is a Scholar by birth, its only hard work and strong
>>>> determination that makes him master.
>>>>
>>>>
>>>
>>> --
>>>
>>> Garvit Sharma
>>> github.com/garvitlnmiit/
>>>
>>> No Body is a Scholar by birth, its only hard work and strong
>>> determination that makes him master.
>>>
>>>
>>
>> --
>>
>> Garvit Sharma
>> github.com/garvitlnmiit/
>>
>> No Body is a Scholar by birth, its only hard work and strong
>> determination that makes him master.
>>
>
>
> --
>
> Garvit Sharma
> github.com/garvitlnmiit/
>
> No Body is a Scholar by birth, its only hard work and strong determination
> that makes him master.
>
>
--
Garvit Sharma
github.com/garvitlnmiit/
No Body is a Scholar by birth, its only hard work and strong determination
that makes him master.
Re: Cleaning of state snapshot in state backend(HDFS)
Now, after clearing state for a key, I don't want that redundant data in
the state backend. This is my concern.
Please let me know if there are any gaps.
Thanks,
On Thu, Jun 21, 2018 at 1:31 PM Garvit Sharma wrote:
> I am maintaining state data for a key in ValueState. As per [0] I can
> clear() state for that key.
>
> [0]
> https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/state.html
>
> Please let me know.
>
> Thanks,
>
>
> On Thu, Jun 21, 2018 at 1:19 PM sihua zhou wrote:
>
>> Hi Garvit,
>>
>> Let's say you clearing the state at timestamp t1, then the checkpoints
>> completed before t1 will still contains the data you cleared. But the
>> future checkpoints won't contain the cleared data again. But I'm not sure
>> what you meaning by the cleared state, you can only clear a key-value pair
>> of the state currently, you can't cleared the whole state currently.
>>
>> Best, Sihua
>>
>> On 06/21/2018 15:41，Garvit Sharma
>> wrote：
>>
>> So, would it delete all the files in HDFS associated with the cleared
>> state?
>>
>> On Thu, Jun 21, 2018 at 12:58 PM sihua zhou wrote:
>>
>>> Hi Garvit,
>>>
>>> > Now, let's say, we clear the state. Would the state data be removed
>>> from HDFS too?
>>>
>>> The state data would not be removed from HDFS immediately, if you clear
>>> the state in your job. But after you clearing the state in your job, the
>>> later completed checkpoint won't contain the state any more.
>>>
>>> > How does Flink manage to clear the state data from state backend on
>>> clearing the keyed state?
>>>
>>> 1. you can use the {{tate.checkpoints.num-retained}} to set the number
>>> of the completed checkpoint maintanced on HDFS.
>>> 2. If you set {{
>>> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
>>> DELETE_ON_CANCELLATION)}} then the checkpoints on HDFS will be removed
>>> once your job is finished(or cancled). And if you set {{
>>> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
>>> RETAIN_ON_CANCELLATION)}} then the checkpoints will be remained.
>>>
>>> Please refer to
>>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/ops/state/checkpoints.html
>>> to
>>> find more information.
>>>
>>>
>>> Additional, I'd like to give a bref info of the checkpoint on HDFS. In a
>>> nutshell, what ever you did with the state in your running job, they only
>>> effect the content on the state backend locally. When checkpointing, flink
>>> takes a snapshot of the local state backend, and send it to the checkpoint
>>> target directory(in your case, the HDFS). The checkpoints on the HDFS looks
>>> like the periodic snapshot of the state backend of your job, they can be
>>> created or deleted but never be changed. Maybe Stefan(cc) could give you
>>> more professional information and plz correct me if I'm incorrect.
>>>
>>> Best, Sihua
>>> On 06/21/2018 14:40，Garvit Sharma
>>> wrote：
>>>
>>> Hi,
>>>
>>> Consider a managed keyed state backed by HDFS with checkpointing
>>> enabled. Now, as the state grows the state data will be saved on HDFS.
>>>
>>> Now, let's say, we clear the state. Would the state data be removed from
>>> HDFS too?
>>>
>>> How does Flink manage to clear the state data from state backend on
>>> clearing the keyed state?
>>>
>>> --
>>>
>>> Garvit Sharma
>>> github.com/garvitlnmiit/
>>>
>>> No Body is a Scholar by birth, its only hard work and strong
>>> determination that makes him master.
>>>
>>>
>>
>> --
>>
>> Garvit Sharma
>> github.com/garvitlnmiit/
>>
>> No Body is a Scholar by birth, its only hard work and strong
>> determination that makes him master.
>>
>>
>
> --
>
> Garvit Sharma
> github.com/garvitlnmiit/
>
> No Body is a Scholar by birth, its only hard work and strong determination
> that makes him master.
>
--
Garvit Sharma
github.com/garvitlnmiit/
No Body is a Scholar by birth, its only hard work and strong determination
that makes him master.
Re: Cleaning of state snapshot in state backend(HDFS)
I am maintaining state data for a key in ValueState. As per [0] I can
clear() state for that key.
[0]
https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/state.html
Please let me know.
Thanks,
On Thu, Jun 21, 2018 at 1:19 PM sihua zhou wrote:
> Hi Garvit,
>
> Let's say you clearing the state at timestamp t1, then the checkpoints
> completed before t1 will still contains the data you cleared. But the
> future checkpoints won't contain the cleared data again. But I'm not sure
> what you meaning by the cleared state, you can only clear a key-value pair
> of the state currently, you can't cleared the whole state currently.
>
> Best, Sihua
>
> On 06/21/2018 15:41，Garvit Sharma
> wrote：
>
> So, would it delete all the files in HDFS associated with the cleared
> state?
>
> On Thu, Jun 21, 2018 at 12:58 PM sihua zhou wrote:
>
>> Hi Garvit,
>>
>> > Now, let's say, we clear the state. Would the state data be removed
>> from HDFS too?
>>
>> The state data would not be removed from HDFS immediately, if you clear
>> the state in your job. But after you clearing the state in your job, the
>> later completed checkpoint won't contain the state any more.
>>
>> > How does Flink manage to clear the state data from state backend on
>> clearing the keyed state?
>>
>> 1. you can use the {{tate.checkpoints.num-retained}} to set the number
>> of the completed checkpoint maintanced on HDFS.
>> 2. If you set {{
>> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
>> DELETE_ON_CANCELLATION)}} then the checkpoints on HDFS will be removed
>> once your job is finished(or cancled). And if you set {{
>> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
>> RETAIN_ON_CANCELLATION)}} then the checkpoints will be remained.
>>
>> Please refer to
>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/ops/state/checkpoints.html
>> to
>> find more information.
>>
>>
>> Additional, I'd like to give a bref info of the checkpoint on HDFS. In a
>> nutshell, what ever you did with the state in your running job, they only
>> effect the content on the state backend locally. When checkpointing, flink
>> takes a snapshot of the local state backend, and send it to the checkpoint
>> target directory(in your case, the HDFS). The checkpoints on the HDFS looks
>> like the periodic snapshot of the state backend of your job, they can be
>> created or deleted but never be changed. Maybe Stefan(cc) could give you
>> more professional information and plz correct me if I'm incorrect.
>>
>> Best, Sihua
>> On 06/21/2018 14:40，Garvit Sharma
>> wrote：
>>
>> Hi,
>>
>> Consider a managed keyed state backed by HDFS with checkpointing enabled.
>> Now, as the state grows the state data will be saved on HDFS.
>>
>> Now, let's say, we clear the state. Would the state data be removed from
>> HDFS too?
>>
>> How does Flink manage to clear the state data from state backend on
>> clearing the keyed state?
>>
>> --
>>
>> Garvit Sharma
>> github.com/garvitlnmiit/
>>
>> No Body is a Scholar by birth, its only hard work and strong
>> determination that makes him master.
>>
>>
>
> --
>
> Garvit Sharma
> github.com/garvitlnmiit/
>
> No Body is a Scholar by birth, its only hard work and strong determination
> that makes him master.
>
>
--
Garvit Sharma
github.com/garvitlnmiit/
No Body is a Scholar by birth, its only hard work and strong determination
that makes him master.
Re: Cleaning of state snapshot in state backend(HDFS)
So, would it delete all the files in HDFS associated with the cleared state?
On Thu, Jun 21, 2018 at 12:58 PM sihua zhou wrote:
> Hi Garvit,
>
> > Now, let's say, we clear the state. Would the state data be removed from
> HDFS too?
>
> The state data would not be removed from HDFS immediately, if you clear
> the state in your job. But after you clearing the state in your job, the
> later completed checkpoint won't contain the state any more.
>
> > How does Flink manage to clear the state data from state backend on
> clearing the keyed state?
>
> 1. you can use the {{tate.checkpoints.num-retained}} to set the number of
> the completed checkpoint maintanced on HDFS.
> 2. If you set {{
> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
> DELETE_ON_CANCELLATION)}} then the checkpoints on HDFS will be removed
> once your job is finished(or cancled). And if you set {{
> env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.
> RETAIN_ON_CANCELLATION)}} then the checkpoints will be remained.
>
> Please refer to
> https://ci.apache.org/projects/flink/flink-docs-release-1.5/ops/state/checkpoints.html
> to
> find more information.
>
>
> Additional, I'd like to give a bref info of the checkpoint on HDFS. In a
> nutshell, what ever you did with the state in your running job, they only
> effect the content on the state backend locally. When checkpointing, flink
> takes a snapshot of the local state backend, and send it to the checkpoint
> target directory(in your case, the HDFS). The checkpoints on the HDFS looks
> like the periodic snapshot of the state backend of your job, they can be
> created or deleted but never be changed. Maybe Stefan(cc) could give you
> more professional information and plz correct me if I'm incorrect.
>
> Best, Sihua
> On 06/21/2018 14:40，Garvit Sharma
> wrote：
>
> Hi,
>
> Consider a managed keyed state backed by HDFS with checkpointing enabled.
> Now, as the state grows the state data will be saved on HDFS.
>
> Now, let's say, we clear the state. Would the state data be removed from
> HDFS too?
>
> How does Flink manage to clear the state data from state backend on
> clearing the keyed state?
>
> --
>
> Garvit Sharma
> github.com/garvitlnmiit/
>
> No Body is a Scholar by birth, its only hard work and strong determination
> that makes him master.
>
>
--
Garvit Sharma
github.com/garvitlnmiit/
No Body is a Scholar by birth, its only hard work and strong determination
that makes him master.
Cleaning of state snapshot in state backend(HDFS)
Hi, Consider a managed keyed state backed by HDFS with checkpointing enabled. Now, as the state grows the state data will be saved on HDFS. Now, let's say, we clear the state. Would the state data be removed from HDFS too? How does Flink manage to clear the state data from state backend on clearing the keyed state? -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Re: Exception while submitting jobs through Yarn
So, finally, I have got this working. The issue was because of a poor
library which was using xerces 2.6 :).
In this process, I found few things missing from the doc would like to
contribute the same.
I really appreciate the support provided.
Thanks,
On Tue, 19 Jun 2018 at 4:05 PM, Ted Yu wrote:
> Since you're using a vendor's distro, I would suggest asking on their user
> forum.
>
> Cheers
>
> Original message ----
> From: Garvit Sharma
> Date: 6/19/18 3:34 AM (GMT-08:00)
> To: trohrm...@apache.org
> Cc: Amit Jain , Chesnay Schepler ,
> Ted Yu , user@flink.apache.org
> Subject: Re: Exception while submitting jobs through Yarn
>
> Any help on this?
>
> On Mon, Jun 18, 2018 at 11:31 PM Garvit Sharma
> wrote:
>
>> Yes, it is.
>>
>> On Mon, Jun 18, 2018 at 7:54 PM Till Rohrmann
>> wrote:
>>
>>> Is `/usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar` a link to `
>>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar`?
>>>
>>> On Mon, Jun 18, 2018 at 4:02 PM Garvit Sharma
>>> wrote:
>>>
>>>> I don't think I can access core-default as it comes with Hadoop jar
>>>>
>>>> On Mon, 18 Jun 2018 at 7:30 PM, Till Rohrmann
>>>> wrote:
>>>>
>>>>> Hmm, could you check whether core-default.xml contains any suspicious
>>>>> entries? Apparently xerces:2.9.1 cannot read it.
>>>>>
>>>>> On Mon, Jun 18, 2018 at 3:40 PM Garvit Sharma
>>>>> wrote:
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> After putting the following log in my code, I can see that the Xerces
>>>>>> version is - Xerces version : Xerces-J 2.9.1
>>>>>>
>>>>>> log.info("Xerces version : {}",
>>>>>> org.apache.xerces.impl.Version.getVersion());
>>>>>>
>>>>>> Also, following is the response of *$* *locate xerces* command on
>>>>>> the server -
>>>>>>
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/falcon/client/lib/xercesImpl-2.10.0.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/hbase/lib/xercesImpl-2.9.1.jar
>>>>>>
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/livy/jars/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/livy2/jars/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/oozie/lib/xercesImpl-2.10.0.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/oozie/libserver/xercesImpl-2.10.0.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/oozie/libtools/xercesImpl-2.10.0.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/slider/lib/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/spark2/jars/xercesImpl-2.9.1.jar
>>>>>>
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.1.0-129/zookeeper/lib/xercesMinimal-1.9.6.2.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/falcon/client/lib/xercesImpl-2.10.0.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/hbase/lib/xercesImpl-2.9.1.jar
>>>>>>
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/livy/jars/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/livy2/jars/xercesImpl-2.9.1.jar
>>>>>>
>>>>>> /usr/hdp/2.6.3.0-235/oozie/lib/xercesImpl-2.10.0.jar
>>>>>>
>>>>>>
Re: Exception while submitting jobs through Yarn
Any help on this?
On Mon, Jun 18, 2018 at 11:31 PM Garvit Sharma wrote:
> Yes, it is.
>
> On Mon, Jun 18, 2018 at 7:54 PM Till Rohrmann
> wrote:
>
>> Is `/usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar` a link to `
>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar`?
>>
>> On Mon, Jun 18, 2018 at 4:02 PM Garvit Sharma
>> wrote:
>>
>>> I don't think I can access core-default as it comes with Hadoop jar
>>>
>>> On Mon, 18 Jun 2018 at 7:30 PM, Till Rohrmann
>>> wrote:
>>>
>>>> Hmm, could you check whether core-default.xml contains any suspicious
>>>> entries? Apparently xerces:2.9.1 cannot read it.
>>>>
>>>> On Mon, Jun 18, 2018 at 3:40 PM Garvit Sharma
>>>> wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> After putting the following log in my code, I can see that the Xerces
>>>>> version is - Xerces version : Xerces-J 2.9.1
>>>>>
>>>>> log.info("Xerces version : {}",
>>>>> org.apache.xerces.impl.Version.getVersion());
>>>>>
>>>>> Also, following is the response of *$* *locate xerces* command on the
>>>>> server -
>>>>>
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/falcon/client/lib/xercesImpl-2.10.0.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/hbase/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/livy/jars/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/livy2/jars/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/oozie/lib/xercesImpl-2.10.0.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/oozie/libserver/xercesImpl-2.10.0.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/oozie/libtools/xercesImpl-2.10.0.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/slider/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/spark2/jars/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.1.0-129/zookeeper/lib/xercesMinimal-1.9.6.2.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/falcon/client/lib/xercesImpl-2.10.0.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/hbase/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/livy/jars/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/livy2/jars/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/oozie/lib/xercesImpl-2.10.0.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/oozie/libserver/xercesImpl-2.10.0.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/oozie/libtools/xercesImpl-2.10.0.jar
>>>>>
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/ranger-admin/ews/webapp/WEB-INF/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/slider/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/spark2/jars/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
>>>>>
>>>>> /usr/hdp/2.6.3.0-235/zookeeper/lib/xercesMinimal-1.9.6.2.jar
>>>>>
>>>>> /usr/hdp/share/hst/hst-common/lib/xercesImpl-2.9.1.jar
>>>>>
>>>>> Now, I can say that the version of xerces are same.
>>>>>
>>>>>
>>>>> So, what is causing this issue if Xerces version is in sync?
>>>>>
>>>>>
>>>>> I am very excited to discover the issue :)
>>
Re: Exception while submitting jobs through Yarn
Yes, it is.
On Mon, Jun 18, 2018 at 7:54 PM Till Rohrmann wrote:
> Is `/usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar` a link to `
> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar`?
>
> On Mon, Jun 18, 2018 at 4:02 PM Garvit Sharma wrote:
>
>> I don't think I can access core-default as it comes with Hadoop jar
>>
>> On Mon, 18 Jun 2018 at 7:30 PM, Till Rohrmann
>> wrote:
>>
>>> Hmm, could you check whether core-default.xml contains any suspicious
>>> entries? Apparently xerces:2.9.1 cannot read it.
>>>
>>> On Mon, Jun 18, 2018 at 3:40 PM Garvit Sharma
>>> wrote:
>>>
>>>> Hi,
>>>>
>>>> After putting the following log in my code, I can see that the Xerces
>>>> version is - Xerces version : Xerces-J 2.9.1
>>>>
>>>> log.info("Xerces version : {}",
>>>> org.apache.xerces.impl.Version.getVersion());
>>>>
>>>> Also, following is the response of *$* *locate xerces* command on the
>>>> server -
>>>>
>>>>
>>>> /usr/hdp/2.6.1.0-129/falcon/client/lib/xercesImpl-2.10.0.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/hbase/lib/xercesImpl-2.9.1.jar
>>>>
>>>>
>>>> /usr/hdp/2.6.1.0-129/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/livy/jars/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/livy2/jars/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/oozie/lib/xercesImpl-2.10.0.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/oozie/libserver/xercesImpl-2.10.0.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/oozie/libtools/xercesImpl-2.10.0.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/slider/lib/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/spark2/jars/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.1.0-129/zookeeper/lib/xercesMinimal-1.9.6.2.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/falcon/client/lib/xercesImpl-2.10.0.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/hbase/lib/xercesImpl-2.9.1.jar
>>>>
>>>>
>>>> /usr/hdp/2.6.3.0-235/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/livy/jars/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/livy2/jars/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/oozie/lib/xercesImpl-2.10.0.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/oozie/libserver/xercesImpl-2.10.0.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/oozie/libtools/xercesImpl-2.10.0.jar
>>>>
>>>>
>>>> /usr/hdp/2.6.3.0-235/ranger-admin/ews/webapp/WEB-INF/lib/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/slider/lib/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/spark2/jars/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
>>>>
>>>> /usr/hdp/2.6.3.0-235/zookeeper/lib/xercesMinimal-1.9.6.2.jar
>>>>
>>>> /usr/hdp/share/hst/hst-common/lib/xercesImpl-2.9.1.jar
>>>>
>>>> Now, I can say that the version of xerces are same.
>>>>
>>>>
>>>> So, what is causing this issue if Xerces version is in sync?
>>>>
>>>>
>>>> I am very excited to discover the issue :)
>>>>
>>>>
>>>> Thanks,
>>>>
>>>> On Mon, Jun 18, 2018 at 6:27 PM Till Rohrmann
>>>> wrote:
>>>>
>>>>> Could you check which xerces version you have on your classpath?
>>>>> Apparently, it cannot read core-default.xml as Ted pointed out. This might
>>>>> be the root cause for the failure.
>>>>>
>>>>> Cheers,
>>>>> Till
>>>>>
>>
Re: Exception while submitting jobs through Yarn
I don't think I can access core-default as it comes with Hadoop jar
On Mon, 18 Jun 2018 at 7:30 PM, Till Rohrmann wrote:
> Hmm, could you check whether core-default.xml contains any suspicious
> entries? Apparently xerces:2.9.1 cannot read it.
>
> On Mon, Jun 18, 2018 at 3:40 PM Garvit Sharma wrote:
>
>> Hi,
>>
>> After putting the following log in my code, I can see that the Xerces
>> version is - Xerces version : Xerces-J 2.9.1
>>
>> log.info("Xerces version : {}", org.apache.xerces.impl.Version.getVersion());
>>
>> Also, following is the response of *$* *locate xerces* command on the
>> server -
>>
>>
>> /usr/hdp/2.6.1.0-129/falcon/client/lib/xercesImpl-2.10.0.jar
>>
>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl.jar
>>
>> /usr/hdp/2.6.1.0-129/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/hbase/lib/xercesImpl-2.9.1.jar
>>
>>
>> /usr/hdp/2.6.1.0-129/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/livy/jars/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/livy2/jars/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/oozie/lib/xercesImpl-2.10.0.jar
>>
>> /usr/hdp/2.6.1.0-129/oozie/libserver/xercesImpl-2.10.0.jar
>>
>> /usr/hdp/2.6.1.0-129/oozie/libtools/xercesImpl-2.10.0.jar
>>
>> /usr/hdp/2.6.1.0-129/slider/lib/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/spark2/jars/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.1.0-129/zookeeper/lib/xercesMinimal-1.9.6.2.jar
>>
>> /usr/hdp/2.6.3.0-235/falcon/client/lib/xercesImpl-2.10.0.jar
>>
>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar
>>
>> /usr/hdp/2.6.3.0-235/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/hbase/lib/xercesImpl-2.9.1.jar
>>
>>
>> /usr/hdp/2.6.3.0-235/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/livy/jars/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/livy2/jars/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/oozie/lib/xercesImpl-2.10.0.jar
>>
>> /usr/hdp/2.6.3.0-235/oozie/libserver/xercesImpl-2.10.0.jar
>>
>> /usr/hdp/2.6.3.0-235/oozie/libtools/xercesImpl-2.10.0.jar
>>
>>
>> /usr/hdp/2.6.3.0-235/ranger-admin/ews/webapp/WEB-INF/lib/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/slider/lib/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/spark2/jars/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
>>
>> /usr/hdp/2.6.3.0-235/zookeeper/lib/xercesMinimal-1.9.6.2.jar
>>
>> /usr/hdp/share/hst/hst-common/lib/xercesImpl-2.9.1.jar
>>
>> Now, I can say that the version of xerces are same.
>>
>>
>> So, what is causing this issue if Xerces version is in sync?
>>
>>
>> I am very excited to discover the issue :)
>>
>>
>> Thanks,
>>
>> On Mon, Jun 18, 2018 at 6:27 PM Till Rohrmann
>> wrote:
>>
>>> Could you check which xerces version you have on your classpath?
>>> Apparently, it cannot read core-default.xml as Ted pointed out. This might
>>> be the root cause for the failure.
>>>
>>> Cheers,
>>> Till
>>>
>>> On Mon, Jun 18, 2018 at 1:31 PM Garvit Sharma
>>> wrote:
>>>
>>>> Hi,
>>>>
>>>> Sorry for the confusion, but the yarn is running on Hadoop version 2.7
>>>> only and hence I am using Flink 1.5 Hadoop 2.7 binary.
>>>>
>>>> Below are the details provided by Yarn version command :
>>>>
>>>> Hadoop 2.7.3.2.6.3.0-235
>>>> Subversion g...@github.com:hortonworks/hadoop.git -r
>>>> 45bfd33bba8acadfa0e6024c80981c023b28d454
>>>> Compiled by jenkins on 2017-10-30T02:31Z
>>>> Compiled with protoc 2.5.0
>>>> From source with checksum cd1a4a466ef450f547c279989f3aa3
>>>> This command was run using
>>>> /usr/hdp/2.6.3.0-235/hadoop/hadoop-common-2.7.3.2.6.3.0-235.jar
>>>>
>>>> Please let me know if you have found the resolution to my issue :)
>>>>
>>>> Thanks,
>>>>
>>>>
>>>> On Mon, Jun 18, 2018
Re: Exception while submitting jobs through Yarn
Hi,
After putting the following log in my code, I can see that the Xerces
version is - Xerces version : Xerces-J 2.9.1
log.info("Xerces version : {}", org.apache.xerces.impl.Version.getVersion());
Also, following is the response of *$* *locate xerces* command on the
server -
/usr/hdp/2.6.1.0-129/falcon/client/lib/xercesImpl-2.10.0.jar
/usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/hadoop/client/xercesImpl.jar
/usr/hdp/2.6.1.0-129/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/hbase/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/livy/jars/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/livy2/jars/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/oozie/lib/xercesImpl-2.10.0.jar
/usr/hdp/2.6.1.0-129/oozie/libserver/xercesImpl-2.10.0.jar
/usr/hdp/2.6.1.0-129/oozie/libtools/xercesImpl-2.10.0.jar
/usr/hdp/2.6.1.0-129/slider/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/spark2/jars/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
/usr/hdp/2.6.1.0-129/zookeeper/lib/xercesMinimal-1.9.6.2.jar
/usr/hdp/2.6.3.0-235/falcon/client/lib/xercesImpl-2.10.0.jar
/usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/hadoop/client/xercesImpl.jar
/usr/hdp/2.6.3.0-235/hadoop-hdfs/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/hbase/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/hive-hcatalog/share/webhcat/svr/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/livy/jars/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/livy2/jars/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/oozie/lib/xercesImpl-2.10.0.jar
/usr/hdp/2.6.3.0-235/oozie/libserver/xercesImpl-2.10.0.jar
/usr/hdp/2.6.3.0-235/oozie/libtools/xercesImpl-2.10.0.jar
/usr/hdp/2.6.3.0-235/ranger-admin/ews/webapp/WEB-INF/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/slider/lib/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/spark2/jars/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/storm/contrib/storm-autocreds/xercesImpl-2.9.1.jar
/usr/hdp/2.6.3.0-235/zookeeper/lib/xercesMinimal-1.9.6.2.jar
/usr/hdp/share/hst/hst-common/lib/xercesImpl-2.9.1.jar
Now, I can say that the version of xerces are same.
So, what is causing this issue if Xerces version is in sync?
I am very excited to discover the issue :)
Thanks,
On Mon, Jun 18, 2018 at 6:27 PM Till Rohrmann wrote:
> Could you check which xerces version you have on your classpath?
> Apparently, it cannot read core-default.xml as Ted pointed out. This might
> be the root cause for the failure.
>
> Cheers,
> Till
>
> On Mon, Jun 18, 2018 at 1:31 PM Garvit Sharma wrote:
>
>> Hi,
>>
>> Sorry for the confusion, but the yarn is running on Hadoop version 2.7
>> only and hence I am using Flink 1.5 Hadoop 2.7 binary.
>>
>> Below are the details provided by Yarn version command :
>>
>> Hadoop 2.7.3.2.6.3.0-235
>> Subversion g...@github.com:hortonworks/hadoop.git -r
>> 45bfd33bba8acadfa0e6024c80981c023b28d454
>> Compiled by jenkins on 2017-10-30T02:31Z
>> Compiled with protoc 2.5.0
>> From source with checksum cd1a4a466ef450f547c279989f3aa3
>> This command was run using
>> /usr/hdp/2.6.3.0-235/hadoop/hadoop-common-2.7.3.2.6.3.0-235.jar
>>
>> Please let me know if you have found the resolution to my issue :)
>>
>> Thanks,
>>
>>
>> On Mon, Jun 18, 2018 at 4:50 PM Till Rohrmann
>> wrote:
>>
>>> Which Hadoop version have you installed? It looks as if Flink has been
>>> build with Hadoop 2.7 but I see /usr/hdp/2.6.3.0-235 in the class path. If
>>> you want to run Flink on Hadoop 2.6, then try to use the Hadoop free Flink
>>> binaries or the one built for Hadoop 2.6.
>>>
>>> Cheers,
>>> Till
>>>
>>> On Mon, Jun 18, 2018 at 10:48 AM Garvit Sharma
>>> wrote:
>>>
>>>> Ok, I have attached the log file.
>>>>
>>>> Please check and let me know.
>>>>
>>>> Thanks,
>>>>
>>>> On Mon, Jun 18, 2018 at 2:07 PM Amit Jain wrote:
>>>>
>>>>> Hi Gravit,
>>>>>
>>>>> I think Till is interested to know about classpath details present at
>>>>> the start of JM and TM logs e.g. following logs provide classpath details
>>>>> used by TM in our case.
>>>>>
>>>>> 2018-06-17 19:01:30,656 INFO
>>>>> org.apache.flink.yarn.YarnTaskExecutorRunner -
>>>>>
>>>>> 2018-06-17 19:01:30,658 INFO
>>>>> org.apache.flink.yarn.YarnTaskExecutorRunner - Sta
Re: Exception while submitting jobs through Yarn
Hi, Sorry for the confusion, but the yarn is running on Hadoop version 2.7 only and hence I am using Flink 1.5 Hadoop 2.7 binary. Below are the details provided by Yarn version command : Hadoop 2.7.3.2.6.3.0-235 Subversion g...@github.com:hortonworks/hadoop.git -r 45bfd33bba8acadfa0e6024c80981c023b28d454 Compiled by jenkins on 2017-10-30T02:31Z Compiled with protoc 2.5.0 >From source with checksum cd1a4a466ef450f547c279989f3aa3 This command was run using /usr/hdp/2.6.3.0-235/hadoop/hadoop-common-2.7.3.2.6.3.0-235.jar Please let me know if you have found the resolution to my issue :) Thanks, On Mon, Jun 18, 2018 at 4:50 PM Till Rohrmann wrote: > Which Hadoop version have you installed? It looks as if Flink has been > build with Hadoop 2.7 but I see /usr/hdp/2.6.3.0-235 in the class path. If > you want to run Flink on Hadoop 2.6, then try to use the Hadoop free Flink > binaries or the one built for Hadoop 2.6. > > Cheers, > Till > > On Mon, Jun 18, 2018 at 10:48 AM Garvit Sharma > wrote: > >> Ok, I have attached the log file. >> >> Please check and let me know. >> >> Thanks, >> >> On Mon, Jun 18, 2018 at 2:07 PM Amit Jain wrote: >> >>> Hi Gravit, >>> >>> I think Till is interested to know about classpath details present at >>> the start of JM and TM logs e.g. following logs provide classpath details >>> used by TM in our case. >>> >>> 2018-06-17 19:01:30,656 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> >>> 2018-06-17 19:01:30,658 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - Starting >>> YARN TaskExecutor runner (Version: 1.5.0, Rev:c61b108, Date:24.05.2018 @ >>> 14:54:44 UTC) >>> 2018-06-17 19:01:30,659 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - OS >>> current user: yarn >>> 2018-06-17 19:01:31,662 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - Current >>> Hadoop/Kerberos user: hadoop >>> 2018-06-17 19:01:31,663 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - JVM: >>> OpenJDK 64-Bit Server VM - Oracle Corporation - 1.8/25.171-b10 >>> 2018-06-17 19:01:31,663 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - Maximum >>> heap size: 6647 MiBytes >>> 2018-06-17 19:01:31,663 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> JAVA_HOME: /usr/lib/jvm/java-openjdk >>> 2018-06-17 19:01:31,664 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - Hadoop >>> version: 2.8.3 >>> 2018-06-17 19:01:31,664 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - JVM >>> Options: >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> -Xms6936m >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> -Xmx6936m >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> -XX:MaxDirectMemorySize=4072m >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> -Dlog.file=/var/log/hadoop-yarn/containers/application_1528342246614_0002/container_1528342246614_0002_01_282649/taskmanager.log >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> -Dlogback.configurationFile=file:./logback.xml >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> -Dlog4j.configuration=file:./log4j.properties >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - Program >>> Arguments: >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> --configDir >>> 2018-06-17 19:01:31,665 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - . >>> *2018-06-17 19:01:31,666 INFO >>> org.apache.flink.yarn.YarnTaskExecutorRunner - >>> Classpath: >>> lib/flink-dist_2.11-1.5.0.jar:lib/flink-python_2.11-1.5.0.jar:lib/flink-shaded-hadoop2-uber-1.5.0.jar:lib/flink-shaded-include-yarn-0.9.1.jar:lib/guava-18.0.jar:lib/log4j-1.2.17.jar:lib/slf4j-log4j12-1.7.7.jar:lo
Re: Exception while submitting jobs through Yarn
Ok, I have attached the log file. Please check and let me know. Thanks, On Mon, Jun 18, 2018 at 2:07 PM Amit Jain wrote: > Hi Gravit, > > I think Till is interested to know about classpath details present at the > start of JM and TM logs e.g. following logs provide classpath details used > by TM in our case. > > 2018-06-17 19:01:30,656 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - > > 2018-06-17 19:01:30,658 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - Starting YARN TaskExecutor runner (Version: 1.5.0, > Rev:c61b108, Date:24.05.2018 @ 14:54:44 UTC) > 2018-06-17 19:01:30,659 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - OS current user: yarn > 2018-06-17 19:01:31,662 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - Current Hadoop/Kerberos user: hadoop > 2018-06-17 19:01:31,663 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - JVM: OpenJDK 64-Bit Server VM - Oracle Corporation - > 1.8/25.171-b10 > 2018-06-17 19:01:31,663 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - Maximum heap size: 6647 MiBytes > 2018-06-17 19:01:31,663 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - JAVA_HOME: /usr/lib/jvm/java-openjdk > 2018-06-17 19:01:31,664 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - Hadoop version: 2.8.3 > 2018-06-17 19:01:31,664 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - JVM Options: > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - -Xms6936m > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - -Xmx6936m > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - -XX:MaxDirectMemorySize=4072m > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - > -Dlog.file=/var/log/hadoop-yarn/containers/application_1528342246614_0002/container_1528342246614_0002_01_282649/taskmanager.log > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - -Dlogback.configurationFile=file:./logback.xml > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - -Dlog4j.configuration=file:./log4j.properties > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - Program Arguments: > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - --configDir > 2018-06-17 19:01:31,665 INFO org.apache.flink.yarn.YarnTaskExecutorRunner > - . > *2018-06-17 19:01:31,666 INFO > org.apache.flink.yarn.YarnTaskExecutorRunner - > Classpath: > lib/flink-dist_2.11-1.5.0.jar:lib/flink-python_2.11-1.5.0.jar:lib/flink-shaded-hadoop2-uber-1.5.0.jar:lib/flink-shaded-include-yarn-0.9.1.jar:lib/guava-18.0.jar:lib/log4j-1.2.17.jar:lib/slf4j-log4j12-1.7.7.jar:log4j.properties:logback.xml:flink.jar:flink-conf.yaml::/etc/hadoop/conf:/usr/lib/hadoop/hadoop-common-2.8.3-amzn-0.jar:/usr/lib/hadoop/hadoop-archive-logs.jar:/usr/lib/hadoop/hadoop-auth.jar:/usr/lib/hadoop/hadoop-archives-2.8.3-amzn-0.jar:/usr/lib/hadoop/hadoop-archive-logs-2.8.3-amzn-0.jar:/usr/lib/hadoop/hadoop-azure-datalake-2.8.3-amzn-0.jar.* > > -- > Thanks, > Amit > > On Mon, Jun 18, 2018 at 2:00 PM, Garvit Sharma > wrote: > >> Hi, >> >> Please refer to my previous mail for complete logs. >> >> Thanks, >> >> On Mon, Jun 18, 2018 at 1:17 PM Till Rohrmann >> wrote: >> >>> Could you also please share the complete log file with us. >>> >>> Cheers, >>> Till >>> >>> On Sat, Jun 16, 2018 at 5:22 PM Ted Yu wrote: >>> >>>> The error for core-default.xml is interesting. >>>> >>>> Flink doesn't have this file. Probably it came with Yarn. Please check >>>> the hadoop version Flink was built with versus the hadoop version in your >>>> cluster. >>>> >>>> Thanks >>>> >>>> Original message >>>> From: Garvit Sharma >>>> Date: 6/16/18 7:23 AM (GMT-08:00) >>>> To: trohrm...@apache.org >>>> Cc: Chesnay Schepler , user@flink.apache.org >>>> Subject: Re: Exception while submitting jobs through Yarn >>>> >>>> I am not able to figure out, got stuck badly in this since last 1 week. >>>> Any little he
Re: Exception while submitting jobs through Yarn
Hi, Please refer to my previous mail for complete logs. Thanks, On Mon, Jun 18, 2018 at 1:17 PM Till Rohrmann wrote: > Could you also please share the complete log file with us. > > Cheers, > Till > > On Sat, Jun 16, 2018 at 5:22 PM Ted Yu wrote: > >> The error for core-default.xml is interesting. >> >> Flink doesn't have this file. Probably it came with Yarn. Please check >> the hadoop version Flink was built with versus the hadoop version in your >> cluster. >> >> Thanks >> >> Original message >> From: Garvit Sharma >> Date: 6/16/18 7:23 AM (GMT-08:00) >> To: trohrm...@apache.org >> Cc: Chesnay Schepler , user@flink.apache.org >> Subject: Re: Exception while submitting jobs through Yarn >> >> I am not able to figure out, got stuck badly in this since last 1 week. >> Any little help would be appreciated. >> >> >> 2018-06-16 19:25:10,523 DEBUG >> org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator - >> Parallelism set: 1 for 8 >> >> 2018-06-16 19:25:10,578 DEBUG >> org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator - >> Parallelism set: 1 for 1 >> >> 2018-06-16 19:25:10,588 DEBUG >> org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator - >> CONNECTED: KeyGroupStreamPartitioner - 1 -> 8 >> >> 2018-06-16 19:25:10,591 DEBUG >> org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator - >> Parallelism set: 1 for 5 >> >> 2018-06-16 19:25:10,597 DEBUG >> org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator - >> CONNECTED: KeyGroupStreamPartitioner - 5 -> 8 >> >> 2018-06-16 19:25:10,618 FATAL org.apache.hadoop.conf.Configuration >> - error parsing conf core-default.xml >> >> javax.xml.parsers.ParserConfigurationException: Feature ' >> http://apache.org/xml/features/xinclude' is not recognized. >> >> at >> org.apache.xerces.jaxp.DocumentBuilderFactoryImpl.newDocumentBuilder(Unknown >> Source) >> >> at >> org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2482) >> >> at >> org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2444) >> >> at org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2361) >> >> at org.apache.hadoop.conf.Configuration.get(Configuration.java:1188) >> >> at >> org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider.getRecordFactory(RecordFactoryProvider.java:49) >> >> at org.apache.hadoop.yarn.util.Records.(Records.java:32) >> >> at >> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getQueueInfoRequest(YarnClientImpl.java:495) >> >> at >> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getAllQueues(YarnClientImpl.java:525) >> >> at >> org.apache.flink.yarn.AbstractYarnClusterDescriptor.checkYarnQueues(AbstractYarnClusterDescriptor.java:658) >> >> at >> org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:486) >> >> at >> org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) >> >> at >> org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) >> >> at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) >> >> at >> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) >> >> at >> org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) >> >> at java.security.AccessController.doPrivileged(Native Method) >> >> at javax.security.auth.Subject.doAs(Subject.java:422) >> >> at >> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1692) >> >> at >> org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) >> >> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) >> >> 2018-06-16 19:25:10,620 WARN >> org.apache.flink.yarn.AbstractYarnClusterDescriptor >> - Error while getting queue information from YARN: null >> >> 2018-06-16 19:25:10,621 DEBUG >> org.apache.flink.yarn.AbstractYarnClusterDescriptor - Error >> details >> >> java.lang.ExceptionInInitializerError >> >> at >> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getQueueInfoRequest(YarnClientImpl.java:495) >> >> at >> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.g
Re: Exception while submitting jobs through Yarn
rserConfigurationException: Feature ' http://apache.org/xml/features/xinclude' is not recognized. at org.apache.xerces.jaxp.DocumentBuilderFactoryImpl.newDocumentBuilder(Unknown Source) at org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2482) ... 19 more 2018-06-16 19:25:10,627 DEBUG org.apache.hadoop.service.AbstractService - Service: org.apache.hadoop.yarn.client.api.impl.YarnClientImpl entered state STOPPED 2018-06-16 19:25:10,628 DEBUG org.apache.hadoop.ipc.Client - stopping client from cache: org.apache.hadoop.ipc.Client@32c726ee 2018-06-16 19:25:10,628 DEBUG org.apache.hadoop.ipc.Client - removing client from cache: org.apache.hadoop.ipc.Client@32c726ee 2018-06-16 19:25:10,628 DEBUG org.apache.hadoop.ipc.Client - stopping actual client because no more references remain: org.apache.hadoop.ipc.Client@32c726ee 2018-06-16 19:25:10,628 DEBUG org.apache.hadoop.ipc.Client - Stopping client 2018-06-16 19:25:10,629 DEBUG org.apache.hadoop.service.AbstractService - Service: org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl entered state STOPPED 2018-06-16 19:25:10,630 ERROR org.apache.flink.client.cli.CliFrontend - Fatal error while running command line interface. java.lang.NoClassDefFoundError: Could not initialize class org.apache.hadoop.yarn.util.Records at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.createApplication(YarnClientImpl.java:212) at org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:497) at org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) at org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1692) at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) On Fri, Jun 15, 2018 at 9:35 PM Till Rohrmann wrote: > Hmm could you maybe share the client logs with us. > > Cheers, > Till > > On Fri, Jun 15, 2018 at 4:54 PM Garvit Sharma wrote: > >> Yes, I did. >> >> On Fri, Jun 15, 2018 at 6:17 PM Till Rohrmann >> wrote: >> >>> Hi Garvit, >>> >>> have you exported the HADOOP_CLASSPATH as described in the release notes >>> [1]? >>> >>> [1] >>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/release-notes/flink-1.5.html#hadoop-classpath-discovery >>> >>> Cheers, >>> Till >>> >>> On Fri, Jun 15, 2018 at 2:22 PM Garvit Sharma >>> wrote: >>> >>>> Does someone has any idea how to get rid if the above parse exception >>>> while submitting flink job to Yarn. >>>> >>>> Already searched on the internet, could not find any solution to it. >>>> >>>> Please help. >>>> >>>> On Fri, Jun 15, 2018 at 9:15 AM Garvit Sharma >>>> wrote: >>>> >>>>> Thanks Chesnay, Now it is connecting to the Resource Manager but I am >>>>> getting the below exception : >>>>> >>>>> 2018-06-15 09:12:44,812 FATAL org.apache.hadoop.conf.Configuration >>>>> - error parsing conf core-default.xml >>>>> >>>>> javax.xml.parsers.ParserConfigurationException: Feature ' >>>>> http://apache.org/xml/features/xinclude' is not recognized. >>>>> >>>>> at >>>>> org.apache.xerces.jaxp.DocumentBuilderFactoryImpl.newDocumentBuilder(Unknown >>>>> Source) >>>>> >>>>> at >>>>> org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2737) >>>>> >>>>> at >>>>> org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2696) >>>>> >>>>> at >>>>> org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2579) >>>>> >>>>> at org.apache.hadoop.conf.Configuration.get(Configuration.java:1350) >>>>> >>>>> at >>>>> org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider.g
Re: Exception while submitting jobs through Yarn
Yes, I did. On Fri, Jun 15, 2018 at 6:17 PM Till Rohrmann wrote: > Hi Garvit, > > have you exported the HADOOP_CLASSPATH as described in the release notes > [1]? > > [1] > https://ci.apache.org/projects/flink/flink-docs-release-1.5/release-notes/flink-1.5.html#hadoop-classpath-discovery > > Cheers, > Till > > On Fri, Jun 15, 2018 at 2:22 PM Garvit Sharma wrote: > >> Does someone has any idea how to get rid if the above parse exception >> while submitting flink job to Yarn. >> >> Already searched on the internet, could not find any solution to it. >> >> Please help. >> >> On Fri, Jun 15, 2018 at 9:15 AM Garvit Sharma >> wrote: >> >>> Thanks Chesnay, Now it is connecting to the Resource Manager but I am >>> getting the below exception : >>> >>> 2018-06-15 09:12:44,812 FATAL org.apache.hadoop.conf.Configuration >>> - error parsing conf core-default.xml >>> >>> javax.xml.parsers.ParserConfigurationException: Feature ' >>> http://apache.org/xml/features/xinclude' is not recognized. >>> >>> at >>> org.apache.xerces.jaxp.DocumentBuilderFactoryImpl.newDocumentBuilder(Unknown >>> Source) >>> >>> at >>> org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2737) >>> >>> at >>> org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2696) >>> >>> at org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2579) >>> >>> at org.apache.hadoop.conf.Configuration.get(Configuration.java:1350) >>> >>> at >>> org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider.getRecordFactory(RecordFactoryProvider.java:49) >>> >>> at org.apache.hadoop.yarn.util.Records.(Records.java:32) >>> >>> at >>> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getQueueInfoRequest(YarnClientImpl.java:584) >>> >>> at >>> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getAllQueues(YarnClientImpl.java:614) >>> >>> at >>> org.apache.flink.yarn.AbstractYarnClusterDescriptor.checkYarnQueues(AbstractYarnClusterDescriptor.java:658) >>> >>> at >>> org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:486) >>> >>> at >>> org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) >>> >>> at >>> org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) >>> >>> at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) >>> >>> at >>> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) >>> >>> at >>> org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) >>> >>> at java.security.AccessController.doPrivileged(Native Method) >>> >>> at javax.security.auth.Subject.doAs(Subject.java:422) >>> >>> at >>> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836) >>> >>> at >>> org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) >>> >>> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) >>> >>> 2018-06-15 09:12:44,825 WARN >>> org.apache.flink.yarn.AbstractYarnClusterDescriptor >>> - Error while getting queue information from YARN: null >>> >>> java.lang.NoClassDefFoundError: Could not initialize class >>> org.apache.hadoop.yarn.util.Records >>> >>> at >>> org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.createApplication(YarnClientImpl.java:230) >>> >>> at >>> org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:497) >>> >>> at >>> org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) >>> >>> at >>> org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) >>> >>> at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) >>> >>> at >>> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) >>> >>> at >>> org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) >>> >>> at java.security.AccessController.doPrivileged(Native Method) >>
Re: Exception while submitting jobs through Yarn
Does someone has any idea how to get rid if the above parse exception while submitting flink job to Yarn. Already searched on the internet, could not find any solution to it. Please help. On Fri, Jun 15, 2018 at 9:15 AM Garvit Sharma wrote: > Thanks Chesnay, Now it is connecting to the Resource Manager but I am > getting the below exception : > > 2018-06-15 09:12:44,812 FATAL org.apache.hadoop.conf.Configuration > - error parsing conf core-default.xml > > javax.xml.parsers.ParserConfigurationException: Feature ' > http://apache.org/xml/features/xinclude' is not recognized. > > at > org.apache.xerces.jaxp.DocumentBuilderFactoryImpl.newDocumentBuilder(Unknown > Source) > > at > org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2737) > > at > org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2696) > > at org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2579) > > at org.apache.hadoop.conf.Configuration.get(Configuration.java:1350) > > at > org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider.getRecordFactory(RecordFactoryProvider.java:49) > > at org.apache.hadoop.yarn.util.Records.(Records.java:32) > > at > org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getQueueInfoRequest(YarnClientImpl.java:584) > > at > org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getAllQueues(YarnClientImpl.java:614) > > at > org.apache.flink.yarn.AbstractYarnClusterDescriptor.checkYarnQueues(AbstractYarnClusterDescriptor.java:658) > > at > org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:486) > > at > org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) > > at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) > > at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) > > at > org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) > > at > org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) > > at java.security.AccessController.doPrivileged(Native Method) > > at javax.security.auth.Subject.doAs(Subject.java:422) > > at > org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836) > > at > org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) > > at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) > > 2018-06-15 09:12:44,825 WARN > org.apache.flink.yarn.AbstractYarnClusterDescriptor > - Error while getting queue information from YARN: null > > java.lang.NoClassDefFoundError: Could not initialize class > org.apache.hadoop.yarn.util.Records > > at > org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.createApplication(YarnClientImpl.java:230) > > at > org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:497) > > at > org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) > > at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) > > at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) > > at > org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) > > at > org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) > > at java.security.AccessController.doPrivileged(Native Method) > > at javax.security.auth.Subject.doAs(Subject.java:422) > > at > org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836) > > at > org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) > > at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) > > Please help. > > Thanks, > > > On Thu, Jun 14, 2018 at 1:28 PM Chesnay Schepler > wrote: > >> My gut feeling is that these classes must be present in jars in the /lib >> directory. I don't think you can supply these with the submitted jar. >> For a simple test, put your jar into the /lib folder before submitting it. >> >> On 14.06.2018 06:56, Garvit Sharma wrote: >> >> Can someone please tell why am I facing this? >> >> On Wed, Jun 13, 2018 at 10:33 PM Garvit Sharma >> wrote: >> >>> Hi, >>> >>> I am using *flink-1.5.0-bin-hadoop27-scala_2.11 *to submit jobs through >>> Yarn, but I am getting the below exception : >>> >>> java.lang.NoClassDefFoundError: >>> com/sun/jersey/core/util/FeaturesAndProperties >>> >>&
Re: Exception while submitting jobs through Yarn
Thanks Chesnay, Now it is connecting to the Resource Manager but I am getting the below exception : 2018-06-15 09:12:44,812 FATAL org.apache.hadoop.conf.Configuration - error parsing conf core-default.xml javax.xml.parsers.ParserConfigurationException: Feature ' http://apache.org/xml/features/xinclude' is not recognized. at org.apache.xerces.jaxp.DocumentBuilderFactoryImpl.newDocumentBuilder(Unknown Source) at org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2737) at org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2696) at org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2579) at org.apache.hadoop.conf.Configuration.get(Configuration.java:1350) at org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider.getRecordFactory(RecordFactoryProvider.java:49) at org.apache.hadoop.yarn.util.Records.(Records.java:32) at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getQueueInfoRequest(YarnClientImpl.java:584) at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.getAllQueues(YarnClientImpl.java:614) at org.apache.flink.yarn.AbstractYarnClusterDescriptor.checkYarnQueues(AbstractYarnClusterDescriptor.java:658) at org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:486) at org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) at org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836) at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) 2018-06-15 09:12:44,825 WARN org.apache.flink.yarn.AbstractYarnClusterDescriptor - Error while getting queue information from YARN: null java.lang.NoClassDefFoundError: Could not initialize class org.apache.hadoop.yarn.util.Records at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.createApplication(YarnClientImpl.java:230) at org.apache.flink.yarn.AbstractYarnClusterDescriptor.deployInternal(AbstractYarnClusterDescriptor.java:497) at org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:75) at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:235) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) at org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836) at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) Please help. Thanks, On Thu, Jun 14, 2018 at 1:28 PM Chesnay Schepler wrote: > My gut feeling is that these classes must be present in jars in the /lib > directory. I don't think you can supply these with the submitted jar. > For a simple test, put your jar into the /lib folder before submitting it. > > On 14.06.2018 06:56, Garvit Sharma wrote: > > Can someone please tell why am I facing this? > > On Wed, Jun 13, 2018 at 10:33 PM Garvit Sharma > wrote: > >> Hi, >> >> I am using *flink-1.5.0-bin-hadoop27-scala_2.11 *to submit jobs through >> Yarn, but I am getting the below exception : >> >> java.lang.NoClassDefFoundError: >> com/sun/jersey/core/util/FeaturesAndProperties >> >> at java.lang.ClassLoader.defineClass1(Native Method) >> >> at java.lang.ClassLoader.defineClass(ClassLoader.java:763) >> >> at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142) >> >> at java.net.URLClassLoader.defineClass(URLClassLoader.java:467) >> >> at java.net.URLClassLoader.access$100(URLClassLoader.java:73) >> >> at java.net.URLClassLoader$1.run(URLClassLoader.java:368) >> >> at java.net.URLClassLoader$1.run(URLClassLoader.java:362) >> >> at java.security.AccessController.doPrivileged(Native Method) >> >> at java.net.URLClassLoader.findClass(URLClassLoader.java:361) >> >> at java.lang.ClassLoader.loadClass(ClassLoader.java:424) >> >> at sun.misc.Launcher
Re: Exception while submitting jobs through Yarn
Can someone please tell why am I facing this? On Wed, Jun 13, 2018 at 10:33 PM Garvit Sharma wrote: > Hi, > > I am using *flink-1.5.0-bin-hadoop27-scala_2.11 *to submit jobs through > Yarn, but I am getting the below exception : > > java.lang.NoClassDefFoundError: > com/sun/jersey/core/util/FeaturesAndProperties > > at java.lang.ClassLoader.defineClass1(Native Method) > > at java.lang.ClassLoader.defineClass(ClassLoader.java:763) > > at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142) > > at java.net.URLClassLoader.defineClass(URLClassLoader.java:467) > > at java.net.URLClassLoader.access$100(URLClassLoader.java:73) > > at java.net.URLClassLoader$1.run(URLClassLoader.java:368) > > at java.net.URLClassLoader$1.run(URLClassLoader.java:362) > > at java.security.AccessController.doPrivileged(Native Method) > > at java.net.URLClassLoader.findClass(URLClassLoader.java:361) > > at java.lang.ClassLoader.loadClass(ClassLoader.java:424) > > at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335) > > at java.lang.ClassLoader.loadClass(ClassLoader.java:357) > > at > org.apache.hadoop.yarn.client.api.TimelineClient.createTimelineClient(TimelineClient.java:55) > > at > org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.createTimelineClient(YarnClientImpl.java:181) > > at > org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.serviceInit(YarnClientImpl.java:168) > > at org.apache.hadoop.service.AbstractService.init(AbstractService.java:163) > > at > org.apache.flink.yarn.cli.FlinkYarnSessionCli.getClusterDescriptor(FlinkYarnSessionCli.java:966) > > at > org.apache.flink.yarn.cli.FlinkYarnSessionCli.createDescriptor(FlinkYarnSessionCli.java:269) > > at > org.apache.flink.yarn.cli.FlinkYarnSessionCli.createClusterDescriptor(FlinkYarnSessionCli.java:444) > > at > org.apache.flink.yarn.cli.FlinkYarnSessionCli.createClusterDescriptor(FlinkYarnSessionCli.java:92) > > at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:221) > > at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) > > at > org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) > > at > org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) > > at java.security.AccessController.doPrivileged(Native Method) > > at javax.security.auth.Subject.doAs(Subject.java:422) > > at > org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1754) > > at > org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) > > at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) > > Caused by: java.lang.ClassNotFoundException: > com.sun.jersey.core.util.FeaturesAndProperties > > at java.net.URLClassLoader.findClass(URLClassLoader.java:381) > > at java.lang.ClassLoader.loadClass(ClassLoader.java:424) > > at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335) > > at java.lang.ClassLoader.loadClass(ClassLoader.java:357) > > > Command : HADOOP_CONF_DIR=/etc/hadoop/conf bin/flink run -m yarn-cluster > -yd -yn 2 -ys 20 -yjm 10240 -ytm 10240 -yst -ynm test -yqu default -p 20 > test.jar > > The class *com/sun/jersey/core/util/FeaturesAndProperties* is already > present in the test.jar so not sure why am I getting this exception. > > Please check and let me know. > > Thanks, > -- > > Garvit Sharma > github.com/garvitlnmiit/ > > No Body is a Scholar by birth, its only hard work and strong determination > that makes him master. > -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Exception while submitting jobs through Yarn
Hi, I am using *flink-1.5.0-bin-hadoop27-scala_2.11 *to submit jobs through Yarn, but I am getting the below exception : java.lang.NoClassDefFoundError: com/sun/jersey/core/util/FeaturesAndProperties at java.lang.ClassLoader.defineClass1(Native Method) at java.lang.ClassLoader.defineClass(ClassLoader.java:763) at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142) at java.net.URLClassLoader.defineClass(URLClassLoader.java:467) at java.net.URLClassLoader.access$100(URLClassLoader.java:73) at java.net.URLClassLoader$1.run(URLClassLoader.java:368) at java.net.URLClassLoader$1.run(URLClassLoader.java:362) at java.security.AccessController.doPrivileged(Native Method) at java.net.URLClassLoader.findClass(URLClassLoader.java:361) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) at org.apache.hadoop.yarn.client.api.TimelineClient.createTimelineClient(TimelineClient.java:55) at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.createTimelineClient(YarnClientImpl.java:181) at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.serviceInit(YarnClientImpl.java:168) at org.apache.hadoop.service.AbstractService.init(AbstractService.java:163) at org.apache.flink.yarn.cli.FlinkYarnSessionCli.getClusterDescriptor(FlinkYarnSessionCli.java:966) at org.apache.flink.yarn.cli.FlinkYarnSessionCli.createDescriptor(FlinkYarnSessionCli.java:269) at org.apache.flink.yarn.cli.FlinkYarnSessionCli.createClusterDescriptor(FlinkYarnSessionCli.java:444) at org.apache.flink.yarn.cli.FlinkYarnSessionCli.createClusterDescriptor(FlinkYarnSessionCli.java:92) at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:221) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020) at org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1754) at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096) Caused by: java.lang.ClassNotFoundException: com.sun.jersey.core.util.FeaturesAndProperties at java.net.URLClassLoader.findClass(URLClassLoader.java:381) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) Command : HADOOP_CONF_DIR=/etc/hadoop/conf bin/flink run -m yarn-cluster -yd -yn 2 -ys 20 -yjm 10240 -ytm 10240 -yst -ynm test -yqu default -p 20 test.jar The class *com/sun/jersey/core/util/FeaturesAndProperties* is already present in the test.jar so not sure why am I getting this exception. Please check and let me know. Thanks, -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Re: Any remote opportunity to work on Flink project?
If that's the case, then I too would be interested to build a product around Flink :). Please let me know. On Wed, Jun 6, 2018 at 11:21 AM, Christophe Salperwyck < christophe.salperw...@gmail.com> wrote: > Still some people are interested to pay people to build a product around > Flink :-) > > Interested too about Flink and online ML! > > Cheers, > Christophe > > On Wed, 6 Jun 2018 at 07:40, Garvit Sharma wrote: > >> Flink is OpenSource!! >> >> On Wed, Jun 6, 2018 at 10:45 AM, Deepak Sharma >> wrote: >> >>> Hi Flink Users, >>> Sorry to spam your inbox and GM all. >>> I am looking for opportunity to work on Flink project , specifically if >>> its Flink ML over streaming >>> Please do let me know if anyone is looking for freelancers around any of >>> their Flink projects. >>> >>> -- >>> Thanks >>> Deepak >>> >> >> >> >> -- >> >> Garvit Sharma >> github.com/garvitlnmiit/ >> >> No Body is a Scholar by birth, its only hard work and strong >> determination that makes him master. >> > -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Re: Any remote opportunity to work on Flink project?
Flink is OpenSource!! On Wed, Jun 6, 2018 at 10:45 AM, Deepak Sharma wrote: > Hi Flink Users, > Sorry to spam your inbox and GM all. > I am looking for opportunity to work on Flink project , specifically if > its Flink ML over streaming > Please do let me know if anyone is looking for freelancers around any of > their Flink projects. > > -- > Thanks > Deepak > -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Re: Implementing a “join” between a DataStream and a “set of rules”
Hi, For the above use case, you should do the following : 1. Convert your DataStream into KeyedDataStream by defining a key which would be used to get validated against your rules. 2. Same as 1 for rules stream. 3. Join the two keyedStreams using Flink's connect operator. 4. Store the rules into Flink's internal state i,e. Flink's managed keyed state. 5. Validate the data coming in the dataStream against the managed keyed state. Refer to [1] [2] for more details. [1] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/ operators/ [2] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/state.html On Tue, Jun 5, 2018 at 5:52 PM, Sandybayev, Turar (CAI - Atlanta) < turar.sandyba...@coxautoinc.com> wrote: > Hi, > > > > What is the best practice recommendation for the following use case? We > need to match a stream against a set of “rules”, which are essentially a > Flink DataSet concept. Updates to this “rules set" are possible but not > frequent. Each stream event must be checked against all the records in > “rules set”, and each match produces one or more events into a sink. Number > of records in a rule set are in the 6 digit range. > > > > Currently we're simply loading rules into a local List of rules and using > flatMap over an incoming DataStream. Inside flatMap, we're just iterating > over a list comparing each event to each rule. > > > > To speed up the iteration, we can also split the list into several > batches, essentially creating a list of lists, and creating a separate > thread to iterate over each sub-list (using Futures in either Java or > Scala). > > > > Questions: > > 1.Is there a better way to do this kind of a join? > > 2.If not, is it safe to add additional parallelism by creating > new threads inside each flatMap operation, on top of what Flink is already > doing? > > > > Thanks in advance! > > Turar > > > -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Re: Checkpointing on cluster shutdown
But job should be terminated gracefully. Why is this behavior not there? On Tue, Jun 5, 2018 at 4:19 PM, Chesnay Schepler wrote: > No checkpoint will be triggered when the cluster is shutdown. For this > case you will have to manually trigger a savepoint. > > If a TM goes down it does not create a checkpoint. IN these cases the job > will be restarted from the last successful checkpoint. > > > On 05.06.2018 12:01, Data Engineer wrote: > >> Hi, >> >> Suppose I have a working Flink cluster with 1 taskmanager and 1 >> jobmanager and I have enabled checkpointing with say an interval of 1 >> minute. >> Now if I shut down the Flink cluster in between checkpoints (say for some >> upgrade), will the JobManager automatically trigger a checkpoint before >> going down? >> >> Or is it mandatory to manually trigger savepoints in these cases? >> Also am I correct in my understanding that if a taskmanager goes down >> first, there is no way the TaskManager can trigger the checkpoint on its >> own? >> >> >> > -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Re: ConcurrentModificationException while accessing managed keyed state
Hi,
Sorry guys for the delay. I was trying to reproduce the complete error on
my local machine but could not get it though. I will try again with actual
traffic and let you the exception Stacktrace.
For now, I have the following details available to me.
Flink version: 1.4.1
State backend: Heap
I am not creating a custom thread to operate the state.
Code snippet :
https://gist.github.com/garvitlnmiit/10db9d4b6eb41135332fba13d908e36c
I will reply back with the Stacktrace soon.
Thanks,
On Sat, Jun 2, 2018 at 6:49 PM, aitozi wrote:
> Hi Garvit Sharma,
>
> Flink run with per parallel with a single thread. Can you show a little
> code
> about how you use the keyed state in processFunction
>
> Garvit Sharma wrote
> > Hi,
> >
> > I have a use case where I am keeping the keyed state in ProcessFunction.
> >
> > Key: Integer personId;
> >
> > /**
> > * The data type stored in the state
> > */public class PersonDetails {
> > public long count;
> > public long lastModified;}
> >
> >
> > I have encountered a lot of ConcurrentModificationException.
> >
> > I thought Flink processes all the operators on a keyed stream in a single
> > thread. It seems like operators being accessed through multiple threads.
> >
> > If I get such exception then the data coming from Kafka would be consumed
> > without making an update to the internal state. Making me lose the data.
> >
> > Please help me in handling the case according to my use case.
> >
> > Thanks,
> >
> > --
> >
> > Garvit Sharma
> > github.com/garvitlnmiit/
> >
> > No Body is a Scholar by birth, its only hard work and strong
> determination
> > that makes him master.
>
>
>
>
>
> --
> Sent from: http://apache-flink-user-mailing-list-archive.2336050.
> n4.nabble.com/
>
--
Garvit Sharma
github.com/garvitlnmiit/
No Body is a Scholar by birth, its only hard work and strong determination
that makes him master.
ConcurrentModificationException while accessing managed keyed state
Hi,
I have a use case where I am keeping the keyed state in ProcessFunction.
Key: Integer personId;
/**
* The data type stored in the state
*/public class PersonDetails {
public long count;
public long lastModified;}
I have encountered a lot of ConcurrentModificationException.
I thought Flink processes all the operators on a keyed stream in a single
thread. It seems like operators being accessed through multiple threads.
If I get such exception then the data coming from Kafka would be consumed
without making an update to the internal state. Making me lose the data.
Please help me in handling the case according to my use case.
Thanks,
--
Garvit Sharma
github.com/garvitlnmiit/
No Body is a Scholar by birth, its only hard work and strong determination
that makes him master.
Re: Regarding Keyed State of Connected Stream
Hi Sihua, Thanks for the quick response. Could you please let me know, where can I find more details about it. Thanks, On Fri, May 25, 2018 at 11:04 AM, sihua zhou wrote: > Hi Garvit, > > I think you don't need to lock it, they are executed in the same thread > sync. > > Best, Sihua > > > > > > > 在2018年05月25日 10:26，Garvit Sharma 写道： > Hi, > > Let's consider, I have two keyed streams one for rules and another for > data and I have created a connected stream. > > I am maintaining a managed keyed state (ValueState), rules stream will > keep updating the state and data stream will keep reading from it. > > Do I need to take a lock explicitly while updating the state or not? > > I need to understand. > > Please let me know. > > Thanks, > > -- > > Garvit Sharma > <http://github.com/garvitlnmiit/>github.com/garvitlnmiit/ > > No Body is a Scholar by birth, its only hard work and strong determination > that makes him master. > -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
Regarding Keyed State of Connected Stream
Hi, Let's consider, I have two keyed streams one for rules and another for data and I have created a connected stream. I am maintaining a managed keyed state (ValueState), rules stream will keep updating the state and data stream will keep reading from it. Do I need to take a lock explicitly while updating the state or not? I need to understand. Please let me know. Thanks, -- Garvit Sharma github.com/garvitlnmiit/ No Body is a Scholar by birth, its only hard work and strong determination that makes him master.
