Re: Kinesis Operator Help
Why use the application id ? Could you generate and use a java.util.UUID for example and save it in HDFS ? Ram On Tue, Feb 16, 2016 at 11:40 AM, Jim wrote: > Good morning, > > > > I am new to Apex, Hadoop and Yarn (nothing like tackling something new, is > there?). > > > > I have my first Apex apps working that are edi processors that read new > edi transactions from an Amazon Kinesis stream, look at the data, and > routes the edi data to an appropriate handler for processing (note the > operatorEs pushes the data to ElasticSearch for logging). Here is a > diagram: > > > > > > Everything launches, and is working fine with the above diagram from the > edi router through the transaction operators. > > > > The final challenge I am having, being new to all of this, is that the > Kinesis operator, by default, stores it’s app id in into > IdempotentStorageManager (aka WindowDataManager) when it is launched, so if > the app it shutdown and restarted this same app id is used by default with > the checkpoint so you don’t reprocess the same records again when the > application is restarted. > > > > You can see this id immediately to the right of the Operations / apps in > gray lettering ‘application_1453741656046_0520’ in the image from the > datatorrent console below: > > > > [image: cid:image004.png@01D168BA.5FE56550] > > > > However, if you kill the application, and re-launch, this id changes, and > it starts reading from the Kinesis stream back from the beginning; and the > only way to restart it so it starts where it left off is using the cli as > follows: > > > > 1.)Run ‘dtcli’ from the command line. > > 2.)Run ‘launch -originalAppId “application_1453741656046_0520” to .apa file>’ > > > > This will launch the application using the same app id identified in the > console screen above. > > > > I want to make this easier, but need some experts help in tweaking this so > it works. > > > > I am thinking that there should be a way with Kinesis to: > > > > 1.)Define in the properties, a Kinesis app id string value. > > 2.)If this value is defined, it will use that, when launching the > application, to check if an Hadoop app id has already been assigned to that > identifier. > > 3.)If that value is not yet stored in the database, it will launch > the app, creating a new app id, and store the app id under the identifier > key value. > > 4.)Now if I kill the app, or install new software, it will always > pick up where it left off by using the identifier key value to retrieve and > assign the app id. > > > > Sounds simple, right? J > > > > Can one of the experts out there help me figure this out as I don’t want > to reprocess already processed edi transactions? > > > > Thanks, > > > > Jim > > > > > > > > > > Jim > > > jim@facility.supplies (414) 760-7711 > -- > The information contained in this communication, including any files or > attachments transmitted with it, may contain copyrighted information or > information that is confidential and exempt from disclosure under > applicable laws and regulations, is intended only for the use of the > recipient(s) named above, and may be legally privileged. If the reader of > this message is not the intended recipient, you are hereby notified that > any dissemination, distribution, or copying of this communication, or any > of its contents, files or attachments, is strictly prohibited. If you have > received this communication in error, please return it to the sender > immediately and delete the original message and any copy of it from your > computer system. If you have any questions concerning this message, please > contact the sender. >
RE: Kinesis Operator Help
Ram, I am not 100% fluent in the details of the base kinesis operator and how it interacts with Hadoop (hence my posting); if it would support that, then yes, you could. My goal is to make it so one can easily pick up where they left off reading the Kinesis stream, regardless of if you kill the application and re-launch it, etc., without needing to go out to the cli to do some commands (because at some point some operator will forget and then we will reprocess a bunch of transactions; that would not be good! Jim From: Munagala Ramanath [mailto:r...@datatorrent.com] Sent: Tuesday, February 16, 2016 5:21 PM To: users@apex.incubator.apache.org Subject: Re: Kinesis Operator Help Why use the application id ? Could you generate and use a java.util.UUID for example and save it in HDFS ? Ram On Tue, Feb 16, 2016 at 11:40 AM, Jim mailto:jim@facility.supplies>> wrote: Good morning, I am new to Apex, Hadoop and Yarn (nothing like tackling something new, is there?). I have my first Apex apps working that are edi processors that read new edi transactions from an Amazon Kinesis stream, look at the data, and routes the edi data to an appropriate handler for processing (note the operatorEs pushes the data to ElasticSearch for logging). Here is a diagram: [cid:image001.jpg@01D168E3.D76CC630] Everything launches, and is working fine with the above diagram from the edi router through the transaction operators. The final challenge I am having, being new to all of this, is that the Kinesis operator, by default, stores it’s app id in into IdempotentStorageManager (aka WindowDataManager) when it is launched, so if the app it shutdown and restarted this same app id is used by default with the checkpoint so you don’t reprocess the same records again when the application is restarted. You can see this id immediately to the right of the Operations / apps in gray lettering ‘application_1453741656046_0520’ in the image from the datatorrent console below: [cid:image004.png@01D168BA.5FE56550] However, if you kill the application, and re-launch, this id changes, and it starts reading from the Kinesis stream back from the beginning; and the only way to restart it so it starts where it left off is using the cli as follows: 1.)Run ‘dtcli’ from the command line. 2.)Run ‘launch -originalAppId “application_1453741656046_0520” ’ This will launch the application using the same app id identified in the console screen above. I want to make this easier, but need some experts help in tweaking this so it works. I am thinking that there should be a way with Kinesis to: 1.)Define in the properties, a Kinesis app id string value. 2.)If this value is defined, it will use that, when launching the application, to check if an Hadoop app id has already been assigned to that identifier. 3.)If that value is not yet stored in the database, it will launch the app, creating a new app id, and store the app id under the identifier key value. 4.)Now if I kill the app, or install new software, it will always pick up where it left off by using the identifier key value to retrieve and assign the app id. Sounds simple, right? ☺ Can one of the experts out there help me figure this out as I don’t want to reprocess already processed edi transactions? Thanks, Jim Jim jim@facility.supplies<mailto:jim@facility.supplies> (414) 760-7711 The information contained in this communication, including any files or attachments transmitted with it, may contain copyrighted information or information that is confidential and exempt from disclosure under applicable laws and regulations, is intended only for the use of the recipient(s) named above, and may be legally privileged. If the reader of this message is not the intended recipient, you are hereby notified that any dissemination, distribution, or copying of this communication, or any of its contents, files or attachments, is strictly prohibited. If you have received this communication in error, please return it to the sender immediately and delete the original message and any copy of it from your computer system. If you have any questions concerning this message, please contact the sender.
Re: Kinesis Operator Help
Ah, I understand now. The path is set in IdempotentStorageManager.FSIdempotentStorageManager,setup() near line 146: appPath = new Path(context.getValue(DAG.APPLICATION_PATH) + Path.SEPARATOR + recoveryPath); You can try creating a new class that extends FSIdempotentStorageManager and override setup() to use a local property for the appPath and simply duplicate the rest of the code. Ram On Tue, Feb 16, 2016 at 3:59 PM, Jim wrote: > Ram, > > > > I am not 100% fluent in the details of the base kinesis operator and how > it interacts with Hadoop (hence my posting); if it would support that, then > yes, you could. > > > > My goal is to make it so one can easily pick up where they left off > reading the Kinesis stream, regardless of if you kill the application and > re-launch it, etc., without needing to go out to the cli to do some > commands (because at some point some operator will forget and then we will > reprocess a bunch of transactions; that would not be good! > > > > Jim > > > > *From:* Munagala Ramanath [mailto:r...@datatorrent.com] > *Sent:* Tuesday, February 16, 2016 5:21 PM > *To:* users@apex.incubator.apache.org > *Subject:* Re: Kinesis Operator Help > > > > Why use the application id ? Could you generate and use a java.util.UUID > for example and save it in HDFS ? > > > > Ram > > > > On Tue, Feb 16, 2016 at 11:40 AM, Jim wrote: > > Good morning, > > > > I am new to Apex, Hadoop and Yarn (nothing like tackling something new, is > there?). > > > > I have my first Apex apps working that are edi processors that read new > edi transactions from an Amazon Kinesis stream, look at the data, and > routes the edi data to an appropriate handler for processing (note the > operatorEs pushes the data to ElasticSearch for logging). Here is a > diagram: > > > > > > Everything launches, and is working fine with the above diagram from the > edi router through the transaction operators. > > > > The final challenge I am having, being new to all of this, is that the > Kinesis operator, by default, stores it’s app id in into > IdempotentStorageManager (aka WindowDataManager) when it is launched, so if > the app it shutdown and restarted this same app id is used by default with > the checkpoint so you don’t reprocess the same records again when the > application is restarted. > > > > You can see this id immediately to the right of the Operations / apps in > gray lettering ‘application_1453741656046_0520’ in the image from the > datatorrent console below: > > > > [image: cid:image004.png@01D168BA.5FE56550] > > > > However, if you kill the application, and re-launch, this id changes, and > it starts reading from the Kinesis stream back from the beginning; and the > only way to restart it so it starts where it left off is using the cli as > follows: > > > > 1.)Run ‘dtcli’ from the command line. > > 2.)Run ‘launch -originalAppId “application_1453741656046_0520” to .apa file>’ > > > > This will launch the application using the same app id identified in the > console screen above. > > > > I want to make this easier, but need some experts help in tweaking this so > it works. > > > > I am thinking that there should be a way with Kinesis to: > > > > 1.)Define in the properties, a Kinesis app id string value. > > 2.)If this value is defined, it will use that, when launching the > application, to check if an Hadoop app id has already been assigned to that > identifier. > > 3.)If that value is not yet stored in the database, it will launch > the app, creating a new app id, and store the app id under the identifier > key value. > > 4.)Now if I kill the app, or install new software, it will always > pick up where it left off by using the identifier key value to retrieve and > assign the app id. > > > > Sounds simple, right? J > > > > Can one of the experts out there help me figure this out as I don’t want > to reprocess already processed edi transactions? > > > > Thanks, > > > > Jim > > > > > > > > > > > Jim > > > jim@facility.supplies (414) 760-7711 > -- > > *The information contained in this communication, including any files or > attachments transmitted with it, may contain copyrighted information or > information that is confidential and exempt from disclosure under > applicable laws and regulations, is intended only for the use of the > recipient(s) named above, and may be legally privileged. If the reader of > this message is not the intended recipient, you are hereby notified that > any dissemination, distribution, or copying of this communication, or any > of its contents, files or attachments, is strictly prohibited. If you have > received this communication in error, please return it to the sender > immediately and delete the original message and any copy of it from your > computer system. If you have any questions concerning this message, please > contact the sender. * > > >
Re: Kinesis Operator Help
Ram, The recovery path, when under the application directory, will be automatically copied to the new app directory when relaunch option is used. This is how the previous instance data is available to the new app. Thomas On Tue, Feb 16, 2016 at 5:23 PM, Munagala Ramanath wrote: > Ah, I understand now. > > The path is set in > IdempotentStorageManager.FSIdempotentStorageManager,setup() near line 146: > appPath = new Path(context.getValue(DAG.APPLICATION_PATH) + Path.SEPARATOR > + recoveryPath); > > You can try creating a new class that extends FSIdempotentStorageManager > and override setup() to use a local property > for the appPath and simply duplicate the rest of the code. > > Ram > > On Tue, Feb 16, 2016 at 3:59 PM, Jim wrote: > >> Ram, >> >> >> >> I am not 100% fluent in the details of the base kinesis operator and how >> it interacts with Hadoop (hence my posting); if it would support that, then >> yes, you could. >> >> >> >> My goal is to make it so one can easily pick up where they left off >> reading the Kinesis stream, regardless of if you kill the application and >> re-launch it, etc., without needing to go out to the cli to do some >> commands (because at some point some operator will forget and then we will >> reprocess a bunch of transactions; that would not be good! >> >> >> >> Jim >> >> >> >> *From:* Munagala Ramanath [mailto:r...@datatorrent.com] >> *Sent:* Tuesday, February 16, 2016 5:21 PM >> *To:* users@apex.incubator.apache.org >> *Subject:* Re: Kinesis Operator Help >> >> >> >> Why use the application id ? Could you generate and use a java.util.UUID >> for example and save it in HDFS ? >> >> >> >> Ram >> >> >> >> On Tue, Feb 16, 2016 at 11:40 AM, Jim wrote: >> >> Good morning, >> >> >> >> I am new to Apex, Hadoop and Yarn (nothing like tackling something new, >> is there?). >> >> >> >> I have my first Apex apps working that are edi processors that read new >> edi transactions from an Amazon Kinesis stream, look at the data, and >> routes the edi data to an appropriate handler for processing (note the >> operatorEs pushes the data to ElasticSearch for logging). Here is a >> diagram: >> >> >> >> >> >> Everything launches, and is working fine with the above diagram from the >> edi router through the transaction operators. >> >> >> >> The final challenge I am having, being new to all of this, is that the >> Kinesis operator, by default, stores it’s app id in into >> IdempotentStorageManager (aka WindowDataManager) when it is launched, so if >> the app it shutdown and restarted this same app id is used by default with >> the checkpoint so you don’t reprocess the same records again when the >> application is restarted. >> >> >> >> You can see this id immediately to the right of the Operations / apps in >> gray lettering ‘application_1453741656046_0520’ in the image from the >> datatorrent console below: >> >> >> >> [image: cid:image004.png@01D168BA.5FE56550] >> >> >> >> However, if you kill the application, and re-launch, this id changes, and >> it starts reading from the Kinesis stream back from the beginning; and the >> only way to restart it so it starts where it left off is using the cli as >> follows: >> >> >> >> 1.)Run ‘dtcli’ from the command line. >> >> 2.)Run ‘launch -originalAppId “application_1453741656046_0520” > to .apa file>’ >> >> >> >> This will launch the application using the same app id identified in the >> console screen above. >> >> >> >> I want to make this easier, but need some experts help in tweaking this >> so it works. >> >> >> >> I am thinking that there should be a way with Kinesis to: >> >> >> >> 1.)Define in the properties, a Kinesis app id string value. >> >> 2.)If this value is defined, it will use that, when launching the >> application, to check if an Hadoop app id has already been assigned to that >> identifier. >> >> 3.)If that value is not yet stored in the database, it will launch >> the app, creating a new app id, and store the app id under the identifier >> key value. >> >> 4.)Now if I kill the app, or install new software, it will always >> pick up where it left off by using the identifier key value to retrieve and >> assign the app id. >&g
Re: Kinesis Operator Help
By "relaunch option" I'm assuming you mean "launch -originalAppId ...". Looks like Jim does not want to use that option. He wants a new launch to automatically detect data from an earlier launch and, if present, use it. Ram On Tue, Feb 16, 2016 at 8:29 PM, Thomas Weise wrote: > Ram, > > The recovery path, when under the application directory, will be > automatically copied to the new app directory when relaunch option is used. > This is how the previous instance data is available to the new app. > > Thomas > > On Tue, Feb 16, 2016 at 5:23 PM, Munagala Ramanath > wrote: > >> Ah, I understand now. >> >> The path is set in >> IdempotentStorageManager.FSIdempotentStorageManager,setup() near line 146: >> appPath = new Path(context.getValue(DAG.APPLICATION_PATH) + >> Path.SEPARATOR + recoveryPath); >> >> You can try creating a new class that extends FSIdempotentStorageManager >> and override setup() to use a local property >> for the appPath and simply duplicate the rest of the code. >> >> Ram >> >> On Tue, Feb 16, 2016 at 3:59 PM, Jim wrote: >> >>> Ram, >>> >>> >>> >>> I am not 100% fluent in the details of the base kinesis operator and how >>> it interacts with Hadoop (hence my posting); if it would support that, then >>> yes, you could. >>> >>> >>> >>> My goal is to make it so one can easily pick up where they left off >>> reading the Kinesis stream, regardless of if you kill the application and >>> re-launch it, etc., without needing to go out to the cli to do some >>> commands (because at some point some operator will forget and then we will >>> reprocess a bunch of transactions; that would not be good! >>> >>> >>> >>> Jim >>> >>> >>> >>> *From:* Munagala Ramanath [mailto:r...@datatorrent.com] >>> *Sent:* Tuesday, February 16, 2016 5:21 PM >>> *To:* users@apex.incubator.apache.org >>> *Subject:* Re: Kinesis Operator Help >>> >>> >>> >>> Why use the application id ? Could you generate and use a java.util.UUID >>> for example and save it in HDFS ? >>> >>> >>> >>> Ram >>> >>> >>> >>> On Tue, Feb 16, 2016 at 11:40 AM, Jim wrote: >>> >>> Good morning, >>> >>> >>> >>> I am new to Apex, Hadoop and Yarn (nothing like tackling something new, >>> is there?). >>> >>> >>> >>> I have my first Apex apps working that are edi processors that read new >>> edi transactions from an Amazon Kinesis stream, look at the data, and >>> routes the edi data to an appropriate handler for processing (note the >>> operatorEs pushes the data to ElasticSearch for logging). Here is a >>> diagram: >>> >>> >>> >>> >>> >>> Everything launches, and is working fine with the above diagram from the >>> edi router through the transaction operators. >>> >>> >>> >>> The final challenge I am having, being new to all of this, is that the >>> Kinesis operator, by default, stores it’s app id in into >>> IdempotentStorageManager (aka WindowDataManager) when it is launched, so if >>> the app it shutdown and restarted this same app id is used by default with >>> the checkpoint so you don’t reprocess the same records again when the >>> application is restarted. >>> >>> >>> >>> You can see this id immediately to the right of the Operations / apps in >>> gray lettering ‘application_1453741656046_0520’ in the image from the >>> datatorrent console below: >>> >>> >>> >>> [image: cid:image004.png@01D168BA.5FE56550] >>> >>> >>> >>> However, if you kill the application, and re-launch, this id changes, >>> and it starts reading from the Kinesis stream back from the beginning; and >>> the only way to restart it so it starts where it left off is using the cli >>> as follows: >>> >>> >>> >>> 1.)Run ‘dtcli’ from the command line. >>> >>> 2.)Run ‘launch -originalAppId “application_1453741656046_0520” >>> ’ >>> >>> >>> >>> This will launch the application using the same app id identified in the >>> console screen above. >>> >>> >>> >>> I want to make thi
Re: Kinesis Operator Help
Hi Jim, Take a look at OffsetManager interface implemented for Kafka operator. There is something similar in kinesis (I guess its ShardManager), which you can override and provide your method to keep offset information at a fixed location. The newly launched application calls ShardManager loadInitialOffsets to resume consumption from last saved state. Let me know, if this is something you are looking for. -Tushar. On Wed, Feb 17, 2016 at 11:52 AM, Munagala Ramanath wrote: > By "relaunch option" I'm assuming you mean "launch -originalAppId ...". > Looks like Jim does not want to use that option. > He wants a new launch to automatically detect data from an earlier launch > and, if present, use it. > > Ram > > On Tue, Feb 16, 2016 at 8:29 PM, Thomas Weise > wrote: > >> Ram, >> >> The recovery path, when under the application directory, will be >> automatically copied to the new app directory when relaunch option is used. >> This is how the previous instance data is available to the new app. >> >> Thomas >> >> On Tue, Feb 16, 2016 at 5:23 PM, Munagala Ramanath >> wrote: >> >>> Ah, I understand now. >>> >>> The path is set in >>> IdempotentStorageManager.FSIdempotentStorageManager,setup() near line 146: >>> appPath = new Path(context.getValue(DAG.APPLICATION_PATH) + >>> Path.SEPARATOR + recoveryPath); >>> >>> You can try creating a new class that extends FSIdempotentStorageManager >>> and override setup() to use a local property >>> for the appPath and simply duplicate the rest of the code. >>> >>> Ram >>> >>> On Tue, Feb 16, 2016 at 3:59 PM, Jim wrote: >>> >>>> Ram, >>>> >>>> >>>> >>>> I am not 100% fluent in the details of the base kinesis operator and >>>> how it interacts with Hadoop (hence my posting); if it would support that, >>>> then yes, you could. >>>> >>>> >>>> >>>> My goal is to make it so one can easily pick up where they left off >>>> reading the Kinesis stream, regardless of if you kill the application and >>>> re-launch it, etc., without needing to go out to the cli to do some >>>> commands (because at some point some operator will forget and then we will >>>> reprocess a bunch of transactions; that would not be good! >>>> >>>> >>>> >>>> Jim >>>> >>>> >>>> >>>> *From:* Munagala Ramanath [mailto:r...@datatorrent.com] >>>> *Sent:* Tuesday, February 16, 2016 5:21 PM >>>> *To:* users@apex.incubator.apache.org >>>> *Subject:* Re: Kinesis Operator Help >>>> >>>> >>>> >>>> Why use the application id ? Could you generate and use a >>>> java.util.UUID for example and save it in HDFS ? >>>> >>>> >>>> >>>> Ram >>>> >>>> >>>> >>>> On Tue, Feb 16, 2016 at 11:40 AM, Jim wrote: >>>> >>>> Good morning, >>>> >>>> >>>> >>>> I am new to Apex, Hadoop and Yarn (nothing like tackling something new, >>>> is there?). >>>> >>>> >>>> >>>> I have my first Apex apps working that are edi processors that read new >>>> edi transactions from an Amazon Kinesis stream, look at the data, and >>>> routes the edi data to an appropriate handler for processing (note the >>>> operatorEs pushes the data to ElasticSearch for logging). Here is a >>>> diagram: >>>> >>>> >>>> >>>> >>>> >>>> Everything launches, and is working fine with the above diagram from >>>> the edi router through the transaction operators. >>>> >>>> >>>> >>>> The final challenge I am having, being new to all of this, is that the >>>> Kinesis operator, by default, stores it’s app id in into >>>> IdempotentStorageManager (aka WindowDataManager) when it is launched, so if >>>> the app it shutdown and restarted this same app id is used by default with >>>> the checkpoint so you don’t reprocess the same records again when the >>>> application is restarted. >>>> >>>> >>>> >>>> You can see this id immediately to the right of the Operations / apps >>>> in gray lettering ‘application_1453741656046_0520’ in the image from the >
Re: Kinesis Operator Help
Jim, You can put the following line in a shell script, dtcli - e "launch -originalAppId apppid apa_path" Even you can shutdown the app from the shell script. On Tue, Feb 16, 2016, 10:22 PM Munagala Ramanath wrote: > By "relaunch option" I'm assuming you mean "launch -originalAppId ...". > Looks like Jim does not want to use that option. > He wants a new launch to automatically detect data from an earlier launch > and, if present, use it. > > Ram > > On Tue, Feb 16, 2016 at 8:29 PM, Thomas Weise > wrote: > >> Ram, >> >> The recovery path, when under the application directory, will be >> automatically copied to the new app directory when relaunch option is used. >> This is how the previous instance data is available to the new app. >> >> Thomas >> >> On Tue, Feb 16, 2016 at 5:23 PM, Munagala Ramanath >> wrote: >> >>> Ah, I understand now. >>> >>> The path is set in >>> IdempotentStorageManager.FSIdempotentStorageManager,setup() near line 146: >>> appPath = new Path(context.getValue(DAG.APPLICATION_PATH) + >>> Path.SEPARATOR + recoveryPath); >>> >>> You can try creating a new class that extends FSIdempotentStorageManager >>> and override setup() to use a local property >>> for the appPath and simply duplicate the rest of the code. >>> >>> Ram >>> >>> On Tue, Feb 16, 2016 at 3:59 PM, Jim wrote: >>> >>>> Ram, >>>> >>>> >>>> >>>> I am not 100% fluent in the details of the base kinesis operator and >>>> how it interacts with Hadoop (hence my posting); if it would support that, >>>> then yes, you could. >>>> >>>> >>>> >>>> My goal is to make it so one can easily pick up where they left off >>>> reading the Kinesis stream, regardless of if you kill the application and >>>> re-launch it, etc., without needing to go out to the cli to do some >>>> commands (because at some point some operator will forget and then we will >>>> reprocess a bunch of transactions; that would not be good! >>>> >>>> >>>> >>>> Jim >>>> >>>> >>>> >>>> *From:* Munagala Ramanath [mailto:r...@datatorrent.com] >>>> *Sent:* Tuesday, February 16, 2016 5:21 PM >>>> *To:* users@apex.incubator.apache.org >>>> *Subject:* Re: Kinesis Operator Help >>>> >>>> >>>> >>>> Why use the application id ? Could you generate and use a >>>> java.util.UUID for example and save it in HDFS ? >>>> >>>> >>>> >>>> Ram >>>> >>>> >>>> >>>> On Tue, Feb 16, 2016 at 11:40 AM, Jim wrote: >>>> >>>> Good morning, >>>> >>>> >>>> >>>> I am new to Apex, Hadoop and Yarn (nothing like tackling something new, >>>> is there?). >>>> >>>> >>>> >>>> I have my first Apex apps working that are edi processors that read new >>>> edi transactions from an Amazon Kinesis stream, look at the data, and >>>> routes the edi data to an appropriate handler for processing (note the >>>> operatorEs pushes the data to ElasticSearch for logging). Here is a >>>> diagram: >>>> >>>> >>>> >>>> >>>> >>>> Everything launches, and is working fine with the above diagram from >>>> the edi router through the transaction operators. >>>> >>>> >>>> >>>> The final challenge I am having, being new to all of this, is that the >>>> Kinesis operator, by default, stores it’s app id in into >>>> IdempotentStorageManager (aka WindowDataManager) when it is launched, so if >>>> the app it shutdown and restarted this same app id is used by default with >>>> the checkpoint so you don’t reprocess the same records again when the >>>> application is restarted. >>>> >>>> >>>> >>>> You can see this id immediately to the right of the Operations / apps >>>> in gray lettering ‘application_1453741656046_0520’ in the image from the >>>> datatorrent console below: >>>> >>>> >>>> >>>> [image: cid:image004.png@01D168BA.5FE56550] >>>> >>>> >>>> >>>> However, if you kill the
RE: Kinesis Operator Help
Yes, Ram was right, I do not want a command line fix, because at some point, an operator is going to launch one without doing that, and the system will be messed up. Thanks, Jim From: Sandesh Hegde [mailto:sand...@datatorrent.com] Sent: Wednesday, February 17, 2016 9:05 AM To: users@apex.incubator.apache.org Subject: Re: Kinesis Operator Help Jim, You can put the following line in a shell script, dtcli - e "launch -originalAppId apppid apa_path" Even you can shutdown the app from the shell script. On Tue, Feb 16, 2016, 10:22 PM Munagala Ramanath mailto:r...@datatorrent.com>> wrote: By "relaunch option" I'm assuming you mean "launch -originalAppId ...". Looks like Jim does not want to use that option. He wants a new launch to automatically detect data from an earlier launch and, if present, use it. Ram On Tue, Feb 16, 2016 at 8:29 PM, Thomas Weise mailto:thomas.we...@gmail.com>> wrote: Ram, The recovery path, when under the application directory, will be automatically copied to the new app directory when relaunch option is used. This is how the previous instance data is available to the new app. Thomas On Tue, Feb 16, 2016 at 5:23 PM, Munagala Ramanath mailto:r...@datatorrent.com>> wrote: Ah, I understand now. The path is set in IdempotentStorageManager.FSIdempotentStorageManager,setup() near line 146: appPath = new Path(context.getValue(DAG.APPLICATION_PATH) + Path.SEPARATOR + recoveryPath); You can try creating a new class that extends FSIdempotentStorageManager and override setup() to use a local property for the appPath and simply duplicate the rest of the code. Ram On Tue, Feb 16, 2016 at 3:59 PM, Jim mailto:jim@facility.supplies>> wrote: Ram, I am not 100% fluent in the details of the base kinesis operator and how it interacts with Hadoop (hence my posting); if it would support that, then yes, you could. My goal is to make it so one can easily pick up where they left off reading the Kinesis stream, regardless of if you kill the application and re-launch it, etc., without needing to go out to the cli to do some commands (because at some point some operator will forget and then we will reprocess a bunch of transactions; that would not be good! Jim From: Munagala Ramanath [mailto:r...@datatorrent.com<mailto:r...@datatorrent.com>] Sent: Tuesday, February 16, 2016 5:21 PM To: users@apex.incubator.apache.org<mailto:users@apex.incubator.apache.org> Subject: Re: Kinesis Operator Help Why use the application id ? Could you generate and use a java.util.UUID for example and save it in HDFS ? Ram On Tue, Feb 16, 2016 at 11:40 AM, Jim mailto:jim@facility.supplies>> wrote: Good morning, I am new to Apex, Hadoop and Yarn (nothing like tackling something new, is there?). I have my first Apex apps working that are edi processors that read new edi transactions from an Amazon Kinesis stream, look at the data, and routes the edi data to an appropriate handler for processing (note the operatorEs pushes the data to ElasticSearch for logging). Here is a diagram: [cid:image001.jpg@01D16981.0461C8F0] Everything launches, and is working fine with the above diagram from the edi router through the transaction operators. The final challenge I am having, being new to all of this, is that the Kinesis operator, by default, stores it’s app id in into IdempotentStorageManager (aka WindowDataManager) when it is launched, so if the app it shutdown and restarted this same app id is used by default with the checkpoint so you don’t reprocess the same records again when the application is restarted. You can see this id immediately to the right of the Operations / apps in gray lettering ‘application_1453741656046_0520’ in the image from the datatorrent console below: [cid:image004.png@01D168BA.5FE56550] However, if you kill the application, and re-launch, this id changes, and it starts reading from the Kinesis stream back from the beginning; and the only way to restart it so it starts where it left off is using the cli as follows: 1.)Run ‘dtcli’ from the command line. 2.)Run ‘launch -originalAppId “application_1453741656046_0520” ’ This will launch the application using the same app id identified in the console screen above. I want to make this easier, but need some experts help in tweaking this so it works. I am thinking that there should be a way with Kinesis to: 1.)Define in the properties, a Kinesis app id string value. 2.)If this value is defined, it will use that, when launching the application, to check if an Hadoop app id has already been assigned to that identifier. 3.)If that value is not yet stored in the database, it will launch the app, creating a new app id, and store the app id under the identifier key value. 4.)Now if I kill the app, or install new software, it will always pick up where it left off by using the identifier key value to retri
Re: Kinesis Operator Help
Jim, Is there state in the rest of the operators in your application? If that cases storing the offsets only the input operator(s) is not sufficient as you will lose the state from the other operators in your DAG. Second modifying idempotent storage manager to store in an external location, while can be done, will not always result in correct behavior. This is because the state of the input operator is saved using idempotent storage manager as soon as the end of a window is reached for the input operator but the other operators downstream in the DAG may not yet have seen or processed the data. So if you restart from these offsets directly that can result in loss of data. Launching of the application by passing the previous application id (the -originalAppId flag) handles this correctly by using the saved state of all the operators and starting the input operator at a window that does not cause any loss of data. There is a limitation to this. It is that when you relaunch the application the application code shouldn't have changed in a way that the operator is no longer deserializable from the previous state. If that is not a problem for you, for example if you are not changing code between reruns then this approach is the best. If you cannot launch from previous application state described above for reasons such as your application code is going to change significantly and if you dont have state in the rest of the application you can save the offsets external and load them back at startup. In that case you want to store the offsets from committed window which are guaranteed to have been processed by all operators in the downstream of the DAG. This is a more complicated approach as the operator has to keep track of committed offsets and save them. I would recommend to go with the launch using previous application id if possible. Thanks On Tue, Feb 16, 2016 at 11:40 AM, Jim wrote: > Good morning, > > > > I am new to Apex, Hadoop and Yarn (nothing like tackling something new, is > there?). > > > > I have my first Apex apps working that are edi processors that read new > edi transactions from an Amazon Kinesis stream, look at the data, and > routes the edi data to an appropriate handler for processing (note the > operatorEs pushes the data to ElasticSearch for logging). Here is a > diagram: > > > > > > Everything launches, and is working fine with the above diagram from the > edi router through the transaction operators. > > > > The final challenge I am having, being new to all of this, is that the > Kinesis operator, by default, stores it’s app id in into > IdempotentStorageManager (aka WindowDataManager) when it is launched, so if > the app it shutdown and restarted this same app id is used by default with > the checkpoint so you don’t reprocess the same records again when the > application is restarted. > > > > You can see this id immediately to the right of the Operations / apps in > gray lettering ‘application_1453741656046_0520’ in the image from the > datatorrent console below: > > > > [image: cid:image004.png@01D168BA.5FE56550] > > > > However, if you kill the application, and re-launch, this id changes, and > it starts reading from the Kinesis stream back from the beginning; and the > only way to restart it so it starts where it left off is using the cli as > follows: > > > > 1.)Run ‘dtcli’ from the command line. > > 2.)Run ‘launch -originalAppId “application_1453741656046_0520” to .apa file>’ > > > > This will launch the application using the same app id identified in the > console screen above. > > > > I want to make this easier, but need some experts help in tweaking this so > it works. > > > > I am thinking that there should be a way with Kinesis to: > > > > 1.)Define in the properties, a Kinesis app id string value. > > 2.)If this value is defined, it will use that, when launching the > application, to check if an Hadoop app id has already been assigned to that > identifier. > > 3.)If that value is not yet stored in the database, it will launch > the app, creating a new app id, and store the app id under the identifier > key value. > > 4.)Now if I kill the app, or install new software, it will always > pick up where it left off by using the identifier key value to retrieve and > assign the app id. > > > > Sounds simple, right? J > > > > Can one of the experts out there help me figure this out as I don’t want > to reprocess already processed edi transactions? > > > > Thanks, > > > > Jim > > > > > > > > > > Jim > > > jim@facility.supplies (414) 760-7711 > -- > The information contained in this communication, including any files or > attachments transmitted with it, may contain copyrighted information or > information that is confidential and exempt from disclosure under > applicable laws and regulations, is intended only for the use of the > recipient(s) named above, and may be legally privileged. If the reader of > this message is not th