apex on emr?

[Ashwin Chandra Putta]

Hey guys,

Any of you running Apex in production on EMR?

-- 

Regards,
Ashwin.

Re: Blocked operator PTOperator

[Ashwin Chandra Putta]

Sunil,
This might be related to checkpointing. See:
https://github.com/apache/apex-core/blob/master/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java#L2211-L2217

Also check this piece of code:
https://github.com/apache/apex-core/blob/master/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java#L2031-L2044

Can you paste the output of the warning from the code above which starts
with 'Marking operator '

Regards,
Ashwin.

On Tue, Feb 28, 2017 at 12:03 PM, Sunil Parmar 
wrote:

> That doesn’t seems to be the case. We do see window id moving in UI as
> well.
>
> On 2017-02-28 11:19 (-0800), Munagala Ramanath 
> wrote:
> > It likely means that that operator is taking too long to return from one
> of
> > the callbacks like beginWindow(), endWindow(),
> > emitTuples(), etc. Do you have any potentially blocking calls to external
> > systems in any of those callbacks ?
> >
> > Ram
> >
> > On Tue, Feb 28, 2017 at 11:09 AM, Sunil Parmar  >
> > wrote:
> >
> > > 2017-02-27 19:43:21,926 INFO com.datatorrent.stram.
> StreamingContainerManager:
> > > Blocked operator PTOperator[id=3,name=eventUpdatesFormatter] container
> > > PTContainer[id=1(container_1487310232732_0027_02_000111),state=ACTIVE]
> > > time 61905ms
> > > 2017-02-27 19:43:22,928 INFO com.datatorrent.stram.
> StreamingAppMasterService:
> > > Completed containerId=container_1487310232732_0027_02_000111,
> > > state=COMPLETE, exitStatus=-105, diagnostics=Container killed by the
> > > ApplicationMaster.
> > > Container killed on request. Exit code is 143
> > > Container exited with a non-zero exit code 143
> > >
> > >
> > > Can anyone help understand this error ? We see one of the operators
> keeps
> > > restarting the container; the above error is from AppMaster log.
> > >
> > > Thanks,
> > > Sunil
> > >
> >
> >
> >
> > --
> >
> > ___
> >
> > Munagala V. Ramanath
> >
> > Software Engineer
> >
> > E: r...@datatorrent.com | M: (408) 331-5034 | Twitter: @UnknownRam
> >
> > www.datatorrent.com  |  apex.apache.org
> >
>



-- 

Regards,
Ashwin.

Re: Occasional Out of order tuples when emitting from a thread

[Ashwin Chandra Putta]

Sunil,

You can poll the queue in end window since process method in the input port
does not get called if there is no incoming tuple. However, end window is
called irrespective of there are incoming tuples or not.

Regards,
Ashwin.

On Tue, Feb 21, 2017 at 11:32 AM, Sunil Parmar 
wrote:

> Ram,
> Thanks for the prompt response. If we use the approach you suggested we’re
> dependent on main thread’s process call I.e. Tuples in the thread safe
> queue gets only processed when main thread is processing incoming tuples.
> How can we explicitly call the process from polling of delay queue ?
>
> Just for reference here’s the sample code snippet for our operator.
>
> public class MyOperator extends BaseOperator implements
>
> Operator.ActivationListener {
> …..
>
> @InputPortFieldAnnotation
>
> public transient DefaultInputPort kafkaStreamInput =
>
> new DefaultInputPort() {
>
> List errors = new ArrayList();
>
> @Override
>
> public void process(String consumerRecord) {
>
> //Code for normal tuple process
>
> //Code to poll thread safe queue
>
> }
>
> ***—*
> *From: *Munagala Ramanath 
> *To: *users@apex.apache.org
> *CC: *"d...@apex.apache.org" , Allan De Leon <
> adel...@threatmetrix.com>, Tim Zhu 
> *Subject: *Re: Occasional Out of order tuples when emitting from a thread
> *Date: *2017-02-21 10:08 (-0800)
> *List: *users@apex.apache.org
> 
>
> Please note that tuples should not be emitted by any thread other than the
> main operator thread.
>
> A common pattern is to use a thread-safe queue and have worker threads
> enqueue
> tuples there; the main operator thread then pulls tuples from the queue and
> emits them.
>
> Ram
>
> ___
>
> Munagala V. Ramanath
>
> Software Engineer
>
> E: r...@datatorrent.com | M: (408) 331-5034 | Twitter: @UnknownRam
> www.datatorrent.com  |  apex.apache.org
>
>
> From: Sunil Parmar 
> Date: Tuesday, February 21, 2017 at 10:05 AM
> To: "users@apex.apache.org" , "d...@apex.apache.org"
> 
> Cc: Allan De Leon , Tim Zhu <
> t...@threatmetrix.com>
> Subject: Occasional Out of order tuples when emitting from a thread
>
> Hi there,
> We have the following setup:
>
>- we have a generic operator that’s processing tuples in its input port
>- in the input port’s process method, we check for a condition, and:
>   - if the condition is met, the tuple is emitted to the next
>   operator right away (in the process method)
>   - Otherwise, if the condition is not met, we store the tuple  in
>   some cache and we use some threads that periodically check the 
> condition to
>   become true. Once the condition is true, the threads call the emit 
> method
>   on the stored tuples.
>
> With this setup, we occasionally encounter the following error:
> 2017-02-15 17:29:09,364 ERROR com.datatorrent.stram.engine.GenericNode:
> Catastrophic Error: Out of sequence BEGIN_WINDOW tuple 58a404613b7f on
> port transformedJSON while expecting 58a404613b7e
>
> Is there a way to make the above work correctly?
> If not, can you recommend a better way of doing this?
> How can we ensure window assignment is done synchronously before emitting
> tuples ?
>
> Thanks very much in advance…
> -allan
>



-- 

Regards,
Ashwin.

Re: One-time Initialization of in-memory data using a data file

[Ashwin Chandra Putta]

Roger,

Depending on the certain requirements on expected latency, size of data
etc, the operator's design will change.

If latency needs to be lowest possible, meaning completely in-memory and
not hitting the disk for read I/O, there are two scenarios
1. If the lookup data size is small --> just load to memory in the setup
call, switch off checkpointing to get rid off checkpoint I/O latency in
between. In case of operator restarts, the data should be reloaded in setup.
2. If the lookup data is large --> have many partitions of this operator to
minimize the footprint of each partition. Still switch off checkpointing
and reload in setup in case of operator restart. Having many partitions
will ensure that the setup load is fast. The incoming query needs to be
partitioned based on the lookup key.

You can use the PojoEnricher with FSLoader for above design.

Code:
https://github.com/apache/apex-malhar/blob/master/contrib/src/main/java/com/datatorrent/contrib/enrich/POJOEnricher.java
Example:
https://github.com/DataTorrent/examples/tree/master/tutorials/enricher

In case of large lookup dataset and latency caused by disk read I/O is
fine, then use HDHT or managed state as a backup mechanism for the
in-memory data to decrease the checkpoint footprint. I could not find
example for managed state but here are the links for HDHT..

Code:
https://github.com/DataTorrent/Megh/tree/master/contrib/src/main/java/com/datatorrent/contrib/hdht
Example:
https://github.com/DataTorrent/examples/blob/master/tutorials/hdht/src/test/java/com/example/HDHTAppTest.java

Regards,
Ashwin.

On Sun, Jan 22, 2017 at 10:45 PM, Sanjay Pujare 
wrote:

> You may want to take a look at com.datatorrent.lib.fileaccess.DTFileReader
> in the malhar-library – not sure whether it gives you reading the whole
> file into memory.
>
>
>
> Also there is a library called Megh at https://github.com/DataTorrent/Megh
> where you might find some useful operators like
> com.datatorrent.contrib.hdht.hfile.HFileImpl .
>
>
>
> *From: *Roger F 
> *Reply-To: *
> *Date: *Sunday, January 22, 2017 at 9:32 PM
> *To: *
> *Subject: *One-time Initialization of in-memory data using a data file
>
>
>
> Hi,
>
> I have a use case where application business data needs migrated from a
> legacy system (such as mainframe) into HDFS and then loaded for use by an
> Apex application.
>
> To get this done, an approach that is being considered to perform one-time
> initialization of the data from the HDFS into application memory. This data
> will then be queried for various business logic functions of the
> application.
>
> Once the data is loaded, this operator/module (?) should no longer perform
> any further function except for acting as a master of this data and then
> supporting operations to query the data (via a key).
>
> Any pointers to how this can be done ? I was looking for an operator or
> any other entity which can load this data at startup (Activation or Setup)
> and then allow queries to be submitted to it via an input port.
>
>
>
> -R
>



-- 

Regards,
Ashwin.

Re: PARTITION_PARALLEL Vs Regular mode

[Ashwin Chandra Putta]

+1 for option 4 that Amol suggested. However, if you definitely need the
shuffle, one thing you can check is if the key based partitioning is
causing skews in the throughputs to individual parquet writer partitions.

Regards,
Ashwin.

On Thu, Dec 22, 2016 at 1:22 PM, Amol Kekre  wrote:

>
> Arvindan,
> Based on what you have it looks like shuffle is not needed between
> Kafka->ParquetWriter. The decision to use parallel partiion should be
> ideally based the need to shuffle. If so option [1] should not be used per
> se. Why even bother to shuffle if you do not need to.
>
> Assuming the ask is really between option [2] and [3], the bottleneck in
> your mini-dags (each parallel partition) will dictate the number of
> partitions. So here are the questions I would ask
> 1. What are the minimum number of ParquetWriters you need to meet your
> SLA? // benchmark each to find out
> 2. Are you using 120 KafkaConsumers because there are 120 topics? or is it
> N topics/120 (where N >> 120) to balance the load? i.e. do you have a
> choice to have more kafka partitions
>
> In general for parallel partiion, the number of partitions should as per
> the bottleneck operator. Assuming in this case it is Parquet and the answer
> is 200 partitions, why would 200 Kafka->200 Parquet with container local
> setup not be ok. If they share containers, the memory may actually be same
> or lower than 300 separate containers (100 Kafka + 200 Parquet). Kafka and
> Parquet most likely do not compete for same resource (I/O vs CPU), so
> option [4] as follows should work
>
> 4.
> - Figure out minimum Parquet writers to meet your SLA
> - Bump up Kafka to those many partitions
> - Use container local + parallel partition
> - Experiment with container memory size to bring net memory down
> In this case, you will save on I/O between Kafka->Parquet routing through
> NIC; no serialization between Kafka->Parquet. There is a small chance that
> thread-local between Kafka->Parquet may work, which will move back-pressure
> (from some topic skew, spikes, ) to Kafka publisher.
>
> Thks
> Amol
>
>
>
> On Thu, Dec 22, 2016 at 11:06 AM, Pramod Immaneni 
> wrote:
>
>> Arvindan,
>>
>> When you had the MxN case with 100 kafka consumers sending to 120 parquet
>> writers what was the cpu utilization of the parquet containers. Was it
>> close to 100% or did you have spare cycles? I am trying to determine if it
>> is an IO bottleneck or processing.
>>
>> Thanks
>>
>> On Thu, Dec 22, 2016 at 10:50 AM, Arvindan Thulasinathan <
>> aravindan.thulasinat...@oracle.com> wrote:
>>
>>> Hi,
>>>   We have an Apex Application which has a DAG structure like this:
>>>
>>> KafkaConsumer —>  ParquetWriter
>>>
>>> The KafkaConsumer is running at a scale where we have 100 containers for
>>> consumer consuming from a Kafka-Cluster with an incoming rate of 300K
>>> msg/sec and each message is about 1KB (Each message is a highly nested Avro
>>> message). We arrived at the 100 container number for the consumer in order
>>> to keep-up with the incoming traffic. The ParquetWriter is a bit CPU
>>> intensive in our case and we thought we may require about 120 -
>>> 130Containers.
>>>
>>>
>>> We have 3 different observations here:
>>>   1. Using 100 KafkaConsumer and 120 ParquetWriter Without Partition
>>> Parallel:
>>>  In this case, Apex automatically introduces a
>>> pass-through unifier.  In this scenario, we have seen that invariably
>>> ParquetWriter processes tuples at a lesser rate than KafkaConsumer’s emit
>>> rate. That is, if Consumer emits Tuples at the rate of 20 million/sec, the
>>> ParquetWriter will only write at the rate of 17 million/sec. Also, it
>>> invariably leads to backpressure and makes the consumer consume at a lower
>>> rate. I have tried going beyond 120 containers as well and
>>> I believe a possible reason could be - Unifier and Writer are in the
>>> same container and presumably share the same core. And hence they are
>>> slower? Is this observation correct? I tried tuning by increasing the
>>> Writer.Inputport.QUEUE_SIZE to 10K. The queue is not even getting half
>>> full, but still the back-pressure is created for the consumer. Is there any
>>> additional tune-up that I can do, to:
>>>A. Make the writer process tuples at almost the same pace as Consumer
>>> without backpressure on the Consumer
>>>
>>>   2. Using Partition Parallel with 120 KafkaConsumer and 120
>>> ParquetWriter:
>>> In this scenario as well, we have seen that ParquetWriter
>>> processes tuples at a lesser rate than KafkaConsumer’s emit rate. That is,
>>> if Consumer emits Tuples at the rate of 20 million/sec, the ParquetWriter
>>> will only write at the rate of 19 million/sec. This behavior is true even
>>> if we keep increasing the Consumers and writers to 130 or 140 containers. I
>>> believe this behavior is because we think the wrier is a bit CPU intensive.
>>>
>>> 3. Using a different DAG structure like 

Re: Operators in Pending Deploy

[Ashwin Chandra Putta]

Ajay,

Can you specify the reason why it did not work before as well?

Regards,
Ashwin.

On Fri, Dec 2, 2016 at 10:55 AM, AJAY GUPTA <ajaygit...@gmail.com> wrote:

> Hi Max,
>
> Can you try adding the following configurations to yarn-site.xml. Restart
> yarn and then try starting wordcount-demo.
>
>
> 
>yarn.scheduler.minimum-allocation-mb
>256
>  
>  
>yarn.scheduler.maximum-allocation-mb
>512
>  
>
>
> Regards,
> Ajay
>
>
> On Fri, Dec 2, 2016 at 10:26 PM, Max Bridgewater <
> max.bridgewa...@gmail.com> wrote:
>
>> Yeah, application has been running for 20h. But no event is flowing
>> through. See dt.log attached.
>>
>> On Fri, Dec 2, 2016 at 11:48 AM, Ashwin Chandra Putta <
>> ashwinchand...@gmail.com> wrote:
>>
>>> Max,
>>>
>>> Can you check app master logs? If application status changed to running,
>>> it means app master is running. You can find operator deployment related
>>> logs from app master dt.log.
>>>
>>> Regards,
>>> Ashwin.
>>>
>>> On Dec 2, 2016 5:25 AM, "Max Bridgewater" <max.bridgewa...@gmail.com>
>>> wrote:
>>>
>>>> I deployed the WordCountDemo in DataTorrent RTS Enterprise with
>>>> evaluation license. The application is in RUNNING state and resulted into
>>>> two processes that are all in ACTIVE state.
>>>>
>>>> On the other hand, however, the operators themselves are in
>>>> PENDING_DEPLOY state. These are wordinput, count, and console. So, nothing
>>>> is really running and no words are being counted.
>>>>
>>>> There seems to be enough resources:
>>>>
>>>> 2016-12-01 20:36:28,287 INFO org.apache.hadoop.yarn.server.
>>>> resourcemanager.sched
>>>> uler.SchedulerNode: Assigned container 
>>>> container_1480549373717_0021_01_02
>>>> of
>>>>  capacity <memory:3000, vCores:1> on host localhost:36079, which has 2
>>>> container
>>>> s, <memory:6000, vCores:2> used and <memory:2192, vCores:6> available
>>>> after allo
>>>> cation.
>>>>
>>>>
>>>> In /var/log/dtgateway.log, here is what I see. Can this be the cause?
>>>> If so, how do I fix this?
>>>>
>>>> 016-12-02 13:21:34,113 WARN com.datatorrent.gateway.x: Cannot update
>>>> license registry for the number of nodes
>>>> com.datatorrent.a.E: Filesystem closed
>>>> at com.datatorrent.a.M.b(w:341)
>>>> at com.datatorrent.a.C.b(m:34)
>>>> at com.datatorrent.gateway.x.b(jd:456)
>>>> at com.datatorrent.gateway.x.b(jd:627)
>>>> at com.datatorrent.gateway.x.b(jd:787)
>>>> at com.datatorrent.gateway.x.b(jd:141)
>>>> at com.datatorrent.gateway.U.run(jd:210)
>>>> at java.util.concurrent.Executors$RunnableAdapter.call(Executor
>>>> s.java:51
>>>> 1)
>>>> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>>>> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPool
>>>> Executor.
>>>> java:1142)
>>>> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoo
>>>> lExecutor
>>>> .java:617)
>>>> at java.lang.Thread.run(Thread.java:745)
>>>> 2016-12-02 13:21:34,113 WARN com.datatorrent.gateway.x: Detected
>>>> containers not
>>>> provisioned for application_1480549373717_0021: # planned containers
>>>> is 3 and #
>>>> allocated containers is 1
>>>> 2016-12-02 13:21:39,158 WARN com.datatorrent.gateway.x: Cannot update
>>>> license re
>>>> gistry for the number of nodes
>>>> com.datatorrent.a.E: Filesystem closed
>>>> at com.datatorrent.a.M.b(w:341)
>>>> at com.datatorrent.a.C.b(m:34)
>>>> at com.datatorrent.gateway.x.b(jd:456)
>>>> at com.datatorrent.gateway.x.b(jd:627)
>>>> at com.datatorrent.gateway.x.b(jd:787)
>>>> at com.datatorrent.gateway.x.b(jd:141)
>>>> at com.datatorrent.gateway.U.run(jd:210)
>>>> at java.util.concurrent.Executors$RunnableAdapter.call(Executor
>>>> s.java:51
>>>> 1)
>>>>
>>>>
>>>>
>>
>


