Hi,
So, in short, the plan is:
- Implement write() that writes a PC<KV<byte[], byte[]>> to memcached. Can
be done as a simple ParDo with some batching to take advantage of multi-put
operation.
- Implement lookup() that converts a PC<byte[]> (keys) to PC<KV<byte[],
byte[]>> (keys paired with values). Can also be done as a simple ParDo with
some batching to take advantage of multi-get operation.
spymemcached takes care of everything else (e.g. distributing a batched
get/put onto the proper servers), so the code of the transform will be
basically trivial - which is great.
Correct?
On Thu, Jul 20, 2017 at 2:54 PM Seshadri Raghunathan <
sraghunat...@etouch.net> wrote:
> Thanks Lukasz, Eugene & Ismaël for your inputs.
>
> Please find below my comments on various aspects of this proposal -
>
> A. read / lookup - takes in a PCollection<key> and transforms it into a
> PCollection<KV<key, value>>
>
> -------------------------------------------------------------------------------------------------------------------------------------
>
> This is a simple lookup rather than a full read / scan.
>
> Splits -
> ---------
> Our idea is similar to Eugene & Ismaeel on splits. There is
> no concept of a split for a 'get' operation , internally the client
> API(spymemcached) calculates an hash for the key and the memcache server
> node mapping to that hashvalue is probed for lookup. spymemcached API
> supports a multi-get/lookup operation which takes in a bunch of keys,
> identifies specific server node (from server farm) for each of these keys
> and groups them by the server node they map to. The API also provides a way
> to enable consistent hashing. Each of these {server node - keys list} is
> grouped as an 'Operation' and enqueued to appropriate server nodes and the
> lookup is done in an asynchronous manner. reference -
> https://github.com/couchbase/spymemcached/blob/master/src/main/java/net/spy/memcached/MemcachedClient.java#L1274
> . All this is done under the hoood by spymemcached API. One way to achieve
> splitting explicitly would be to instantiate a separate spymemcached client
> for each of the server nodes and treat each of them as a separate split.
> However in this case the split doesn't make sense as for a given
> key/hashvalue we need not probe all the servers, simply probing the server
> node that the hashvalue maps to should suffice. Instead, considering a more
> granular split at a 'slab' level (per Lukasz inputs) by using lru_crawler
> metadump operations is another way to look at it. This approach may not be
> ideal for this operation as we could end up reading all the slabs as an
> overkill.
>
> Consistency -
> ------------------
> We concur with Ismaeel's thoughts here, 'get' is a
> point-in-time operation and will transparently reflect the value that is
> bound with a particular key at a given point of time in the memcache
> keystore. This is similar to reading a FileSystem or querying a database
> etc at a specific time and returning the contents / resultset.
>
> B. write - takes in a PCollection<key,value> and writes it to the memcache
>
> ----------------------------------------------------------------------------------------------------
>
> C. Other operations / mutations :
> -------------------------------------------
> Other operations that can be supported in subsequent iteration - add, cas,
> delete, replace, gets( get with CAS )
> There are a few more operations such as incr, decr, append, prepend etc
> which needs a broader discussion on whether to implememnt them in the
> transform.
>
> A few points on other proposals -
>
> Full read Vs Key based read -
> --------------------------------------
> We think that a key based read makes more sense here as it seems to be the
> primary usecase for memcache. Most of the applications using memcache use
> it as a key-value lookup store and hence makes sense to build on the same
> principles while developing a connector in Apache Beam. Also please note
> that key-value set/lookup is what all memcache implementations do best,
> though there are other operations which are supported. Hence we feel that
> key-based read would be the primary use case for a memcache IO in Apache
> Beam.
>
> Consistency / Snapshot -
> ---------------------------------
> This makes perfect sense for a full read / unbounded source, not sure if
> it suits a key-based read. Key-based read will simply return the value from
> memcache store transparently at a given point of time.
>
> Please let me know your comments, I plan to start developing this once we
> have a consensus.
>
> Regards,
> Seshadri
>
> -----Original Message-----
> From: Ismaël Mejía [mailto:ieme...@gmail.com]
> Sent: Tuesday, July 11, 2017 9:54 AM
> To: dev@beam.apache.org
> Subject: Re: [PROPOSAL] Connectors for memcache and Couchbase
>
> Hello again,
>
> Thanks Lukasz for the details. We will take a look and discuss with the
> others on how to achieve this. We hadn’t considered the case of a full scan
> Read (as Eugene mentions) so now your comments about the snapshot make more
> sense, however I am still wondering if the snapshot is worth the effort,
> and I say this because we are not doing something like this for other data
> stores, but it is a really interesting idea.
> I also didn’t know Memcached had watches, this reminds me of Redis’
> pubsub mechanism and could make sense for a possible unbounded source as
> you mention. Another idea to explore, and I think JB is doing something
> like this for RedisIO.
>
> @Eugene, thanks also for your comments, you are right this is more of a
> lookup but I am not sure that renaming it lookup will make things easier
> for the end users considering that other IOs use the read() convention and
> they indeed can do lookups as well as full scans of the data. I partially
> agree with you in the usefulness of lookup, but a simple example that comes
> to my mind is doing a lookup in Memcached to use it as a Side Input of a
> Pipeline. Finally I agree that supporting other commands is something
> important we just have to be sure to get the correct abstraction for this,
> I suppose we should restrict it to idempotent operations (so not
> incr/decr), and eventually make users pass the expiry time in date format
> so it does not get ‘overwritten’
> if a worker fails and the operation is re-executed. And about this point
> it is probably a good idea that we have some common semantics of the API
> for the different in memory stores (Redis, Memcached, JCache, etc),
>
> Any other ideas/comments?
>
> I think it is important now that we get a first working version now and
> then we can refine it incrementally with the different ideas.
>
> Ismaël
>
> On Tue, Jul 11, 2017 at 3:20 AM, Eugene Kirpichov
> <kirpic...@google.com.invalid> wrote:
> > I think Madhusudan's proposal does not involve reading the whole
> > contents of the memcached cluster - it's applied to a
> PCollection<byte[]> of keys.
> > So I'd suggest to call it MemcachedIO.lookup() rather than
> > MemcachedIO.read(). And it will not involve the questions of splitting
> > - however, it *will* involve snapshot consistency (looking up the same
> > key at different times may yield different results, including a null
> result).
> >
> > Concur with others - please take a look at
> > https://beam.apache.org/documentation/io/authoring-overview/ and
> > https://beam.apache.org/contribute/ptransform-style-guide/ , as well
> > as at the code of other IO transforms. The proposed API contradicts
> > several best practices described in these documents, but is easily
> fixable.
> >
> > I recommend to also consider how you plan to extend this to support
> > other commands - and which commands do you expect to ever support.
> > Also, I'm unsure about the usefulness of MemcachedIO.lookup(). What's
> > an example real-world use case for such a bulk lookup operation, where
> > you transform a PCollection of keys into a PCollection of key/value
> > pairs? I suppose such a use case exists, but I'd like to know more
> > about it, to see whether this is the best API for it.
> >
> > On Mon, Jul 10, 2017 at 9:18 AM Lukasz Cwik <lc...@google.com.invalid>
> > wrote:
> >
> >> Splitting on slabs should allow you to split more finely grained then
> >> per server since each server itself maintains this information. If
> >> you take a look at the memcached protocol, you can see that
> >> lru_crawler supports a metadump command which will enumerate all the
> >> key for a set of given slabs or for all the slabs.
> >>
> >> For the consistency part, you can get a snapshot like effect
> >> (snapshot like since its per server and not across the server farm)
> >> by combining the "watch mutations evictions" command on one
> >> connection with the "lru_crawler metadump all" on another connection
> >> to the same memcached server. By first connecting using a watcher and
> >> then performing a dump you can create two logical streams of data
> >> that can be joined to get a snapshot per server. If the amount of
> >> data/mutations/evications is small, you can perform all of this
> >> within a DoFn otherwise you can just treat each as two different
> >> outputs which you join and perform the same logical operation to
> rebuild the snapshot on a per key basis.
> >>
> >> Interestingly, the "watch mutations" command would allow one to build
> >> a streaming memcache IO which shows all changes occurring underneath.
> >>
> >> memcached protocol:
> >> https://github.com/memcached/memcached/blob/master/doc/protocol.txt
> >>
> >> On Mon, Jul 10, 2017 at 2:41 AM, Ismaël Mejía <ieme...@gmail.com>
> wrote:
> >>
> >> > Hello,
> >> >
> >> > Thanks Lukasz for bring some of this subjects. I have briefly
> >> > discussed with the guys working on this they are the same team who
> >> > did HCatalogIO (Hive).
> >> >
> >> > We just analyzed the different libraries that allowed to develop
> >> > this integration from Java and decided that the most complete
> >> > implementation was spymemcached. One thing I really didn’t like of
> >> > their API is that there is not an abstraction for Mutation (like in
> >> > Bigtable/Hbase) but a corresponding method for each operation so to
> >> > make things easier we discussed to focus first on read/write.
> >> >
> >> > @Lukasz for the enumeration part, I am not sure I follow, we had
> >> > just discussed a naive approach for splitting by server given that
> >> > Memcached is not a cluster but a server farm ‘which means every
> >> > server is its own’ we thought this will be the easiest way to
> >> > partition, is there any technical issue that impeaches this
> >> > (creating a BoundedSource and just read per each server)? Or
> >> > partitioning by slabs will bring us a better optimization? (Notice
> >> > I am far from an expert on Memcached).
> >> >
> >> > For the consistency part I assumed it will be inconsistent when
> >> > reading, because I didn’t know how to do the snapshot but if you
> >> > can give us more details on how to do this, and why it is worth the
> >> > effort (vs the cost of the snapshot), this will be something
> >> > interesting to integrate.
> >> >
> >> > Thanks,
> >> > Ismaël
> >> >
> >> >
> >> > On Sun, Jul 9, 2017 at 7:39 PM, Lukasz Cwik
> >> > <lc...@google.com.invalid>
> >> > wrote:
> >> > > For the source:
> >> > > Do you plan to support enumerating all the keys via cachedump /
> >> > lru_crawler
> >> > > metadump / ...?
> >> > > If there is an option which doesn't require enumerating the keys,
> >> > > how
> >> > will
> >> > > splitting be done (no splitting / splitting on slab ids / ...)?
> >> > > Can the cache be read while its still being modified (will
> >> > > effectively
> >> a
> >> > > snapshot be made using a watcher or is it expected that the cache
> >> > > will
> >> be
> >> > > read only or inconsistent when reading)?
> >> > >
> >> > > Also, as a usability point, all PTransforms are meant to be
> >> > > applied to PCollections and not vice versa.
> >> > > e.g.
> >> > > PCollection<byte[]> keys = ...;
> >> > > keys.apply(MemCacheIO.withConfig());
> >> > >
> >> > > This makes it so that people can write:
> >> > > PCollection<...> output =
> >> > > input.apply(ptransform1).apply(ptransform2).apply(...);
> >> > > It also makes it so that a PTransform can be applied to multiple
> >> > > PCollections.
> >> > >
> >> > > If you haven't already, I would also suggest that you take a look
> >> > > at
> >> the
> >> > > Pipeline I/O guide:
> >> > > https://beam.apache.org/documentation/io/io-toc/
> >> > > Talks about various usability points and how to write a good I/O
> >> > connector.
> >> > >
> >> > >
> >> > > On Sat, Jul 8, 2017 at 9:31 PM, Jean-Baptiste Onofré
> >> > > <j...@nanthrax.net>
> >> > > wrote:
> >> > >
> >> > >> Hi,
> >> > >>
> >> > >> Great job !
> >> > >>
> >> > >> I'm looking forward for the PRs review.
> >> > >>
> >> > >> Regards
> >> > >> JB
> >> > >>
> >> > >>
> >> > >> On 07/08/2017 09:50 AM, Madhusudan Borkar wrote:
> >> > >>
> >> > >>> Hi,
> >> > >>> We are proposing to build connectors for memcache first and
> >> > >>> then use
> >> it
> >> > >>> for
> >> > >>> Couchbase. The connector for memcache will be build as a
> >> > >>> IOTransform
> >> > and
> >> > >>> then it can be used for other memcache implementations
> >> > >>> including Couchbase.
> >> > >>>
> >> > >>> 1. As Source
> >> > >>>
> >> > >>> input will be a key(String / byte[]), output will be a
> >> > >>> KV<key,
> >> > value>
> >> > >>>
> >> > >>> where key - String / byte[]
> >> > >>>
> >> > >>> value - String / byte[]
> >> > >>>
> >> > >>> Spymemcached supports a multi-get operation where it takes
> >> > >>> a
> >> bunch
> >> > of
> >> > >>> keys and retrieves the associated values, the input
> >> > >>> PCollection<key>
> >> > can
> >> > >>> be
> >> > >>> bundled into multiple batches and each batch can be submitted
> >> > >>> via the multi-get operation.
> >> > >>>
> >> > >>> PCollection<KV<byte[], byte[]>> values =
> >> > >>>
> >> > >>> MemCacheIO
> >> > >>>
> >> > >>> .withConfig()
> >> > >>>
> >> > >>> .read()
> >> > >>>
> >> > >>> .withKey(PCollection<byte[]>);
> >> > >>>
> >> > >>>
> >> > >>> 2. As Sink
> >> > >>>
> >> > >>> input will be a KV<key, value>, output will be none or
> >> > >>> probably a boolean indicating the outcome of the operation
> >> > >>>
> >> > >>>
> >> > >>>
> >> > >>>
> >> > >>>
> >> > >>> //write
> >> > >>>
> >> > >>> MemCacheIO
> >> > >>>
> >> > >>> .withConfig()
> >> > >>>
> >> > >>> .write()
> >> > >>>
> >> > >>> .withEntries(PCollection<KV<byte[],byte[]>>);
> >> > >>>
> >> > >>>
> >> > >>> Implementation plan
> >> > >>>
> >> > >>> 1. Develop Memcache connector with 'set' and 'add' operation
> >> > >>>
> >> > >>> 2. Then develop other operations
> >> > >>>
> >> > >>> 3. Use Memcache connector for Couchbase
> >> > >>>
> >> > >>>
> >> > >>> Thanks @Ismael for help
> >> > >>>
> >> > >>> Please, let us know your views.
> >> > >>>
> >> > >>> Madhu Borkar
> >> > >>>
> >> > >>>
> >> > >> --
> >> > >> Jean-Baptiste Onofré
> >> > >> jbono...@apache.org
> >> > >> http://blog.nanthrax.net
> >> > >> Talend - http://www.talend.com
> >> > >>
> >> >
> >>
>
>
