I think we've discussed this before... It is true that all of our second-order APIs can be re-expressed as first-order APIs, but that would come at a very serious performance cost - e.g. significant increase in amount of data shuffled / materialized. The second-order APIs (most importantly, Dynamic Destinations in BigQueryIO and FileIO.write) were deliberately designed this way to minimize amount of data shuffled (grouped by destination) and postpone the redundant representation (record, destination) as late as possible in the graph.
Experience from https://github.com/apache/beam/pull/3894 shows that getting rid of TableRow in the representation alone gives 2x-3x end-to-end performance gains. Gains from avoiding redundancy are likely still higher. Some back-of-the-envelope calculations: - Imagine we're writing purchases to BigQuery tables (store_id, customer_id, product_id, quantity), one table per store, e.g. "stores_dataset:purchases_$storeid". - With the second-order representation, we group records (store_id, customer_id, product_id, quantity) by store_id - both key and value are a handful of bytes. - With the first-order representation, we group records of the form {table_name:"stores_dataset:purchases_$storeid", row: {"store_id":"...", "customer_id":"...", "product_id":"...", "quantity":"..."}} which is many times more in encoded form, both key and value. Schemas, perhaps, might help with encoding the TableRow less redundantly, but a) I don't know if that's actually the case b) the schemas work will also take more time to land in non-Java SDKs c) this doesn't get rid of the redundancy in keys (at a minimum, in a cross-language world, we'll definitely need to *encode* all such keys redundantly) I.e., yes, first-order APIs are possible, and likely even desirable for the first version of cross-language IO, but I do not believe that staying first-order-only is an acceptable long-term state. Note also that second-order doesn't necessarily require ability to invoke cross-language lambdas - there are other approaches, e.g. rewriting in the SDK language only the sub-parts of the transforms that actually use the lambdas. On Mon, Apr 30, 2018 at 10:34 AM Lukasz Cwik <lc...@google.com> wrote: > I believe that most (all?) of these cases of executing a lambda could be > avoided if we passed along structured records like: > { > table_name: > row: { ... } > } > > > On Mon, Apr 30, 2018 at 10:24 AM Chamikara Jayalath <chamik...@google.com> > wrote: > >> >> >> On Mon, Apr 30, 2018 at 9:54 AM Kenneth Knowles <k...@google.com> wrote: >> >>> I agree with Cham's motivations as far as "we need it now" and getting >>> Python SDF up and running and exercised on a real connector. >>> >>> But I do find the current API of BigQueryIO to be a poor example. That >>> particular functionality on BigQueryIO seems extraneous and goes against >>> our own style guide [1]. The recommended way to write it would be for >>> BigQueryIO to output a natural concrete type (like TableRow) and allow the >>> following step to do conversions. This is a broader programming best >>> practice - unless there is a compelling reason, you should just return the >>> value rather than accept a higher-order function to operate on the value. >>> Is there a compelling reason in this case? I just dug through the code and >>> just see that it bottoms out in AvroSource where it does not seem to add >>> functionality. >>> >> >> I think what Reuven was referring to was a functionality of the BQ sink >> not the source. Even if we set the destination in an additional step, this >> will still have to cross SDK harness boundary just to set the property >> (invoking BQ destination lambda in this case), won't it? Java will be able >> to fuse steps in this case but Python/Go will involve "some work in Java >> SDK harness" + "invoking Python/Go harness to invoke the BQ destination >> lambda" + "invoke Java harness to continue processing", including >> serialization/deserialization in between. This could still result in a >> significant amount of overhead for SDKs other than the one where the IO is >> implemented. >> >> >>> >>> Considering cross-language pipelines as a primary use case for all >>> connectors, perhaps we should audit them and bring them into alignment now, >>> deprecating paths using higher-order functions. We can still consider >>> host-language convenience composites. >>> >>> For an unbounded source like KafkaIO the compelling reason is the >>> timestamp extracting function to be able to maintain a watermark. Notably, >>> PubsubIO does not accept such a function, but requires the timestamp to be >>> in a metadata field that any language can describe (versus having to parse >>> the message to pull out the timestamp). >>> >>> Kenn >>> >>> [1] >>> https://beam.apache.org/contribute/ptransform-style-guide/#choosing-types-of-input-and-output-pcollections >>> >>> On Mon, Apr 30, 2018 at 9:27 AM Reuven Lax <re...@google.com> wrote: >>> >>>> Another point: cross-language IOs might add a performance penalty in >>>> many cases. For an example of this look at BigQueryIO. The user can >>>> register a SerializableFunction that is evaluated on every record, and >>>> determines which destination to write the record to. Now a Python user >>>> would want to register a Python function for this of course. this means >>>> that the Java IO would have to invoke Python code for each record it sees, >>>> which will likely be a big performance hit. >>>> >>>> Of course the downside of duplicating IOs is exactly as you say - >>>> multiple versions to maintain, and potentially duplicate bugs. I think the >>>> right answer will need to be on a case-by-case basis. >>>> >>>> Reuven >>>> >>>> On Mon, Apr 30, 2018 at 8:05 AM Chamikara Jayalath < >>>> chamik...@google.com> wrote: >>>> >>>>> Hi Aljoscha, >>>>> >>>>> I tried to cover this in the doc. Once we have full support for >>>>> cross-language IO, we can decide this on a case-by-case basis. But I don't >>>>> think we should cease defining new sources/sinks for Beam Python SDK till >>>>> we get to that point. I think there are good reasons for adding Kafka >>>>> support for Python today and many Beam users have request this. Also, note >>>>> that proposed Python Kafka source will be based on the Splittable DoFn >>>>> framework while the current Java version is based on the UnboundedSource >>>>> framework. Here are the reasons that are currently listed in the doc. >>>>> >>>>> >>>>> - >>>>> >>>>> Users might find it useful to have at least one unbounded source >>>>> and sink combination implemented in Python SDK and Kafka is the >>>>> streaming >>>>> system that makes most sense to support if we just want to add support >>>>> for >>>>> only one such system in Python SDK. >>>>> - >>>>> >>>>> Not all runners might support cross-language IO. Also some >>>>> user/runner/deployment combinations might require an unbounded >>>>> source/sink >>>>> implemented in Python SDK. >>>>> - >>>>> >>>>> We recently added Splittable DoFn support to Python SDK. It will >>>>> be good to have at least one production quality Splittable DoFn >>>>> that will server as a good example for any users who wish to implement >>>>> new >>>>> Splittable DoFn implementations on top of Beam Python SDK. >>>>> - >>>>> >>>>> Cross-language transform feature is currently is in the initial >>>>> discussion phase and it could be some time before we can offer existing >>>>> Java implementation of Kafka for Python SDK users. >>>>> - >>>>> >>>>> Cross-language IO might take even longer to reach the point where >>>>> it's fully equivalent in expressive power to a transform written in the >>>>> host language - e.g. supporting host-language lambdas as part of the >>>>> transform configuration is likely to take a lot longer than >>>>> "first-order" >>>>> cross-language IO. KafkaIO in Java uses lambdas as part of transform >>>>> configuration, e.g. timestamp functions. >>>>> >>>>> >>>>> Thanks, >>>>> Cham >>>>> >>>>> On Mon, Apr 30, 2018 at 2:14 AM Aljoscha Krettek <aljos...@apache.org> >>>>> wrote: >>>>> >>>>>> Is this what we want to do in the long run, i.e. implement copies of >>>>>> connectors for different SDKs? I thought the plan was to enable using >>>>>> connectors written in different languages, i.e. use the Java Kafka I/O >>>>>> from >>>>>> python. This way we wouldn't duplicate bugs for three different language >>>>>> (Java, Python, and Go for now). >>>>>> >>>>>> Best, >>>>>> Aljoscha >>>>>> >>>>>> >>>>>> On 29. Apr 2018, at 20:46, Eugene Kirpichov <kirpic...@google.com> >>>>>> wrote: >>>>>> >>>>>> Thanks Cham, this is great! I left just a couple of comments on the >>>>>> doc. >>>>>> >>>>>> On Fri, Apr 27, 2018 at 10:06 PM Chamikara Jayalath < >>>>>> chamik...@google.com> wrote: >>>>>> >>>>>>> Hi All, >>>>>>> >>>>>>> I'm looking into adding a Kafka connector to Beam Python SDK. I >>>>>>> think this will benefits many Python SDK users and will serve as a good >>>>>>> example for recently added Splittable DoFn API (Fn API support which >>>>>>> will >>>>>>> allow all runners to use Python Splittable DoFn is in active >>>>>>> development). >>>>>>> I created a document [1] that makes the case for adding this connector >>>>>>> and >>>>>>> compares the performance of available Python Kafka client libraries. >>>>>>> Also I >>>>>>> created a POC [2] that illustrates the API and how Python SDF API can be >>>>>>> used to implement a Kafka source. I extremely appreciate any feedback >>>>>>> related to this. >>>>>>> >>>>>>> [1] >>>>>>> https://docs.google.com/document/d/1ogRS-e-HYYTHsXi_l2zDUUOnvfzEbub3BFkPrYIOawU/edit?usp=sharing >>>>>>> [2] >>>>>>> https://github.com/chamikaramj/beam/commit/982767b69198579b22522de6794242142d12c5f9 >>>>>>> >>>>>>> Thanks, >>>>>>> Cham >>>>>>> >>>>>> >>>>>>