-- 

Regards,
Ashwin.

Re: Operators in Pending Deploy

[Ashwin Chandra Putta]

Max,

Can you check app master logs? If application status changed to running, it
means app master is running. You can find operator deployment related logs
from app master dt.log.

Regards,
Ashwin.

On Dec 2, 2016 5:25 AM, "Max Bridgewater"  wrote:

> I deployed the WordCountDemo in DataTorrent RTS Enterprise with evaluation
> license. The application is in RUNNING state and resulted into two
> processes that are all in ACTIVE state.
>
> On the other hand, however, the operators themselves are in PENDING_DEPLOY
> state. These are wordinput, count, and console. So, nothing is really
> running and no words are being counted.
>
> There seems to be enough resources:
>
> 2016-12-01 20:36:28,287 INFO org.apache.hadoop.yarn.server.
> resourcemanager.sched
> uler.SchedulerNode: Assigned container container_1480549373717_0021_01_02
> of
>  capacity  on host localhost:36079, which has 2
> container
> s,  used and  available
> after allo
> cation.
>
>
> In /var/log/dtgateway.log, here is what I see. Can this be the cause? If
> so, how do I fix this?
>
> 016-12-02 13:21:34,113 WARN com.datatorrent.gateway.x: Cannot update
> license registry for the number of nodes
> com.datatorrent.a.E: Filesystem closed
> at com.datatorrent.a.M.b(w:341)
> at com.datatorrent.a.C.b(m:34)
> at com.datatorrent.gateway.x.b(jd:456)
> at com.datatorrent.gateway.x.b(jd:627)
> at com.datatorrent.gateway.x.b(jd:787)
> at com.datatorrent.gateway.x.b(jd:141)
> at com.datatorrent.gateway.U.run(jd:210)
> at java.util.concurrent.Executors$RunnableAdapter.
> call(Executors.java:51
> 1)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at java.util.concurrent.ThreadPoolExecutor.runWorker(
> ThreadPoolExecutor.
> java:1142)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(
> ThreadPoolExecutor
> .java:617)
> at java.lang.Thread.run(Thread.java:745)
> 2016-12-02 13:21:34,113 WARN com.datatorrent.gateway.x: Detected
> containers not
> provisioned for application_1480549373717_0021: # planned containers is 3
> and #
> allocated containers is 1
> 2016-12-02 13:21:39,158 WARN com.datatorrent.gateway.x: Cannot update
> license re
> gistry for the number of nodes
> com.datatorrent.a.E: Filesystem closed
> at com.datatorrent.a.M.b(w:341)
> at com.datatorrent.a.C.b(m:34)
> at com.datatorrent.gateway.x.b(jd:456)
> at com.datatorrent.gateway.x.b(jd:627)
> at com.datatorrent.gateway.x.b(jd:787)
> at com.datatorrent.gateway.x.b(jd:141)
> at com.datatorrent.gateway.U.run(jd:210)
> at java.util.concurrent.Executors$RunnableAdapter.
> call(Executors.java:51
> 1)
>
>
>

