Awesome!! Thanks for the heads up, very exciting, this is going to make a lot of people happy :)
On Tue, Jul 3, 2018, 3:40 AM Carlos Alonso <car...@mrcalonso.com> wrote: > + dev@beam.apache.org > > Just a quick email to let you know that I'm starting developing this. > > On Fri, Apr 20, 2018 at 10:30 PM Eugene Kirpichov <kirpic...@google.com> > wrote: > >> Hi Carlos, >> >> Thank you for expressing interest in taking this on! Let me give you a >> few pointers to start, and I'll be happy to help everywhere along the way. >> >> Basically we want BigQueryIO.write() to return something (e.g. a >> PCollection) that can be used as input to Wait.on(). >> Currently it returns a WriteResult, which only contains a >> PCollection<TableRow> of failed inserts - that one can not be used >> directly, instead we should add another component to WriteResult that >> represents the result of successfully writing some data. >> >> Given that BQIO supports dynamic destination writes, I think it makes >> sense for that to be a PCollection<KV<DestinationT, ???>> so that in theory >> we could sequence different destinations independently (currently Wait.on() >> does not provide such a feature, but it could); and it will require >> changing WriteResult to be WriteResult<DestinationT>. As for what the "???" >> might be - it is something that represents the result of successfully >> writing a window of data. I think it can even be Void, or "?" (wildcard >> type) for now, until we figure out something better. >> >> Implementing this would require roughly the following work: >> - Add this PCollection<KV<DestinationT, ?>> to WriteResult >> - Modify the BatchLoads transform to provide it on both codepaths: >> expandTriggered() and expandUntriggered() >> ...- expandTriggered() itself writes via 2 codepaths: single-partition >> and multi-partition. Both need to be handled - we need to get a >> PCollection<KV<DestinationT, ?>> from each of them, and Flatten these two >> PCollections together to get the final result. The single-partition >> codepath (writeSinglePartition) under the hood already uses WriteTables >> that returns a KV<DestinationT, ...> so it's directly usable. The >> multi-partition codepath ends in WriteRenameTriggered - unfortunately, this >> codepath drops DestinationT along the way and will need to be refactored a >> bit to keep it until the end. >> ...- expandUntriggered() should be treated the same way. >> - Modify the StreamingWriteTables transform to provide it >> ...- Here also, the challenge is to propagate the DestinationT type all >> the way until the end of StreamingWriteTables - it will need to be >> refactored. After such a refactoring, returning a KV<DestinationT, ...> >> should be easy. >> >> Another challenge with all of this is backwards compatibility in terms of >> API and pipeline update. >> Pipeline update is much less of a concern for the BatchLoads codepath, >> because it's typically used in batch-mode pipelines that don't get updated. >> I would recommend to start with this, perhaps even with only the >> untriggered codepath (it is much more commonly used) - that will pave the >> way for future work. >> >> Hope this helps, please ask more if something is unclear! >> >> On Fri, Apr 20, 2018 at 12:48 AM Carlos Alonso <car...@mrcalonso.com> >> wrote: >> >>> Hey Eugene!! >>> >>> I’d gladly take a stab on it although I’m not sure how much available >>> time I might have to put into but... yeah, let’s try it. >>> >>> Where should I begin? Is there a Jira issue or shall I file one? >>> >>> Thanks! >>> On Thu, 12 Apr 2018 at 00:41, Eugene Kirpichov <kirpic...@google.com> >>> wrote: >>> >>>> Hi, >>>> >>>> Yes, you're both right - BigQueryIO.write() is currently not >>>> implemented in a way that it can be used with Wait.on(). It would certainly >>>> be a welcome contribution to change this - many people expressed interest >>>> in specifically waiting for BigQuery writes. Is any of you interested in >>>> helping out? >>>> >>>> Thanks. >>>> >>>> On Fri, Apr 6, 2018 at 12:36 AM Carlos Alonso <car...@mrcalonso.com> >>>> wrote: >>>> >>>>> Hi Simon, I think your explanation was very accurate, at least to my >>>>> understanding. I'd also be interested in getting batch load result's >>>>> feedback on the pipeline... hopefully someone may suggest something, >>>>> otherwise we could propose submitting a Jira, or even better, a PR!! :) >>>>> >>>>> Thanks! >>>>> >>>>> On Thu, Apr 5, 2018 at 2:01 PM Simon Kitching < >>>>> simon.kitch...@unbelievable-machine.com> wrote: >>>>> >>>>>> Hi All, >>>>>> >>>>>> I need to write some data to BigQuery (batch-mode) and then send a >>>>>> Pubsub message to trigger further processing. >>>>>> >>>>>> I found this thread titled "Callbacks/other functions run after a >>>>>> PDone/output transform" on the user-list which was very relevant: >>>>>> >>>>>> https://lists.apache.org/thread.html/ddcdf93604396b1cbcacdff49aba60817dc90ee7c8434725ea0d26c0@%3Cuser.beam.apache.org%3E >>>>>> >>>>>> Thanks to the author of the Wait transform (Beam 2.4.0)! >>>>>> >>>>>> Unfortunately, it appears that the Wait.on transform does not work >>>>>> with BiqQueryIO in FILE_LOADS mode - or at least I cannot get it to work. >>>>>> Advice appreciated. >>>>>> >>>>>> Here's (most of) the relevant test code: >>>>>> Pipeline p = Pipeline.create(options); >>>>>> PCollection<String> lines = p.apply("Read Input", >>>>>> Create.of("line1", "line2", "line3", "line4")); >>>>>> >>>>>> TableFieldSchema f1 = new >>>>>> TableFieldSchema().setName("value").setType("string"); >>>>>> TableSchema s2 = new >>>>>> TableSchema().setFields(Collections.singletonList(f1)); >>>>>> >>>>>> WriteResult writeResult = lines.apply("Write and load data", >>>>>> BigQueryIO.<String>write() // >>>>>> .to(options.getTableSpec()) // >>>>>> .withFormatFunction(new SlowFormatter()) // >>>>>> .withMethod(BigQueryIO.Write.Method.FILE_LOADS) // >>>>>> // >>>>>> .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS) // >>>>>> .withSchema(s2) >>>>>> >>>>>> .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED) >>>>>> // >>>>>> >>>>>> .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)); >>>>>> >>>>>> >>>>>> lines.apply(Wait.on(writeResult.getFailedInserts())).apply(ParDo.of(new >>>>>> OnCompletion())); >>>>>> >>>>>> where >>>>>> + format-function "SlowFormatter" prints out each line and has a >>>>>> small sleep for testing purposes, and >>>>>> + DoFn OnCompletion just prints out the contents of each line >>>>>> >>>>>> In production code, OnCompletion would be fed some collection derived >>>>>> from lines, eg min/max record id, and the operation would be "send pubsub >>>>>> message" rather than print.. >>>>>> >>>>>> My expectation is that the "SlowFormatter" would run for each line, >>>>>> then the data would be uploaded, then OnCompletion would print each line. >>>>>> And indeed that happens when STREAMING_INSERTS is used. However for >>>>>> FILE_LOADS, LinePrinter runs before the upload takes place. >>>>>> >>>>>> I use WriteResult.getFailedInserts as that is the only "output" that >>>>>> BiqQueryIO.write() generates AFAICT. I don't expect any failed records, >>>>>> but >>>>>> believe that it can be used as a "signal" for the Wait.on - ie the output >>>>>> is "complete for window" only after all data has been uploaded, which is >>>>>> what I need. And that does seem to work for STREAMING_LOADS. >>>>>> >>>>>> I suspect the reason that this does not work for FILE_LOADS is that >>>>>> method BatchLoads.writeResult returns a WriteResult that wraps an "empty" >>>>>> failedInserts collection, ie data which is not connected to the >>>>>> batch-load-job that is triggered: >>>>>> private WriteResult writeResult(Pipeline p) { >>>>>> PCollection<TableRow> empty = >>>>>> p.apply("CreateEmptyFailedInserts", >>>>>> Create.empty(TypeDescriptor.of(TableRow.class))); >>>>>> return WriteResult.in(p, new TupleTag<>("failedInserts"), empty); >>>>>> } >>>>>> >>>>>> Note that BatchLoads does "synchronously" invoke BigQuery load jobs; >>>>>> once a job is submitted the code repeatedly polls the job status until it >>>>>> reaches DONE or FAILED. However that information does not appear to be >>>>>> exposed anywhere (unlike streaming which effectively exposes >>>>>> completion-state via the failedInserts stream). >>>>>> >>>>>> If I have misunderstood something, corrections welcome! If not, >>>>>> suggestions for workarounds or alternate solutions are also welcome :-) >>>>>> >>>>>> Thanks, >>>>>> Simon >>>>>> >>>>>>
