Custom Dataflow Template - BigQuery to CloudStorage - documentation? general solution advice?

0
votes

I am consuming a BigQuery table datasource. It is 'unbounded' as it is updated via a batch process. It contains session keyed reporting data from server logs where each row captures a request. I do not have access to the original log data and must consume the BigQuery table.

I would like to develop a custom Java based google Dataflow template using beam api with the goals of :

  1. collating keyed session objects

  2. deriving session level metrics

  3. deriving filterable window level metrics based on session metrics, e.g., percentage of sessions with errors during previous window and percentage of errors per filtered property, e.g., error percentage per device type

  4. writing the result as a formatted/compressed report to cloud storage.

This seems like a fairly standard use case? In my research thus far, I have not yet found a perfect example and still have not been able to determine the best practice approach for certain basic requirements. I would very much appreciate any pointers. Keywords to research? Documentation, tutorials. Is my current thinking right or do I need to consider other approaches?

Questions :

  1. beam windowing and BigQuery I/O Connector - I see that I can specify a window type and size via beam api. My BQ table has a timestamp field per row. Am I supposed to somehow pass this via configuration or is it supposed to be automagic? Do I need to do this manually via a SQL query somehow? This is not clear to me.

  2. fixed time windowing vs. session windowing functions - examples are basic and do not address any edge cases. My sessions can last hours. There are potentially 100ks plus session keys per window. Would session windowing support this?

  3. BigQuery vs. BigQueryClientStorage - The difference is not clear to me. I understand that BQCS provides a performance benefit, but do I have to store BQ data in a preliminary step to use this? Or can I simply query my table directly via BQCS and it takes care of that for me?

1
google-bigquerygoogle-cloud-dataflowapache-beam
Take into account that you want to use BigQuery as a streaming source (unbounded). Regarding questions #1 and #3, BigQueryIO Reads and Storage API will read once a table snapshot and not consider newly arrived data. Consuming streaming data is usually a Pub/Sub use case but I understand that you only have access to the BigQuery table. You can use the BigQuery Client Library in a side input that it's refreshed every X time (beam.apache.org/documentation/patterns/side-inputs) but that would entail other challenges (query to fetch incremental data, late data, deduplication, etc) - Guillem Xercavins
For Question #2, you would need to window into sessions and then re-window the results into fixed-time windows. Of course, having a lot of keys and events in a single window could cause OOM issues so I would recommend using a combiner as soon as possible (i.e. when calculating the sessions) Of course if you can do the analysis a posteriori you can simply run a daily batch job with data from the day before and simplify the whole design. - Guillem Xercavins
Thank you Guillem. I think I understand. The BQ table source is time partitioned into days. So I am thinking I should take this into account somehow. Also, once I generate a report for a given day and persist it to Cloud Storage - I shouldn't need to query that BQ data again. For the current day I was thinking I might simply overwrite the report at an interval - say one hour - since the report doesn't need to be realtime. If I overwrite the entire report I wouldn't need to fiddle with side inputs, late data, deduplication, etc. (discounting any sessions that overlap across days) - user3205931
- this might cost a little more I guess, but I think this might be worth it in order to have hourly data. So yes the daily batch job seems like a good way to simplify. OOM was a real issue when I was using BQ client library. In that case the key was to deallocate memory as soon as it wasn't needed by clearing out event data and just keeping session summary data. I'm still new to beam but I have the sense that combiner has a similar effect. - user3205931

1 Answers

1
votes

For number 1 you can simply use a withTimestamps function before applying windowing, this assigns the timestamp to your items. Here are some python examples.

For number 2 the documentation states:

Session windowing applies on a per-key basis and is useful for data that is irregularly distributed with respect to time. [...] If data arrives after the minimum specified gap duration time, this initiates the start of a new window.

Also in the java documentation, you can only specify a minimum gap duration, but not a maximum. This means that session windowing can easily support hour-lasting sessions. After all, the only thing it does is putting a watermark on your data and keeping it alive.

For number 3, the differences between the BigQuery IO Connector and the BigQuery storage APIs is that the latter (an experimental feature as of 01/2020) access directly data stored, without the logical passage through BigQuery (BigQuery data isn't stored in BigQuery). This means that with storage APIs, the documentation states:

you can't use it to read data sources such as federated tables and logical views

Also, there are different limits and quotas between the two methods, that you can find in the documentation link above.