Re: java.lang.NoClassDefFoundError: com/github/fge/jsonschema/exceptions/ProcessingException

[Ashwin Chandra Putta]

Max,

Seems like you need to add the following dependency.


  com.github.fge
  json-schema-validator
  2.0.1


See the following example, but it is marked as optional so not sure if it
is really required.

https://github.com/DataTorrent/examples/blob/master/tutorials/parser/pom.xml

Shubham, Priyanka,

Any thoughts?

Regards,
Ashwin.

On Thu, Dec 1, 2016 at 11:14 AM, Max Bridgewater 
wrote:

> I am trying to use the Malhar JSonParser. My code goes like this:
>
>
>   KafkaSinglePortInputOperator in = dag.addOperator("kafkaIn", new
> KafkaSinglePortInputOperator());
> in.setInitialOffset(AbstractKafkaInputOperator.
> InitialOffset.EARLIEST.name());
> JsonParser parser= dag.addOperator("jsonParser",new JsonParser());
> LineOutputOperator out = dag.addOperator("fileOut", new
> LineOutputOperator());
> dag.addStream("parse", in.outputPort, parser.in);
> dag.addStream("data",parser.out , out.input);
>
>
> However, when I deploy it , I get following exception. This seems to be a
> guava class. Am I supposed to add guava as dependency in my app? I thought
> all of  contrib dependencies are provided by the Apex runtime environment.
> Any thoughts?
>
>
> the application. Server message: java.lang.NoClassDefFoundError:
> com/github/fge/jsonschema/exceptions/ProcessingException
> at com.example.myapexapp.KafkaApp.populateDAG(KafkaApp.java:24)
> at com.datatorrent.stram.plan.logical.LogicalPlanConfiguration.prepareDAG(
> LogicalPlanConfiguration.java:2226)
> at com.datatorrent.stram.client.StramAppLauncher$1.createApp(
> StramAppLauncher.java:490)
> at com.datatorrent.stram.client.StramAppLauncher.launchApp(
> StramAppLauncher.java:601)
> at com.datatorrent.stram.cli.ApexCli$LaunchCommand.execute(
> ApexCli.java:2062)
> at com.datatorrent.stram.cli.ApexCli.launchAppPackage(ApexCli.java:3451)
> at com.datatorrent.stram.cli.ApexCli.access$7400(ApexCli.java:153)
> at com.datatorrent.stram.cli.ApexCli$LaunchCommand.execute(
> ApexCli.java:1906)
> at com.datatorrent.stram.cli.ApexCli$3.run(ApexCli.java:1468)
> Caused by: java.lang.ClassNotFoundException: com.github.fge.jsonschema.
> exceptions.ProcessingException
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at java.net.FactoryURLClassLoader.loadClass(URLClassLoader.java:814)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> ... 9 more
>



-- 

Regards,
Ashwin.

Re: Connection refused exception

[Ashwin Chandra Putta]

Can you check in stram events on the gateway UI, you can select a time
range to begin at the start of the application so you can see the series of
events that happened in a timeline. Then you will find the first error in
that list. You can find stram events in the logical tab to the right for
the application in monitor page.

Regards,
Ashwin.

On Wed, Nov 30, 2016 at 8:20 AM, Vlad Rozov  wrote:

> Hmm, that is strange. I would expect an ERROR in the KafkaInputOperator
> around 21:01. Please check the earliest killed container log.
>
> 2016-11-24 21:01:01,839 [IPC Server handler 4 on 44453] ERROR 
> stram.StreamingContainerManager
> processOperatorFailure - Initiating container restart after operator
> failure PTOperator[id=1,name=kafkaInputOperator]
>
> Thank you,
>
> Vlad
>
> On 11/30/16 07:17, Feldkamp, Brandon (CONT) wrote:
>
> This is the only other stacktrace I could find but it’s dated after the
> initial cause of failure.
>
>
>
> 2016-11-24 21:52:24,681 [14/kafkaInputOperator:KafkaInputOperator] ERROR
> engine.StreamingContainer run - Operator set [OperatorDeployInfo[id=14,
> name=kafkaInputOperator,type=INPUT,checkpoint={58378663239f, 0,
> 0},inputs=[],outputs=[OperatorDeployInfo.OutputDeployInfo[portName=outputPort,streamId=KafkaInput
> -> AuthParser,bufferServer=ip-200-120-36 stopped running due to an
> exception.
>
> java.lang.RuntimeException: replay
>
> at com.datatorrent.contrib.kafka.AbstractKafkaInputOperator.
> replay(AbstractKafkaInputOperator.java:330)
>
> at com.datatorrent.contrib.kafka.AbstractKafkaInputOperator.
> beginWindow(AbstractKafkaInputOperator.java:266)
>
> at com.datatorrent.stram.engine.InputNode.run(InputNode.java:122)
>
> at com.datatorrent.stram.engine.StreamingContainer$2.run(
> StreamingContainer.java:1407)
>
> Caused by: java.io.EOFException: Received -1 when reading from channel,
> socket has likely been closed.
>
> at kafka.utils.Utils$.read(Utils.scala:376)
>
> at kafka.network.BoundedByteBufferReceive.readFrom(
> BoundedByteBufferReceive.scala:54)
>
> at kafka.network.Receive$class.readCompletely(Transmission.
> scala:56)
>
> at kafka.network.BoundedByteBufferReceive.readCompletely(
> BoundedByteBufferReceive.scala:29)
>
> at kafka.network.BlockingChannel.receive(BlockingChannel.scala:
> 100)
>
> at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.
> scala:81)
>
> at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$
> $sendRequest(SimpleConsumer.scala:71)
>
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$
> apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:109)
>
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$
> apply$mcV$sp$1.apply(SimpleConsumer.scala:109)
>
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$
> apply$mcV$sp$1.apply(SimpleConsumer.scala:109)
>
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(
> SimpleConsumer.scala:108)
>
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(
> SimpleConsumer.scala:108)
>
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(
> SimpleConsumer.scala:108)
>
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>
> at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:107)
>
> at kafka.javaapi.consumer.SimpleConsumer.fetch(
> SimpleConsumer.scala:47)
>
> at com.datatorrent.contrib.kafka.AbstractKafkaInputOperator.
> replay(AbstractKafkaInputOperator.java:304)
>
> ... 3 more
>
> 2016-11-24 21:52:24,690 [14/kafkaInputOperator:KafkaInputOperator] ERROR
> engine.StreamingContainer run - Shutdown of operator
> OperatorDeployInfo[id=14,name=kafkaInputOperator,type=INPUT,checkpoint={58378663239f,
> 0, 
> 0},inputs=[],outputs=[OperatorDeployInfo.OutputDeployInfo[portName=outputPort,streamId=KafkaInput
> -> AuthParser,bufferServer=ip-200-120-36]]] failed due to an exception.
>
> java.lang.NullPointerException
>
> at com.datatorrent.contrib.kafka.SimpleKafkaConsumer.close(
> SimpleKafkaConsumer.java:348)
>
> at org.apache.commons.io.IOUtils.closeQuietly(IOUtils.java:303)
>
> at com.datatorrent.contrib.kafka.KafkaConsumer.stop(
> KafkaConsumer.java:157)
>
> at com.datatorrent.contrib.kafka.AbstractKafkaInputOperator.
> deactivate(AbstractKafkaInputOperator.java:405)
>
> at com.datatorrent.stram.engine.Node.deactivate(Node.java:646)
>
> at com.datatorrent.stram.engine.StreamingContainer.teardownNode(
> StreamingContainer.java:1347)
>
> at com.datatorrent.stram.engine.StreamingContainer.access$500(
> StreamingContainer.java:130)
>
> at com.datatorrent.stram.engine.StreamingContainer$2.run(
> StreamingContainer.java:1438)
>
>
>
>
>
> *From: *Vlad Rozov  

Re: Apex Communication Protocols

[Ashwin Chandra Putta]

Max,

Apex deploys a fast pub-sub server called buffer server in every yarn
container it gets (except AM) before deploying operators on it. For all the
operators which are connected downstream to operators outside the current
container, their output ports become publishers to the buffer server. The
downstream operators' input ports become subscribers to the buffer server.
So there is no concept a central operator/port registry, however all the
downstream operators do register their input ports with the buffer server.
The serialization algorithm is kryo based. The data transport protocol is
based on Netlet: https://github.com/DataTorrent/Netlet which is on top of
TCP/IP.

Regards,
Ashwin.

On Fri, Nov 25, 2016 at 8:33 AM, Max Bridgewater 
wrote:

> Hi Folks,
>
> I was giving an Apex demo the other day and people asked following
> questions:
>
> 1) what is the communication protocol between operators when they are on
> distant nodes. That means, how does Apex transport the tuples from one node
> to the other?
> Is it a custom protocol on top of TCP/IP or is it RPC?
> 2) What is the serialization algorithm used?
> 3) What is the addressing scheme between operators? That means how does
> Apex know where an operator is located and how to route data to it? Is
> there an operator registry? If so, where does it reside?
>
> Thoughts?
>
> Thanks,
> Max.
>
>


-- 

Regards,
Ashwin.

Re: Apex with SSH Tunel

[Ashwin Chandra Putta]

Cool. I did not see these errors in the AM logs, were these in the node
manager logs?

Regards,
Ashwin.

On Mon, Nov 21, 2016 at 4:20 AM, Max Bridgewater <max.bridgewa...@gmail.com>
wrote:

