Hi, Shekar,
This is probably what's missing:
private KeyValueStore<String, Integer> store;
public void init(Config config, TaskContext context) {
...
store = (KeyValueStore<String, Integer>) context.getStore("store");
...
}
And, it looks like your key and value data types for the store are
primitive types, why do you want to use JSON serde?
-Yi
On Wed, Jul 1, 2015 at 4:15 PM, Shekar Tippur <ctip...@gmail.com> wrote:
> Sorry .. That was a copy paste issue
>
> This is what I have
>
> stores.store.key.serde=org.apache.samza.serializers.JsonSerdeFactory
>
> stores.store.msg.serde=org.apache.samza.serializers.JsonSerdeFactory
>
> stores.store.changelog=argos.windowchangelog
>
>
> and this is how i am initializing it
>
> private KeyValueStore<String, Integer> store;
>
> On Wed, Jul 1, 2015 at 3:55 PM, Yan Fang <yanfang...@gmail.com> wrote:
>
> > So do you use the "store-name" as the kv storage name in your StreamTask
> > code?
> >
> > Fang, Yan
> > yanfang...@gmail.com
> >
> > On Wed, Jul 1, 2015 at 3:41 PM, Shekar Tippur <ctip...@gmail.com> wrote:
> >
> > > Yan,
> > >
> > > yes. I do have it.
> > >
> > > - Shekar
> > >
> > > On Wed, Jul 1, 2015 at 3:09 PM, Yan Fang <yanfang...@gmail.com> wrote:
> > >
> > > > Do you have
> > > >
> > > > serializers.registry.json.class
> > > > =org.apache.samza.serializers.JsonSerdeFactory
> > > >
> > > > in your config file?
> > > >
> > > >
> > > > Fang, Yan
> > > > yanfang...@gmail.com
> > > >
> > > > On Wed, Jul 1, 2015 at 2:59 PM, Shekar Tippur <ctip...@gmail.com>
> > wrote:
> > > >
> > > > > Yi/Milinda,
> > > > >
> > > > > I am trying to initialize a kv store. I have the following
> properties
> > > > > defined:
> > > > >
> > > > > stores.store-name.key.serde=json
> > > > >
> > > > > stores.store-name.msg.serde=json
> > > > >
> > > > > stores.store-name.changelog=argos.windowchangelog
> > > > > How do I define a key serde as I am getting this exception:
> > > > >
> > > > > Exception in thread "main" org.apache.samza.SamzaException: Must
> > > define a
> > > > > key serde when using key value storage.
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.samza.storage.kv.BaseKeyValueStorageEngineFactory$class.getStorageEngine(BaseKeyValueStorageEngineFactory.scala:86)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory.getStorageEngine(RocksDbKeyValueStorageEngineFactory.scala:28)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.samza.container.SamzaContainer$$anonfun$35$$anonfun$38.apply(SamzaContainer.scala:455)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.samza.container.SamzaContainer$$anonfun$35$$anonfun$38.apply(SamzaContainer.scala:439)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> > > > >
> > > > > at scala.collection.immutable.Map$Map1.foreach(Map.scala:109)
> > > > >
> > > > > at
> > > scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
> > > > >
> > > > > at scala.collection.AbstractTraversable.map(Traversable.scala:105)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.samza.container.SamzaContainer$$anonfun$35.apply(SamzaContainer.scala:439)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.samza.container.SamzaContainer$$anonfun$35.apply(SamzaContainer.scala:416)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> > > > >
> > > > > at scala.collection.immutable.Set$Set1.foreach(Set.scala:74)
> > > > >
> > > > > at
> > > scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> scala.collection.AbstractSet.scala$collection$SetLike$$super$map(Set.scala:47)
> > > > >
> > > > > at scala.collection.SetLike$class.map(SetLike.scala:93)
> > > > >
> > > > > at scala.collection.AbstractSet.map(Set.scala:47)
> > > > >
> > > > > at
> > > > >
> > > >
> > >
> >
> org.apache.samza.container.SamzaContainer$.apply(SamzaContainer.scala:416)
> > > > >
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.samza.job.local.ThreadJobFactory.getJob(ThreadJobFactory.scala:63)
> > > > >
> > > > > at org.apache.samza.job.JobRunner.run(JobRunner.scala:62)
> > > > >
> > > > > at org.apache.samza.job.JobRunner$.main(JobRunner.scala:37)
> > > > >
> > > > > at org.apache.samza.job.JobRunner.main(JobRunner.scala)
> > > > >
> > > > > On Mon, Jun 29, 2015 at 12:41 PM, Shekar Tippur <ctip...@gmail.com
> >
> > > > wrote:
> > > > >
> > > > > > Yi,
> > > > > >
> > > > > > My use case is more of the latter. Your explanation makes sense
> > now.
> > > I
> > > > > was
> > > > > > also looking into Milinda's wiki. She has a section for Kafka
> > > > > > partition SimplePartitioner, which is simple enough as well.
> > > > > >
> > > > > > Thanks for all the inputs. Let me see what I come up with while
> > > > > > implementing it.
> > > > > >
> > > > > > - Shekar
> > > > > >
> > > > > > On Mon, Jun 29, 2015 at 10:42 AM, Yi Pan <nickpa...@gmail.com>
> > > wrote:
> > > > > >
> > > > > >> Hi, Shekar,
> > > > > >>
> > > > > >> First, I would like to clarify what you meant by sliding window:
> > is
> > > it
> > > > > >> defined as windows with size N and advance step size of 1 (which
> > > means
> > > > > >> that
> > > > > >> windows overlap and each input message would contribute to
> > multiple
> > > > > counts
> > > > > >> in different windows)? Or windows with size N and advance step
> > size
> > > > of N
> > > > > >> (i.e. each incoming message only contribute to one counter in a
> > > single
> > > > > >> window)?
> > > > > >>
> > > > > >> If your use case falls into the first category, you will need
> > > > something
> > > > > >> more sophisticated as discussed in SAMZA-552. If your use case
> is
> > > the
> > > > > >> second one, there could be a simpler version of SAMZA-552 that
> you
> > > can
> > > > > go
> > > > > >> with:
> > > > > >>
> > > > > >> 1) Initiate a KV-store that uses the application name as the key
> > > > > >> 2) For each incoming message, look for the windows that the
> > message
> > > by
> > > > > the
> > > > > >> application name
> > > > > >> 3) Update the counter and update the value in the KV-store based
> > on
> > > > the
> > > > > >> application name
> > > > > >> 4) Every 5 min when window() method is triggered, set all
> counters
> > > to
> > > > > zero
> > > > > >> (this can be done in a lazy way as well, by keeping the last
> reset
> > > > > >> timestamp in the record in the KV-store, keyed by application
> > name.
> > > > > Then,
> > > > > >> resetting counter to zero can be done when next time the
> > application
> > > > > >> counter is updated again)
> > > > > >>
> > > > > >> Hope that makes sense.
> > > > > >>
> > > > > >> -Yi
> > > > > >>
> > > > > >> On Mon, Jun 29, 2015 at 10:06 AM, Shekar Tippur <
> > ctip...@gmail.com>
> > > > > >> wrote:
> > > > > >>
> > > > > >> > Benjamin,
> > > > > >> >
> > > > > >> > Thanks for the explanation. We dont have any specific
> partition
> > > > scheme
> > > > > >> as
> > > > > >> > yet. We just have 2 topics - raw and processed and we use
> > default
> > > > > >> > partitioning scheme.
> > > > > >> > Can you share any code snippet so I can understand it better?
> > > > > >> >
> > > > > >> > - Shekar
> > > > > >> >
> > > > > >>
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>
