beam-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Anton Kedin <>
Subject Re: Pubsub to Beam SQL
Date Fri, 04 May 2018 19:10:50 GMT
There are few aspects of the event timestamp definition in SQL, which we
are talking about here:

   - configuring the source. E.g. for PubsubIO you can choose whether to
   extract event timestamp from the message attributes or the message publish
   - this is source-specific and cannot be part of the common DDL;
      - TBLPROPERTIES, on the other hand, is an opaque json blob which
      exists specifically for source configuration;
      - as Kenn is saying, some sources might not even have such
      - at processing time, event timestamp is available in
      ProcessContext.timestamp() regardless of the specifics of the source
      configuration, so it can be extracted the same way for all sources, as
      Raghu said;
   - designating one of the table columns as an event timestamp:
      - query needs to be able to reference the event timestamp so we have
      to declare which column to populate with the event timestamp;
      - this is common for all sources and we can create a special syntax,
      e.g. "columnName EVENT_TIMESTAMP". It must not contain source-specific
      configuration at this point, in my opinion;
      - when SQL knows which column is supposed to be the timestamp, then
      it can get it from the ProcessContext.timestamp() and put it into the
      designated field the same way regardless of the source configuration;
      - pubsub-specific message formatting:
      - on top of the above we want to be able to expose pubsub message
      attributes, payload, and timestamp to the user queries, and do it without
      magic or user schema modifications. To do this we can enforce some
      pubsub-specific schema limitations, e.g. by exposing attributes and
      timestamp fields at a top-level schema, with payload going into
the second
      level in its own field;
      - this aspect is not fully implementable until we have support for
      complex types. Until then we cannot map full JSON to the payload field;

I will update the doc and the implementation to reflect these comments
where possible.

Thank you,

On Fri, May 4, 2018 at 9:48 AM Raghu Angadi <> wrote:

> On Thu, May 3, 2018 at 12:47 PM Anton Kedin <> wrote:
>> I think it makes sense for the case when timestamp is provided in the
>> payload (including pubsub message attributes).  We can mark the field as an
>> event timestamp. But if the timestamp is internally defined by the source
>> (pubsub message publish time) and not exposed in the event body, then we
>> need a source-specific mechanism to extract and map the event timestamp to
>> the schema. This is, of course, if we don't automatically add a magic
>> timestamp field which Beam SQL can populate behind the scenes and add to
>> the schema. I want to avoid this magic path for now.
> Commented on the PR. As Kenn mentioned, every element in Beam has an event
> timestamp, there is no requirement to extract the timestamp by the SQL
> transform. Using the element timestamp takes care of Pubsub publish
> timestamp as well (in fact, this is the default when timestamp attribute is
> not specified in PubsubIO).
> How timestamp are customized is specific to each source. That way custom
> timestamp option seem like they belong in TBLPROPERTIES. E.g. for KafkaIO,
> it could specify "logAppendTime", "createTime", or "processingTime" etc
> (though I am not sure how user can provide their own custom extractor in
> Beam SQL, may be it could support a timestamp field in json records).
> Raghu.
>> On Thu, May 3, 2018 at 11:10 AM Andrew Pilloud <>
>> wrote:
>>> This sounds awesome!
>>> Is event timestamp something that we need to specify for every source?
>>> If so, I would suggest we add this as a first class option on CREATE TABLE
>>> rather then something hidden in TBLPROPERTIES.
>>> Andrew
>>> On Wed, May 2, 2018 at 10:30 AM Anton Kedin <> wrote:
>>>> Hi
>>>> I am working on adding functionality to support querying Pubsub
>>>> messages directly from Beam SQL.
>>>> *Goal*
>>>>   Provide Beam users a pure  SQL solution to create the pipelines with
>>>> Pubsub as a data source, without the need to set up the pipelines in
>>>> Java before applying the query.
>>>> *High level approach*
>>>>    -
>>>>    - Build on top of PubsubIO;
>>>>    - Pubsub source will be declared using CREATE TABLE DDL statement:
>>>>       - Beam SQL already supports declaring sources like Kafka and
>>>>       Text using CREATE TABLE DDL;
>>>>       - it supports additional configuration using TBLPROPERTIES
>>>>       clause. Currently it takes a text blob, where we can put a JSON
>>>>       configuration;
>>>>       - wrapping PubsubIO into a similar source looks feasible;
>>>>    - The plan is to initially support messages only with JSON payload:
>>>>    -
>>>>       - more payload formats can be added later;
>>>>    - Messages will be fully described in the CREATE TABLE statements:
>>>>       - event timestamps. Source of the timestamp is configurable. It
>>>>       is required by Beam SQL to have an explicit timestamp column for windowing
>>>>       support;
>>>>       - messages attributes map;
>>>>       - JSON payload schema;
>>>>    - Event timestamps will be taken either from publish time or
>>>>    user-specified message attribute (configurable);
>>>> Thoughts, ideas, comments?
>>>> More details are in the doc here:
>>>> Thank you,
>>>> Anton

View raw message