> The issue turned out to be memory allocation. Here is the relevant YARN
> error message:
>
> 2016-11-21 11:44:30,020 INFO org.apache.hadoop.yarn.server.
> nodemanager.containermanager.container.ContainerImpl: Container
> container_1479728463466_0001_02_01 transitioned from LOCALIZED to
> RUNNING
> 2016-11-21 11:44:31,858 INFO org.apache.hadoop.yarn.server.
> nodemanager.containermanager.monitor.ContainersMonitorImpl: Starting
> resource-monitoring for container_1479728463466_0001_02_01
> 2016-11-21 11:44:31,858 INFO org.apache.hadoop.yarn.server.
> nodemanager.containermanager.monitor.ContainersMonitorImpl: Stopping
> resource-monitoring for container_1479728463466_0001_01_01
> 2016-11-21 11:44:31,867 INFO org.apache.hadoop.yarn.server.
> nodemanager.containermanager.monitor.ContainersMonitorImpl: Memory usage
> of ProcessTree 26632 for container-id container_1479728463466_0001_02_01:
> 194.5 MB of 1 GB physical memory used; 2.5 GB of 2.1 GB virtual memory used
> 2016-11-21 11:44:34,875 INFO org.apache.hadoop.yarn.server.
> nodemanager.containermanager.monitor.ContainersMonitorImpl: Memory usage
> of ProcessTree 26632 for container-id container_1479728463466_0001_02_01:
> 532.4 MB of 1 GB physical memory used; 2.6 GB of 2.1 GB virtual memory used
> 2016-11-21 11:44:34,876 WARN org.apache.hadoop.yarn.server.
> nodemanager.containermanager.monitor.ContainersMonitorImpl: Process tree
> for container: container_1479728463466_0001_02_01 has processes older
> than 1 iteration running over the configured limit. Limit=2254857728,
> current usage = 2822131712
> 2016-11-21 11:44:34,876 WARN org.apache.hadoop.yarn.server.
> nodemanager.containermanager.monitor.ContainersMonitorImpl: Container
> [pid=26632,containerID=container_1479728463466_0001_02_01] is running
> beyond virtual memory limits. Current usage: 532.4 MB of 1 GB physical
> memory used; 2.6 GB of 2.1 GB virtual memory used. Killing container.
>
>
> I solved it by adding this to yarn-site.xml:
>
> 
> yarn.scheduler.minimum-allocation-mb
> 1000
> 
>
>
> Thanks,
> Max.
>
>
>
>
> On Sat, Nov 19, 2016 at 10:30 PM, Ashwin Chandra Putta <
> ashwinchand...@gmail.com> wrote:
>
>> Max,
>>
>> The app failure does not depend on the gateway. The gateway is a daemon
>> to launch Apex apps on YARN and to get metrics for the Apex apps from YARN
>> and AM for each app, so it won't affect app execution once YARN accepts the
>> application. For some reason the AM itself is failing. I cannot figure out
>> the cause from the logs. It is possible that the app packages for these
>> apps have hadoop dependencies packaged, it is one of the most common causes
>> for AM failure.
>>
>> Regards,
>> Ashwin.
>>
>> On Sat, Nov 19, 2016 at 3:08 PM, Max Bridgewater <
>> max.bridgewa...@gmail.com> wrote:
>>
>>> Please find the AppMaster.stderr attached as well as dt.log.
>>> AppMaster.stdout is empty. I am still wondering if there is another port
>>> that is needed or if the UI is using websocket.
>>>
>>> On Sat, Nov 19, 2016 at 5:40 PM, Ashwin Chandra Putta <
>>> ashwinchand...@gmail.com> wrote:
>>>
>>>> Max,
>>>>
>>>> Can you share the app master logs of the failed application?
>>>>
>>>> Regards,
>>>> Ashwin.
>>>>
>>>> On Sat, Nov 19, 2016 at 4:45 AM, Max Bridgewater <
>>>> max.bridgewa...@gmail.com> wrote:
>>>>
>>>>> Hi Ahswin,
>>>>>
>>>>> Thanks for the feedback. I created a completely new instance, trying
>>>>> the follow the instructions more precisely. I attached the logs again. As
>>>>> you can see they are very clean. Despite this, PIDemo is still failing
>>>>> without any meaningful error message. Same things happens with
>>>>> WorldCountDemo. After launching, it stays in ACCEPTED status for 10 to 15
>>>>> seconds and switch to FAILED.
>>>>>
>>>>> Max.
>>>>>
>>>>> On Fri, Nov 18, 2016 at 2:30 PM, Ashwin Chandra Putta <
>>>>> ashwinchand...@gmail.com> wrote:
>>>>>
>>>>>> Also, there are write permission errors on /user/dtadmin/datatorrent
>>>>>> in hdfs. Please make dtadmin user own /user/dtadmin/
>>>>>>
>>>>>&

Re: Apex with SSH Tunel

[Ashwin Chandra Putta]

Max,

The app failure does not depend on the gateway. The gateway is a daemon to
launch Apex apps on YARN and to get metrics for the Apex apps from YARN and
AM for each app, so it won't affect app execution once YARN accepts the
application. For some reason the AM itself is failing. I cannot figure out
the cause from the logs. It is possible that the app packages for these
apps have hadoop dependencies packaged, it is one of the most common causes
for AM failure.

Regards,
Ashwin.

On Sat, Nov 19, 2016 at 3:08 PM, Max Bridgewater <max.bridgewa...@gmail.com>
wrote:

> Please find the AppMaster.stderr attached as well as dt.log.
> AppMaster.stdout is empty. I am still wondering if there is another port
> that is needed or if the UI is using websocket.
>
> On Sat, Nov 19, 2016 at 5:40 PM, Ashwin Chandra Putta <
> ashwinchand...@gmail.com> wrote:
>
>> Max,
>>
>> Can you share the app master logs of the failed application?
>>
>> Regards,
>> Ashwin.
>>
>> On Sat, Nov 19, 2016 at 4:45 AM, Max Bridgewater <
>> max.bridgewa...@gmail.com> wrote:
>>
>>> Hi Ahswin,
>>>
>>> Thanks for the feedback. I created a completely new instance, trying the
>>> follow the instructions more precisely. I attached the logs again. As you
>>> can see they are very clean. Despite this, PIDemo is still failing without
>>> any meaningful error message. Same things happens with WorldCountDemo.
>>> After launching, it stays in ACCEPTED status for 10 to 15 seconds and
>>> switch to FAILED.
>>>
>>> Max.
>>>
>>> On Fri, Nov 18, 2016 at 2:30 PM, Ashwin Chandra Putta <
>>> ashwinchand...@gmail.com> wrote:
>>>
>>>> Also, there are write permission errors on /user/dtadmin/datatorrent in
>>>> hdfs. Please make dtadmin user own /user/dtadmin/
>>>>
>>>> Permission denied: user=dtadmin, access=WRITE,
>>>> inode="/user/dtadmin/datatorrent":hduser:supergroup:drwxr-xr-x
>>>>
>>>> Regards,
>>>> Ashwin.
>>>>
>>>> On Fri, Nov 18, 2016 at 11:27 AM, Ashwin Chandra Putta <
>>>> ashwinchand...@gmail.com> wrote:
>>>>
>>>>> The end  tag is missing between line 30 and 31. It is for
>>>>> the property dt.attr.DEBUG.
>>>>>
>>>>> Regards,
>>>>> Ashwin.
>>>>>
>>>>> On Fri, Nov 18, 2016 at 10:16 AM, Max Bridgewater <
>>>>> max.bridgewa...@gmail.com> wrote:
>>>>>
>>>>>> Here is the log folder. Note that it refers to a malformed
>>>>>> properties.xml. I am attaching that properties file as well.
>>>>>>
>>>>>> On Fri, Nov 18, 2016 at 1:08 PM, Ashwin Chandra Putta <
>>>>>> ashwinchand...@gmail.com> wrote:
>>>>>>
>>>>>>> Max,
>>>>>>>
>>>>>>> Can you share the gateway logs?
>>>>>>>
>>>>>>> You will find them under /var/log/datatorrent for global install, or
>>>>>>> under ~/.dt/logs for local install.
>>>>>>>
>>>>>>> Regards,
>>>>>>> Ashwin.
>>>>>>>
>>>>>>> Regards,
>>>>>>> Ashwin.
>>>>>>>
>>>>>>> On Nov 18, 2016 9:41 AM, "Max Bridgewater" <
>>>>>>> max.bridgewa...@gmail.com> wrote:
>>>>>>>
>>>>>>>> Hi Folks,
>>>>>>>>
>>>>>>>> I am playing with Apex (DataTorrent RTS Enterprise). Local
>>>>>>>> deployment in Ubuntu 16 box works fine.
>>>>>>>>
>>>>>>>> However, when I deploy on a remote host, I am not apple to launch
>>>>>>>> demo applications. My suspicion is that this is due to having to open 
>>>>>>>> an
>>>>>>>> SSH tunnel to access the gateway. All activities other than launching 
>>>>>>>> the
>>>>>>>> apps seem to work fine.
>>>>>>>>
>>>>>>>> My question: is there another port I need to open? Anybody is aware
>>>>>>>> of issues running/accessing Apex behind a proxy or firewall?
>>>>>>>>
>>>>>>>> Unfortunately the UI does not provide much information. I am
>>>>>>>> attaching some screenshots.
>>>>>>>>
>>>>>>>> Thanks,
>>>>>>>> Max.
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>>
>>>>> --
>>>>>
>>>>> Regards,
>>>>> Ashwin.
>>>>>
>>>>
>>>>
>>>>
>>>> --
>>>>
>>>> Regards,
>>>> Ashwin.
>>>>
>>>
>>>
>>
>>
>> --
>>
>> Regards,
>> Ashwin.
>>
>
>


-- 

Regards,
Ashwin.

Re: Apex with SSH Tunel

[Ashwin Chandra Putta]

Max,

Can you share the app master logs of the failed application?

Regards,
Ashwin.

On Sat, Nov 19, 2016 at 4:45 AM, Max Bridgewater <max.bridgewa...@gmail.com>
wrote:

> Hi Ahswin,
>
> Thanks for the feedback. I created a completely new instance, trying the
> follow the instructions more precisely. I attached the logs again. As you
> can see they are very clean. Despite this, PIDemo is still failing without
> any meaningful error message. Same things happens with WorldCountDemo.
> After launching, it stays in ACCEPTED status for 10 to 15 seconds and
> switch to FAILED.
>
> Max.
>
> On Fri, Nov 18, 2016 at 2:30 PM, Ashwin Chandra Putta <
> ashwinchand...@gmail.com> wrote:
>
>> Also, there are write permission errors on /user/dtadmin/datatorrent in
>> hdfs. Please make dtadmin user own /user/dtadmin/
>>
>> Permission denied: user=dtadmin, access=WRITE,
>> inode="/user/dtadmin/datatorrent":hduser:supergroup:drwxr-xr-x
>>
>> Regards,
>> Ashwin.
>>
>> On Fri, Nov 18, 2016 at 11:27 AM, Ashwin Chandra Putta <
>> ashwinchand...@gmail.com> wrote:
>>
>>> The end  tag is missing between line 30 and 31. It is for the
>>> property dt.attr.DEBUG.
>>>
>>> Regards,
>>> Ashwin.
>>>
>>> On Fri, Nov 18, 2016 at 10:16 AM, Max Bridgewater <
>>> max.bridgewa...@gmail.com> wrote:
>>>
>>>> Here is the log folder. Note that it refers to a malformed
>>>> properties.xml. I am attaching that properties file as well.
>>>>
>>>> On Fri, Nov 18, 2016 at 1:08 PM, Ashwin Chandra Putta <
>>>> ashwinchand...@gmail.com> wrote:
>>>>
>>>>> Max,
>>>>>
>>>>> Can you share the gateway logs?
>>>>>
>>>>> You will find them under /var/log/datatorrent for global install, or
>>>>> under ~/.dt/logs for local install.
>>>>>
>>>>> Regards,
>>>>> Ashwin.
>>>>>
>>>>> Regards,
>>>>> Ashwin.
>>>>>
>>>>> On Nov 18, 2016 9:41 AM, "Max Bridgewater" <max.bridgewa...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Hi Folks,
>>>>>>
>>>>>> I am playing with Apex (DataTorrent RTS Enterprise). Local deployment
>>>>>> in Ubuntu 16 box works fine.
>>>>>>
>>>>>> However, when I deploy on a remote host, I am not apple to launch
>>>>>> demo applications. My suspicion is that this is due to having to open an
>>>>>> SSH tunnel to access the gateway. All activities other than launching the
>>>>>> apps seem to work fine.
>>>>>>
>>>>>> My question: is there another port I need to open? Anybody is aware
>>>>>> of issues running/accessing Apex behind a proxy or firewall?
>>>>>>
>>>>>> Unfortunately the UI does not provide much information. I am
>>>>>> attaching some screenshots.
>>>>>>
>>>>>> Thanks,
>>>>>> Max.
>>>>>>
>>>>>
>>>>
>>>
>>>
>>> --
>>>
>>> Regards,
>>> Ashwin.
>>>
>>
>>
>>
>> --
>>
>> Regards,
>> Ashwin.
>>
>
>


-- 

Regards,
Ashwin.

Re: Apex with SSH Tunel

[Ashwin Chandra Putta]

Also, there are write permission errors on /user/dtadmin/datatorrent in
hdfs. Please make dtadmin user own /user/dtadmin/

Permission denied: user=dtadmin, access=WRITE,
inode="/user/dtadmin/datatorrent":hduser:supergroup:drwxr-xr-x

Regards,
Ashwin.

On Fri, Nov 18, 2016 at 11:27 AM, Ashwin Chandra Putta <
ashwinchand...@gmail.com> wrote:

> The end  tag is missing between line 30 and 31. It is for the
> property dt.attr.DEBUG.
>
> Regards,
> Ashwin.
>
> On Fri, Nov 18, 2016 at 10:16 AM, Max Bridgewater <
> max.bridgewa...@gmail.com> wrote:
>
>> Here is the log folder. Note that it refers to a malformed
>> properties.xml. I am attaching that properties file as well.
>>
>> On Fri, Nov 18, 2016 at 1:08 PM, Ashwin Chandra Putta <
>> ashwinchand...@gmail.com> wrote:
>>
>>> Max,
>>>
>>> Can you share the gateway logs?
>>>
>>> You will find them under /var/log/datatorrent for global install, or
>>> under ~/.dt/logs for local install.
>>>
>>> Regards,
>>> Ashwin.
>>>
>>> Regards,
>>> Ashwin.
>>>
>>> On Nov 18, 2016 9:41 AM, "Max Bridgewater" <max.bridgewa...@gmail.com>
>>> wrote:
>>>
>>>> Hi Folks,
>>>>
>>>> I am playing with Apex (DataTorrent RTS Enterprise). Local deployment
>>>> in Ubuntu 16 box works fine.
>>>>
>>>> However, when I deploy on a remote host, I am not apple to launch demo
>>>> applications. My suspicion is that this is due to having to open an SSH
>>>> tunnel to access the gateway. All activities other than launching the apps
>>>> seem to work fine.
>>>>
>>>> My question: is there another port I need to open? Anybody is aware of
>>>> issues running/accessing Apex behind a proxy or firewall?
>>>>
>>>> Unfortunately the UI does not provide much information. I am attaching
>>>> some screenshots.
>>>>
>>>> Thanks,
>>>> Max.
>>>>
>>>
>>
>
>
> --
>
> Regards,
> Ashwin.
>



-- 

Regards,
Ashwin.

Re: Apex with SSH Tunel

[Ashwin Chandra Putta]

Max,

Can you share the gateway logs?

You will find them under /var/log/datatorrent for global install, or under
~/.dt/logs for local install.

Regards,
Ashwin.

Regards,
Ashwin.

On Nov 18, 2016 9:41 AM, "Max Bridgewater" 
wrote:

> Hi Folks,
>
> I am playing with Apex (DataTorrent RTS Enterprise). Local deployment in
> Ubuntu 16 box works fine.
>
> However, when I deploy on a remote host, I am not apple to launch demo
> applications. My suspicion is that this is due to having to open an SSH
> tunnel to access the gateway. All activities other than launching the apps
> seem to work fine.
>
> My question: is there another port I need to open? Anybody is aware of
> issues running/accessing Apex behind a proxy or firewall?
>
> Unfortunately the UI does not provide much information. I am attaching
> some screenshots.
>
> Thanks,
> Max.
>

Re: Operators stay in PENDING_DEPLOY state

[Ashwin Chandra Putta]

Dimple,

Cluster does not have enough resources. Here is the initial container
request set from the log.

2016-11-04 15:06:26,901 INFO
com.datatorrent.stram.StreamingAppMasterService: Asking RM for
containers: [Capability[]Priority[0],
Capability[]Priority[1],
Capability[]Priority[2],
Capability[]Priority[3],
Capability[]Priority[4],
Capability[]Priority[5],
Capability[]Priority[6],
Capability[]Priority[7],
Capability[]Priority[8],
Capability[]Priority[9]]

AM then keeps asking for the following 4 containers every 30 seconds but
never gets.

2016-11-04 15:07:01,132 INFO
com.datatorrent.stram.StreamingAppMasterService: Asking RM for
containers: [Capability[]Priority[7],
Capability[]Priority[8],
Capability[]Priority[9],
Capability[]Priority[6]]

Regards,
Ashwin.

On Fri, Nov 4, 2016 at 3:28 PM, Dimple Patel 
wrote:

> Hello,
>
> I am trying to run Top N Word application from DataTorrent's  documentation
>   .
> After running the application on DataTorrent's console, when I monitor it
> in
> dtDashboard, the app's status is running but all my operators are in
> PENDING_DEPLOY state and are not switching to Running state. Attached
> herewith is my AppMaster log. I think the problem is with memory allocation
> in all operators as when I run a simple random number generation app, it's
> operators run fine. But I am not entirely sure what exactly the problem is.
>
> What is the problem here?
>
> dt.log  n1110/dt.log>
>
> Thanks,
> Dimple
>
>
>
> --
> View this message in context: http://apache-apex-users-list.
> 78494.x6.nabble.com/Operators-stay-in-PENDING-DEPLOY-state-tp1110.html
> Sent from the Apache Apex Users list mailing list archive at Nabble.com.
>



-- 

Regards,
Ashwin.

Re: How to use DataTorrent

[Ashwin Chandra Putta]

Dimple,

You don't need to use dtAssemble. You can use java to write the the
applications. See example tutorials:
https://github.com/DataTorrent/examples/tree/master/tutorials

Regards,
Ashwin.

On Thu, Nov 3, 2016 at 4:01 PM, dimple  wrote:

> Is it necessary to use dtAssemble, or I can get away with just Java?
>
> Thanks,
> Dimple
>
>
>
> --
> View this message in context: http://apache-apex-users-list.
> 78494.x6.nabble.com/How-to-use-DataTorrent-tp1102.html
> Sent from the Apache Apex Users list mailing list archive at Nabble.com.
>



-- 

Regards,
Ashwin.

Re: Retry functionality in Datatorrent

[Ashwin Chandra Putta]

It has hadoop dependencies in the list. Please exclude them. Also, it is
always good to exclude all the dependencies that you do not need or that
are already available in hadoop classpath.

Run the maven dependency tree command to check the transitive dependencies
and exclude the hadoop related dependencies from there.

mvn dependency:tree

example: https://github.com/apache/apex-malhar/blob/master/contrib/pom.xml

Regards,
Ashwin.


On Wed, Oct 19, 2016 at 1:54 PM, Jaspal Singh <jaspal.singh1...@gmail.com>
wrote:

> Attached is the list of jars inside target/deps directory.
>
>
> Thanks!!
>
> On Wed, Oct 19, 2016 at 3:48 PM, Ashwin Chandra Putta <
> ashwinchand...@gmail.com> wrote:
>
>> Ok, versions seem compatible. Can you do an ls on the target/deps/
>> directory and send the list of included jars?
>>
>> Regards,
>> Ashwin.
>>
>> On Wed, Oct 19, 2016 at 1:12 PM, Bandaru, Srinivas <
>> srinivas.band...@optum.com> wrote:
>>
>>> Mapr RD version:
>>>
>>> org.apache.hbase
>>>   hbase-client
>>>   *1.1.1-mapr-1602*
>>>
>>>
>>>
>>>
>>>
>>> Hadoop version : hadoop-2.7.0 (Mapr distribution 5.1)
>>>
>>>
>>>
>>> Thanks,
>>>
>>> Srinivas
>>>
>>>
>>>
>>> *From:* Ashwin Chandra Putta [mailto:ashwinchand...@gmail.com]
>>> *Sent:* Wednesday, October 19, 2016 3:05 PM
>>>
>>> *To:* users@apex.apache.org
>>> *Subject:* Re: Retry functionality in Datatorrent
>>>
>>>
>>>
>>> What is your HBase client version? Also, what is the version of hadoop
>>> on your cluster?
>>>
>>>
>>>
>>> Make sure that the HBase client version is compatible with the version
>>> of hadoop you are running.
>>>
>>>
>>>
>>> Regards,
>>>
>>> Ashwin.
>>>
>>>
>>>
>>> On Wed, Oct 19, 2016 at 12:54 PM, Bandaru, Srinivas <
>>> srinivas.band...@optum.com> wrote:
>>>
>>> We have a Hbase client dependency and do not have any Hadoop
>>> dependencies in project.
>>>
>>>
>>>
>>> *From:* Ashwin Chandra Putta [mailto:ashwinchand...@gmail.com]
>>> *Sent:* Wednesday, October 19, 2016 2:40 PM
>>> *To:* users@apex.apache.org
>>> *Subject:* Re: Retry functionality in Datatorrent
>>>
>>>
>>>
>>> Jaspal,
>>>
>>>
>>>
>>> Ensure that the app package is not including any hadoop jars as
>>> dependencies. Check target/deps/
>>>
>>>
>>>
>>> You can exclude those dependencies in the maven pom file.
>>>
>>>
>>>
>>> Regards,
>>>
>>> Ashwin.
>>>
>>>
>>>
>>> On Wed, Oct 19, 2016 at 12:09 PM, Jaspal Singh <
>>> jaspal.singh1...@gmail.com> wrote:
>>>
>>> Hi Ashwin, That's correct flow !!
>>>
>>> Team - ANother thing we are receiving below exception in the logs and
>>> application is FAILING when we are trying to make an update to HBase table.
>>> Any idea what could be the reason ?
>>>
>>> Container: container_e32_1476503307399_0172_02_01 on
>>> dbslt0079.uhc.com_8091
>>>
>>> 
>>> =
>>>
>>> LogType:AppMaster.stderr
>>>
>>> Log Upload Time:Wed Oct 19 11:35:36 -0500 2016
>>>
>>> LogLength:1246
>>>
>>> Log Contents:
>>>
>>> SLF4J: Class path contains multiple SLF4J bindings.
>>>
>>> SLF4J: Found binding in [jar:file:/opt/mapr/tmp/hadoop
>>> -mapr/nm-local-dir/usercache/mapr/appcache/application_14765
>>> 03307399_0172/filecache/26/slf4j-log4j12-1.7.5.jar!/org/
>>> slf4j/impl/StaticLoggerBinder.class]
>>>
>>> SLF4J: Found binding in [jar:file:/opt/mapr/lib/slf4j-
>>> log4j12-1.7.12.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>>>
>>> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
>>> explanation.
>>>
>>> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
>>>
>>> Exception in thread "main" java.lang.IllegalArgumentException: Invalid
>>> ContainerId: container_e32_1476503307399_0172_02_01
>>>
>>> at org.apache.hadoop.yarn.util.ConverterUtils.toContainerId(Con
&

Re: Retry functionality in Datatorrent

[Ashwin Chandra Putta]

Ok, versions seem compatible. Can you do an ls on the target/deps/
directory and send the list of included jars?

Regards,
Ashwin.

On Wed, Oct 19, 2016 at 1:12 PM, Bandaru, Srinivas <
srinivas.band...@optum.com> wrote:

> Mapr RD version:
>
> org.apache.hbase
>   hbase-client
>   *1.1.1-mapr-1602*
>
>
>
>
>
> Hadoop version : hadoop-2.7.0 (Mapr distribution 5.1)
>
>
>
> Thanks,
>
> Srinivas
>
>
>
> *From:* Ashwin Chandra Putta [mailto:ashwinchand...@gmail.com]
> *Sent:* Wednesday, October 19, 2016 3:05 PM
>
> *To:* users@apex.apache.org
> *Subject:* Re: Retry functionality in Datatorrent
>
>
>
> What is your HBase client version? Also, what is the version of hadoop on
> your cluster?
>
>
>
> Make sure that the HBase client version is compatible with the version of
> hadoop you are running.
>
>
>
> Regards,
>
> Ashwin.
>
>
>
> On Wed, Oct 19, 2016 at 12:54 PM, Bandaru, Srinivas <
> srinivas.band...@optum.com> wrote:
>
> We have a Hbase client dependency and do not have any Hadoop dependencies
> in project.
>
>
>
> *From:* Ashwin Chandra Putta [mailto:ashwinchand...@gmail.com]
> *Sent:* Wednesday, October 19, 2016 2:40 PM
> *To:* users@apex.apache.org
> *Subject:* Re: Retry functionality in Datatorrent
>
>
>
> Jaspal,
>
>
>
> Ensure that the app package is not including any hadoop jars as
> dependencies. Check target/deps/
>
>
>
> You can exclude those dependencies in the maven pom file.
>
>
>
> Regards,
>
> Ashwin.
>
>
>
> On Wed, Oct 19, 2016 at 12:09 PM, Jaspal Singh <jaspal.singh1...@gmail.com>
> wrote:
>
> Hi Ashwin, That's correct flow !!
>
> Team - ANother thing we are receiving below exception in the logs and
> application is FAILING when we are trying to make an update to HBase table.
> Any idea what could be the reason ?
>
> Container: container_e32_1476503307399_0172_02_01 on
> dbslt0079.uhc.com_8091
>
> 
> =
>
> LogType:AppMaster.stderr
>
> Log Upload Time:Wed Oct 19 11:35:36 -0500 2016
>
> LogLength:1246
>
> Log Contents:
>
> SLF4J: Class path contains multiple SLF4J bindings.
>
> SLF4J: Found binding in [jar:file:/opt/mapr/tmp/hadoop-mapr/nm-local-dir/
> usercache/mapr/appcache/application_1476503307399_0172/filecache/26/slf4j-
> log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>
> SLF4J: Found binding in [jar:file:/opt/mapr/lib/slf4j-
> log4j12-1.7.12.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
>
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
>
> Exception in thread "main" java.lang.IllegalArgumentException: Invalid
> ContainerId: container_e32_1476503307399_0172_02_01
>
> at org.apache.hadoop.yarn.util.ConverterUtils.toContainerId(
> ConverterUtils.java:182)
>
> at com.datatorrent.stram.StreamingAppMaster.main(
> StreamingAppMaster.java:91)
>
> Caused by: java.lang.NumberFormatException: For input string: "e32"
>
> at java.lang.NumberFormatException.forInputString(
> NumberFormatException.java:65)
>
> at java.lang.Long.parseLong(Long.java:589)
>
> at java.lang.Long.parseLong(Long.java:631)
>
> at org.apache.hadoop.yarn.util.ConverterUtils.
> toApplicationAttemptId(ConverterUtils.java:137)
>
> at org.apache.hadoop.yarn.util.ConverterUtils.toContainerId(
> ConverterUtils.java:177)
>
> ... 1 more
>
> End of LogType:AppMaster.stderr
>
>
>
> Thanks!!
>
>
>
> On Tue, Oct 18, 2016 at 7:47 PM, Ashwin Chandra Putta <
> ashwinchand...@gmail.com> wrote:
>
> Jaspal,
>
>
>
> If I understand this correctly, the flow looks like this.
>
>
>
> REST API --> Kafka topic --> Apex Kafka Input Operator.
>
>
>
> If this is the flow, then the kafka input operator should be reading
> messages from Kafka without losing them. There is no retry attempts
> necessary.
>
>
> Let me know if the understanding of the flow is incorrect.
>
>
>
> Regards,
>
> Ashwin.
>
>
>
> On Tue, Oct 18, 2016 at 2:49 PM, Jaspal Singh <jaspal.singh1...@gmail.com>
> wrote:
>
> Hi Team,
>
> We are pushing messages from Kafka to Datatorrent application using REST
> API service as a producer. If say due to some issue with the service the
> message couldn't be pushed/processed so we want to "retry it for n times"
> before it is dropped.
>
>

Re: Retry functionality in Datatorrent

[Ashwin Chandra Putta]

What is your HBase client version? Also, what is the version of hadoop on
your cluster?

Make sure that the HBase client version is compatible with the version of
hadoop you are running.

Regards,
Ashwin.

On Wed, Oct 19, 2016 at 12:54 PM, Bandaru, Srinivas <
srinivas.band...@optum.com> wrote:

> We have a Hbase client dependency and do not have any Hadoop dependencies
> in project.
>
>
>
> *From:* Ashwin Chandra Putta [mailto:ashwinchand...@gmail.com]
> *Sent:* Wednesday, October 19, 2016 2:40 PM
> *To:* users@apex.apache.org
> *Subject:* Re: Retry functionality in Datatorrent
>
>
>
> Jaspal,
>
>
>
> Ensure that the app package is not including any hadoop jars as
> dependencies. Check target/deps/
>
>
>
> You can exclude those dependencies in the maven pom file.
>
>
>
> Regards,
>
> Ashwin.
>
>
>
> On Wed, Oct 19, 2016 at 12:09 PM, Jaspal Singh <jaspal.singh1...@gmail.com>
> wrote:
>
> Hi Ashwin, That's correct flow !!
>
> Team - ANother thing we are receiving below exception in the logs and
> application is FAILING when we are trying to make an update to HBase table.
> Any idea what could be the reason ?
>
> Container: container_e32_1476503307399_0172_02_01 on
> dbslt0079.uhc.com_8091
>
> 
> =
>
> LogType:AppMaster.stderr
>
> Log Upload Time:Wed Oct 19 11:35:36 -0500 2016
>
> LogLength:1246
>
> Log Contents:
>
> SLF4J: Class path contains multiple SLF4J bindings.
>
> SLF4J: Found binding in [jar:file:/opt/mapr/tmp/hadoop-mapr/nm-local-dir/
> usercache/mapr/appcache/application_1476503307399_0172/filecache/26/slf4j-
> log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>
> SLF4J: Found binding in [jar:file:/opt/mapr/lib/slf4j-
> log4j12-1.7.12.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
>
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
>
> Exception in thread "main" java.lang.IllegalArgumentException: Invalid
> ContainerId: container_e32_1476503307399_0172_02_01
>
> at org.apache.hadoop.yarn.util.ConverterUtils.toContainerId(
> ConverterUtils.java:182)
>
> at com.datatorrent.stram.StreamingAppMaster.main(
> StreamingAppMaster.java:91)
>
> Caused by: java.lang.NumberFormatException: For input string: "e32"
>
> at java.lang.NumberFormatException.forInputString(
> NumberFormatException.java:65)
>
> at java.lang.Long.parseLong(Long.java:589)
>
> at java.lang.Long.parseLong(Long.java:631)
>
> at org.apache.hadoop.yarn.util.ConverterUtils.
> toApplicationAttemptId(ConverterUtils.java:137)
>
> at org.apache.hadoop.yarn.util.ConverterUtils.toContainerId(
> ConverterUtils.java:177)
>
> ... 1 more
>
> End of LogType:AppMaster.stderr
>
>
>
> Thanks!!
>
>
>
> On Tue, Oct 18, 2016 at 7:47 PM, Ashwin Chandra Putta <
> ashwinchand...@gmail.com> wrote:
>
> Jaspal,
>
>
>
> If I understand this correctly, the flow looks like this.
>
>
>
> REST API --> Kafka topic --> Apex Kafka Input Operator.
>
>
>
> If this is the flow, then the kafka input operator should be reading
> messages from Kafka without losing them. There is no retry attempts
> necessary.
>
>
> Let me know if the understanding of the flow is incorrect.
>
>
>
> Regards,
>
> Ashwin.
>
>
>
> On Tue, Oct 18, 2016 at 2:49 PM, Jaspal Singh <jaspal.singh1...@gmail.com>
> wrote:
>
> Hi Team,
>
> We are pushing messages from Kafka to Datatorrent application using REST
> API service as a producer. If say due to some issue with the service the
> message couldn't be pushed/processed so we want to "retry it for n times"
> before it is dropped.
>
> Is there any retry functionality built within Datatorrent or we have write
> some code logic for the same ??
>
> Thanks!!
>
>
>
>
>
> --
>
>
>
> Regards,
>
> Ashwin.
>
>
>
>
>
>
>
> --
>
>
>
> Regards,
>
> Ashwin.
>
>
> This e-mail, including attachments, may include confidential and/or
> proprietary information, and may be used only by the person or entity
> to which it is addressed. If the reader of this e-mail is not the intended
> recipient or his or her authorized agent, the reader is hereby notified
> that any dissemination, distribution or copying of this e-mail is
> prohibited. If you have received this e-mail in error, please notify the
> sender by replying to this message and delete this e-mail immediately.
>



-- 

Regards,
Ashwin.

Re: Retry functionality in Datatorrent

[Ashwin Chandra Putta]

Jaspal,

Ensure that the app package is not including any hadoop jars as
dependencies. Check target/deps/

You can exclude those dependencies in the maven pom file.

Regards,
Ashwin.

On Wed, Oct 19, 2016 at 12:09 PM, Jaspal Singh <jaspal.singh1...@gmail.com>
wrote:

> Hi Ashwin, That's correct flow !!
>
> Team - ANother thing we are receiving below exception in the logs and
> application is FAILING when we are trying to make an update to HBase table.
> Any idea what could be the reason ?
>
> Container: container_e32_1476503307399_0172_02_01 on
> dbslt0079.uhc.com_8091
>
> 
> =
>
> LogType:AppMaster.stderr
>
> Log Upload Time:Wed Oct 19 11:35:36 -0500 2016
>
> LogLength:1246
>
> Log Contents:
>
> SLF4J: Class path contains multiple SLF4J bindings.
>
> SLF4J: Found binding in [jar:file:/opt/mapr/tmp/hadoop-mapr/nm-local-dir/
> usercache/mapr/appcache/application_1476503307399_0172/filecache/26/slf4j-
> log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>
> SLF4J: Found binding in [jar:file:/opt/mapr/lib/slf4j-
> log4j12-1.7.12.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
>
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
>
> Exception in thread "main" java.lang.IllegalArgumentException: Invalid
> ContainerId: container_e32_1476503307399_0172_02_01
>
> at org.apache.hadoop.yarn.util.ConverterUtils.toContainerId(
> ConverterUtils.java:182)
>
> at com.datatorrent.stram.StreamingAppMaster.main(
> StreamingAppMaster.java:91)
>
> Caused by: java.lang.NumberFormatException: For input string: "e32"
>
> at java.lang.NumberFormatException.forInputString(
> NumberFormatException.java:65)
>
> at java.lang.Long.parseLong(Long.java:589)
>
> at java.lang.Long.parseLong(Long.java:631)
>
> at org.apache.hadoop.yarn.util.ConverterUtils.
> toApplicationAttemptId(ConverterUtils.java:137)
>
> at org.apache.hadoop.yarn.util.ConverterUtils.toContainerId(
> ConverterUtils.java:177)
>
> ... 1 more
>
> End of LogType:AppMaster.stderr
>
>
> Thanks!!
>
> On Tue, Oct 18, 2016 at 7:47 PM, Ashwin Chandra Putta <
> ashwinchand...@gmail.com> wrote:
>
>> Jaspal,
>>
>> If I understand this correctly, the flow looks like this.
>>
>> REST API --> Kafka topic --> Apex Kafka Input Operator.
>>
>> If this is the flow, then the kafka input operator should be reading
>> messages from Kafka without losing them. There is no retry attempts
>> necessary.
>>
>> Let me know if the understanding of the flow is incorrect.
>>
>> Regards,
>> Ashwin.
>>
>> On Tue, Oct 18, 2016 at 2:49 PM, Jaspal Singh <jaspal.singh1...@gmail.com
>> > wrote:
>>
>>> Hi Team,
>>>
>>> We are pushing messages from Kafka to Datatorrent application using REST
>>> API service as a producer. If say due to some issue with the service the
>>> message couldn't be pushed/processed so we want to "retry it for n times"
>>> before it is dropped.
>>>
>>> Is there any retry functionality built within Datatorrent or we have
>>> write some code logic for the same ??
>>>
>>>
>>> Thanks!!
>>>
>>
>>
>>
>> --
>>
>> Regards,
>> Ashwin.
>>
>
>


-- 

Regards,
Ashwin.

Re: balanced of Stream Codec

[Ashwin Chandra Putta]

Sunil,

For key based partitioning, the getPartition method is supposed to return a
consistent integer representing the key for partitioning. Typically the
java hashCode of the key. The tuples are then routed based on the integer
and looking at its lower bits on the mask (number of lower bits) based on
number of downstream partitions.

A partition key with only 100 possible values may not load balance properly
and will most likely introduce skew, as in your case. It is recommended to
use a key that will generate a high range of values for the return value.

Regards,
Ashwin.

On Fri, Oct 14, 2016 at 1:17 PM, Sunil Parmar 
wrote:

> We’re using Stream codec to consistently / parallel processing of the data
> across the operator partitions. Our requirement is to serialize processing
> of the data based on particular tuple attribute let’s call it
> ‘catagory_name’ . In order to achieve the parallel processing of different
> category names we’re written our stream codec as following.
>
>public class CatagoryStreamCodec extends 
> KryoSerializableStreamCodec
> {
>
> private static final long serialVersionUID = -687991492884005033L;
>
>
>
> @Override
>
> public int getPartition(Object in) {
>
> try {
>
> InputTuple tuple = (InputTuple) in;
>
> String partitionKehy = tuple.getName();
>
> if(partitionKehy != null) {
>
> return partitionKehy.hashCode();
>
> }
>
> }
>}
>
> It’s working as expected *but *we observed inconsistent partitions when
> we run this in production env with 20 partitioner of the operator following
> the codec in the dag.
>
>- Some operator instance didn’t process any data
>- Some operator instance process as many tuples as combined everybody
>else
>
>
> Questions :
>
>- getPartition method supposed to return the actual partition or just
>some lower bit used for deciding partition ?
>- Number of partitions is known to application properties and can vary
>between deployments or environments. Is it best practice to use that
>property in the stream codec ?
>- Any recommended hash function for getting consistent variations in
>the lower bit with less variety of data. we’ve ~100+ categories and I’m
>thinking to have 10+ operator partitions.
>
>
> Thanks,
> Sunil
>



-- 

Regards,
Ashwin.

Re: Datatorrent UI capabilities

[Ashwin Chandra Putta]

Hey Srinivas,

I will ask someone from DataTorrent to get in touch with you regarding the
details. This forum is mainly for Apex related discussion which includes
the open source Apex Core and Apex Malhar operator library. So DataTorrent
UI discussion is usually outside the scope of this mailing list.

Regards,
Ashwin.

On Tue, Oct 11, 2016 at 12:39 PM, Bandaru, Srinivas <
srinivas.band...@optum.com> wrote:

> Hi, Need some help with building Datatorrent application and before
> writing a code in java, wanted to understand the Datatorrent UI
> capabilities.
>
> Use case: Maprstream producer emits messages and processing the messages
> using Datatorrent application, Based on the DT application logic divide the
> messages into two different groups and write it to two different Mapr strem
> topics.
>
> Is there any UI capability I can build application without writing a
> code?? I am just trying to see DT application UI capabilities.
>
>
>
> Thanks,
>
> Srinivas Bandaru
>
>
> This e-mail, including attachments, may include confidential and/or
> proprietary information, and may be used only by the person or entity
> to which it is addressed. If the reader of this e-mail is not the intended
> recipient or his or her authorized agent, the reader is hereby notified
> that any dissemination, distribution or copying of this e-mail is
> prohibited. If you have received this e-mail in error, please notify the
> sender by replying to this message and delete this e-mail immediately.
>



-- 

Regards,
Ashwin.

Re: dag.addStream Error

[Ashwin Chandra Putta]

Is there only one KeyValue class in the classpath?

Regards,
Ashwin.

On Mon, Aug 15, 2016 at 11:38 AM, Sanjay Pujare 
wrote:

> This should work and I don’t see anything obviously wrong.
>
>
>
> Where is KeyValue defined and are you sure the same type is used
> in both places?
>
>
>
> *From: *"Mukkamula, Suryavamshivardhan (CWM-NR)" <
> suryavamshivardhan.mukkam...@rbc.com>
> *Reply-To: *
> *Date: *Monday, August 15, 2016 at 10:40 AM
> *To: *"users@apex.apache.org" 
> *Subject: *dag.addStream Error
>
>
>
> Hi,
>
>
>
> Can you please help me resolve the below error at the highlighted portion?
> The port types are matching but still I get the below error, let me know If
> I am missing anything?
>
>
>
> ##Error Message###
>
> The method addStream(String, Operator.OutputPort,
> Operator.InputPort...) in the type DAG is not applicable for the
> arguments (String, DefaultOutputPort>,
> DefaultInputPort>)
>
>
>
> ###Application Dag ##
>
>
>
> InputFileReader inputFileStreamReader = dag.addOperator("
> inputFileStreamReader", InputFileReader.*class*);
>
> GroupByteArrayParser silverParser = dag.addOperator("silverParser",
> GroupByteArrayParser.*class*);
>
>
>
> dag.addStream("file stream to silver parser",
> inputFileStreamReader.output, silverParser.inputPort);
>
>
>
> Oper1 code###
>
> *public* *class* InputFileReader *extends* BaseOperator{
>
> *private* *static* Logger *LOG* = LoggerFactory.*getLogger*(InputF
> ileReader.*class*);
>
>
>
> *public* *transient* DefaultOutputPort>
> output = *new* DefaultOutputPort>();
>
>
>
>
>
> oper2 code#
>
> *public* *class* GroupByteArrayParser *extends* BaseOperator {
>
>
>
>
>
> *public* *final* *transient* DefaultInputPort>
> inputPort = *new* DefaultInputPort>() {
>
>
>
> @Override
>
> *public* *void* process(KeyValue tuple) {
>
>
>
> ___
>
> If you received this email in error, please advise the sender (by return
> email or otherwise) immediately. You have consented to receive the attached
> electronically at the above-noted email address; please retain a copy of
> this confirmation for future reference.
>
> Si vous recevez ce courriel par erreur, veuillez en aviser l'expéditeur
> immédiatement, par retour de courriel ou par un autre moyen. Vous avez
> accepté de recevoir le(s) document(s) ci-joint(s) par voie électronique à
> l'adresse courriel indiquée ci-dessus; veuillez conserver une copie de
> cette confirmation pour les fins de reference future.
>



-- 

Regards,
Ashwin.

RE: Apex Finished Status

[Ashwin Chandra Putta]

Doyle,

Have you thrown it in all input operators? It starts from the operator you
throw it in and propagates downstream. All operators need to shutdown
before application changes state.

Regards,
Ashwin.

On Aug 11, 2016 5:01 AM, "Doyle, Austin O." 
wrote:

> The ShutdownException does stop the operator but doesn’t seem to update
> the state of the application, still in the RUNNING state.  Is there
> something specific I need to put in the endwindows or teardowns?
>
>
>
> *From:* Chinmay Kolhatkar [mailto:chin...@datatorrent.com]
> *Sent:* Thursday, August 11, 2016 2:12 AM
> *To:* users@apex.apache.org
> *Subject:* Re: Apex Finished Status
>
>
>
> Just one additional information to what Sandesh said. ShutdownException
> shuts down the operator from which its thrown and all downstream operators
> to that operator.
>
> Hence if you want the DAG to shutdown completely, you need to throw
> ShutdownException from all *input* operators.
>
>
>
> -Chinmay.
>
>
>
>
>
> On Thu, Aug 11, 2016 at 9:58 AM, Sandesh Hegde 
> wrote:
>
> Hello Austin,
>
>
>
> Throwing ShutdownException  from the operators will shutdown the Apex
> Application.
>
>
>
>
>
> On Wed, Aug 10, 2016 at 8:56 AM Doyle, Austin O. <
> austin.o.doyl...@leidos.com> wrote:
>
> When would an Apex Application return a FINISHED status from YARN as
> opposed to continually being in the RUNNING state?
>
> Is there a way to force the YARN status to FINISHED from within the
> application when processing is complete?
>
>
>

Re: HDFS Write File Operator struggling to start

[Ashwin Chandra Putta]

Can you check in the logs the first time the issue occurred for what
triggered it? Look for the first container that failed.

Regards,
Ashwin

On Fri, Aug 5, 2016 at 12:58 PM, Raja.Aravapalli <raja.aravapa...@target.com
> wrote:

>
> Hi Ashwin,
>
> It happens when the application is running!!
>
>
> Regards,
> Raja.
>
> From: Ashwin Chandra Putta <ashwinchand...@gmail.com>
> Reply-To: "users@apex.apache.org" <users@apex.apache.org>
> Date: Thursday, August 4, 2016 at 4:42 PM
> To: "users@apex.apache.org" <users@apex.apache.org>
> Subject: Re: HDFS Write File Operator struggling to start
>
> Raja,
>
> When do you see this issue? Does it happen while the application is
> running? Does this happen while restarting a failed application? Or does
> this happen while starting a new application?
>
> Regards,
> Ashwin.
>
> On Thu, Aug 4, 2016 at 11:25 AM, Samba Surabhi <sasur...@outlook.com>
> wrote:
>
>> If it is the issue with file size, you can rotate the output file.
>>
>> writer.setAlwaysWriteToTmp(true);
>>
>> writer.setRotationWindows(240);
>>
>> Thanks,
>>
>> Samba Surabhi.
>>
>>
>> --
>> From: raja.aravapa...@target.com
>> To: users@apex.apache.org
>> Subject: HDFS Write File Operator struggling to start
>> Date: Thu, 4 Aug 2016 14:49:16 +
>>
>>
>>
>> Hi
>>
>> I have a HDFS file write operator in my DAG, which is failing to start a
>> new operator and keep on trying to start one!!
>>
>> It created approx. 800 temporary files in the destination hdfs
>> directory!! How can I fix this issue…. And debug the root cause…. ?
>>
>> All I can see in container log is, File corrupted message…!!
>>
>>
>> Can someone please help me fix this ?
>>
>>
>> Regards,
>> Raja.
>>
>
>
>
> --
>
> Regards,
> Ashwin.
>



-- 

Regards,
Ashwin.

Re: Apache Apex and Dip blog

[Ashwin Chandra Putta]

Created PR to add blog link: https://github.com/apache/apex-site/pull/46

Regards,
Ashwin.

On Mon, Aug 1, 2016 at 5:51 PM, Pramod Immaneni 
wrote:

> Hey Neeraj,
>
> Nice blog. It would be great to link to it in the apex website blog
> section in this page
>
> https://apex.apache.org/docs.html
>
> On Mon, Aug 1, 2016 at 2:09 PM, Neeraj Sabharwal 
> wrote:
>
>>
>>
>>
>> https://techblog.xavient.com/real-time-data-ingestion-dip-apache-apex-co-dev-opportunity/
>>
>> *Neeraj Sabharwal*
>>
>> Director, Cloud and Big Data Solutions
>>
>> *X**avient Information Systems   *
>>
>> *Great Ideas…Simple Solutions*
>>
>> Email:  nsabhar...@xavient.com
>>
>> Mobile:5715995087
>>
>> Website:  www.xavient.com
>>
>> Address:  13800 Coppermine Road, Suite 200, Herndon, VA 20171
>> ***DISCLAIMER*** This communication & accompanying documents ("this
>> e-mail") contains confidential and/or privileged information for exclusive
>> use of the individual to whom it is addressed. If you are not the intended
>> recipient, please immediately notify the company & delete this e-mail. Any
>> unauthorized use or disclosure of this e-mail is strictly prohibited.
>> Representations in this e-mail are subject to contract. As an e-mail user
>> please be cautious of the technical & other vulnerabilities of the internet
>> which may result in malicious and/or unauthorized access to / use /
>> alteration of e-mails/e-mail IDs. Thank you.
>>
>
>


-- 

Regards,
Ashwin.

Re: Sharing Properties Between Operators

[Ashwin Chandra Putta]

Alex,

I tried to give regular expression on the partial operator name and it
works.

Eg:


dt.application.*.operator.hazel[\w].prop.host
127.0.0.1


Regards,
Ashwin.

On Mon, Aug 1, 2016 at 2:26 PM, David Yan  wrote:

> Hi Alex,
>
> You can make multiple operators sharing the same property by doing
> something like this:
>
> 
>your.own.property.name
>127.0.0.1
> 
>
> 
>dt.operator.opName1.prop.[property-name]
>${your.own.property.name}
> 
>
> 
>dt.operator.opName2.prop.[property-name]
>${your.own.property.name}
> 
>
> Or you can specify your.own.property.name's value at launch time.
>
> David
>
> On Mon, Aug 1, 2016 at 12:16 PM, McCullough, Alex <
> alex.mccullo...@capitalone.com> wrote:
>
>> Hey,
>>
>>
>>
>> I have an input and processing operator that both rely on hazelcast. I
>> defined a base operator where the setup method and common properties are
>> defined, each have an abstract class that extends this base operator to
>> handle default input or processing logic along with unique properties at
>> this level, and then of course each has a concrete implementation.
>>
>>
>>
>> Is there a way the two can share properties in the properties.xml? I know
>> you can wildcard some property/attribute descriptors, but could I wildcard
>> part of an operator name? meaning having an operator called hazelInput and
>> one called hazelProcess and a property defined as:
>>
>>
>>
>> 
>> dt.application.*.operator.hazel*.prop.host
>> 127.0.0.1
>> 
>>
>>
>>
>> Thanks,
>>
>> Alex
>>
>> --
>>
>> The information contained in this e-mail is confidential and/or
>> proprietary to Capital One and/or its affiliates and may only be used
>> solely in performance of work or services for Capital One. The information
>> transmitted herewith is intended only for use by the individual or entity
>> to which it is addressed. If the reader of this message is not the intended
>> recipient, you are hereby notified that any review, retransmission,
>> dissemination, distribution, copying or other use of, or taking of any
>> action in reliance upon this information is strictly prohibited. If you
>> have received this communication in error, please contact the sender and
>> delete the material from your computer.
>>
>
>


-- 

Regards,
Ashwin.

Re: Sub Partitioning the parallel partitions

[Ashwin Chandra Putta]

Yogi,
What is the reason for avoiding unifiers?

Regards,
Ashwin.

On Mon, Jul 25, 2016 at 9:41 AM, Yogi Devendra 
wrote:

> Hi,
>
> I have a DAG A->B->C.
>
> 1. A is kafka input operator reading from 4 different topics configured
> with ONE_TO_ONE strategy. Thus creating 4 partitons of A.
> 2. B and C are configured to have parallel partitions w.r.t. their input
> port. Thus, currently both B, C have 4 partitions.
>
> I am observing B significant latency in B operator. Thus, would like to
> have 2 partitions for B per partition of A. Since, application is latency
> intensive, I want to avoid unifiers as far as possible.
>
> How to achieve this partitioning?
>
>
> ~ Yogi
>



-- 

Regards,
Ashwin.

Re: Operators not working

[Ashwin Chandra Putta]

When you say it does not emit anything, are you referring to your input
operator? Is it an operator from malhar? Which one?

And how many records do you have in the source table?

Regards,
Ashwin.

On Jul 14, 2016 3:55 PM, "Jaikit Jilka"  wrote:

> Hello,
>
> I am parsing data from one table to another table. My operators are
> running but it does not parse or emit anything after it parses 5 rows.
> I also partitioned my parser operator and jdbcoutput operator. It 5 it
> works fine and later it slows down and finally it does not emit anything.
> Latency of the operator is also around 120 - 150. What can be wrong here.
>
> Thank You,
> Jaikit JIlka
>

Re: Runtime property modifications for an Apex Application

[Ashwin Chandra Putta]

Raja,

You can set log levels for class path patterns for running applications
from gateway. Not sure if you can do it from CLI.

http://docs.datatorrent.com/configuration/#adjusting-logging-levels

Regards,
Ashwin.

On Tue, Jul 5, 2016 at 2:53 PM, Raja.Aravapalli 
wrote:

>
> Hi,
>
> Is it possible for me to set below *DEBUG *property for a running
> applicaion using *apex *command line tool, *without killing or shutting
> down* a running application ?
>
>  dt.attr.DEBUG true 
>
>
> If so can someone help me with the command, how I can achieve this ? Thank
> you.
>
>
>
> Regards,
> Raja.
>



-- 

Regards,
Ashwin.