You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Matthew Darwin <Ma...@carfinance247.co.uk> on 2019/07/12 08:35:59 UTC
[python SDK] Returning Pub/Sub message_id and timestamp
Good morning,
I'm very new to Beam, and pretty new to Python so please first accept my apologies for any obvious misconceptions/mistakes in the following.
I am currently trying to develop a sample pipeline in Python to pull messages from Pub/Sub and then write them to either files in cloud storage or to BigQuery. The ultimate goal will be to utilise the pipeline for real time streaming of event data to BigQuery (with various transformations) but also to store the raw messages long term in files in cloud storage.
At the moment, I'm simply trying to parse the message to get the PubSub messageId and publishTime in order to be able to write them into the output. The json of my PubSub message looks like this:-
[
{
"ackId": "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
"message": {
"attributes": {
"source": "python"
},
"data": "eyJyb3dudW1iZXIiOiAyfQ==",
"messageId": "619310330691403",
"publishTime": "2019-07-12T08:27:58.522Z"
}
}
]
According to the documentation<https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html> the PubSub message payload returns the data and attributes properties; is there simply no way of retrieving the messageId and publishTime, or are these exposed somewhere else? If not, will the inclusion of these be in the roadmap, and are they available if using Java (I have zero Java experience hence why reaching for Python first).
Kind regards,
Matthew
Disclaimer
The information contained in this communication from the sender is confidential. It is intended solely for use by the recipient and others authorized to receive it. If you are not the recipient, you are hereby notified that any disclosure, copying, distribution or taking action in relation of the contents of this information is strictly prohibited and may be unlawful.
This email has been scanned for viruses and malware, and may have been automatically archived by Mimecast Ltd, an innovator in Software as a Service (SaaS) for business. Providing a safer and more useful place for your human generated data. Specializing in; Security, archiving and compliance. To find out more visit the Mimecast website.
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Ahmet Altay <al...@google.com>.
Your analysis and suggestion looks correct to me. Thank you for reporting
this. I filed https://issues.apache.org/jira/browse/BEAM-7819 to track this.
On Thu, Jul 25, 2019 at 4:51 AM Matthew Darwin <
Matthew.Darwin@carfinance247.co.uk> wrote:
> Thanks Ahmet. Looking at the source code, with my untrained python eyes, I
> think if the intention is to include the message id and the publish time in
> the attributes attribute of the PubSubMessage type, then the protobuf
> mapping is missing something:-
>
>
> @staticmethod
> def _from_proto_str(proto_msg):
> """Construct from serialized form of ``PubsubMessage``.
>
> Args:
> proto_msg: String containing a serialized protobuf of type
>
> https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
>
> Returns:
> A new PubsubMessage object.
> """
> msg = pubsub.types.pubsub_pb2.PubsubMessage()
> msg.ParseFromString(proto_msg)
> # Convert ScalarMapContainer to dict.
> attributes = dict((key, msg.attributes[key]) for key in msg.attributes)
> return PubsubMessage(msg.data, attributes)
>
> The protobuf definition is here:-
>
>
> https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
>
> and so it looks as if the message_id and publish_time are not being parsed
> as they are seperate from the attributes. Perhaps the PubsubMessage class
> needs expanding to include these as attributes, or they would need adding
> to the dictionary for attributes. This would only need doing for the
> _from_proto_str as obviously they would not need to be populated when
> transmitting a message to PubSub.
>
> My python is not great, I'm assuming the latter option would need to look
> something like this?
>
> attributes = dict((key, msg.attributes[key]) for key in msg.attributes)
> attributes.update({'message_id': msg.message_id, 'publish_time':
> msg.publish_time})
> return PubsubMessage(msg.data, attributes)
>
> On Wed, 2019-07-24 at 10:01 -0700, Ahmet Altay wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> No, I do not see any issues with the code you posted on stackoverflow. It
> might very well be an issue with Beam.
>
> +Udi Meiri <eh...@google.com> is the author of this code, might have an
> idea.
>
> On Wed, Jul 24, 2019 at 9:47 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Thanks Ahmet, I'm aware of the documentation and what should be happening,
> however, as you can see in the example I posted on stackoverflow, this does
> not appear to be the case, hence my question here! If there are any glaring
> issues with my code, I'd be grateful if you could point them out.
>
> On Wed, 2019-07-24 at 09:16 -0700, Ahmet Altay wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> When with_attributes is set to True, the elements will be of type
> PubsubMessage [1]. I could not find a test/example for this, but
> documentation suggests [2], PubsubMessage will have an attributes map
> including the system provided values. One of those keys will be message_id
> [3].
>
> [1]
> https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L156
> [2]
> https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L94
> [3]
> https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
>
> On Wed, Jul 24, 2019 at 6:23 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Thank you both for you replies; apologies for the delay in responding, I
> was away from my machine over the weekend.
>
> I'm still a bit confused over how the message id from PubSub is exposed.
>
> From the documentation, I was expecting the PubSub message id to be in the
> attributes property of the message; but when running this into BigQuery as
> a string, the only properties are those that I pass in to my PubSub message
> as attributes. None of the examples you've linked use the with_attributes
> flag, unfortunately.
>
> I've posted the code and output onto stackoverflow:-
> https://stackoverflow.com/questions/57183876/how-do-you-access-the-message-id-from-google-pub-sub-using-apache-beam
>
> Secondly, is the 2.14 release available for preview?
>
>
> Release candidate is not out yet. This will happen soon (~this week) and
> you can watch dev@ for this. If it makes sense for your case, you can
> also build yourself from head and test it out.
>
>
>
> Kind regards,
>
> Matthew
>
>
> On Fri, 2019-07-19 at 13:22 -0700, Valentyn Tymofieiev wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> Also, see
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/complete/game/leader_board.py which
> involves both PubSub and Bigquery IOs.
>
> On Fri, Jul 19, 2019 at 12:31 PM Pablo Estrada <pa...@google.com> wrote:
>
> Beam 2.14.0 will include support for writing files in the fileio module
> (the support will include GCS, local files, HDFS). It will also support
> streaming. The transform is still marked as experimental, and is likely to
> receive improvements - but you can check it out for your pipelines, and see
> if it helps you : )
> Best
> -P.
>
> On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>
> wrote:
>
> As of today, Beam Python streaming does not support writing to GCS yet,
> which explains
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python .
>
>
> You are right - id_label and timestamp_attribute does not work on Direct
> runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I
> checked with a few folks and that seems to be the current status, but you
> can still give them a try on Dataflow runner.
>
> You may also find the following examples helpful:
>
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py
> (streaming pipeline).
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch
> pipeline)
>
> https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o
> .
>
> On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Hi Valentyn,
>
> Thank you for your reply. I'm already using the with_attributes=True
> option, however this returns the attributes property of the JSON, i.e :-
>
> {
> "attributes": {
> "source": "python"
> }
>
>
>
> My pipeline currently looks like this (id_label is commented out when
> running directly, as it causes a not implemented error):-
>
> messages = (p
> | 'Read From Pub Sub' >> ReadFromPubSub(subscription
> =known_args.input_subscription,with_attributes=True)
> #,id_label='message_id')
> | 'Parse JSON' >> beam.Map(format_message_element)
>
> My function to parse the message looks like this:-
>
> def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
> messagedict = json.loads(message.data)
> rownumber = messagedict['rownumber']
> fullmessage = {'data' : json.dumps(message.data),
> 'rownumber' : int(rownumber),
> 'attributes' : json.dumps(message.attributes),
> 'timestamp' : float(timestamp)}
>
> logging.info(message.attributes)
> logging.info(message)
>
> return (rownumber, fullmessage)
>
> I'm aware there are the id_label and with_timestamp parameters for the
> ReadFromPubSub method, however, these don't seem to work with the direct
> runner, as per
> https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22
> which makes testing somewhat difficult.
>
> My full code is attached, when running above 2.9.0 of the SDK I can't get
> passed the windowing function, due to an issue that appears related to this
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
> and this
> https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue
> as I was receiving the following error when running on 2.12.0:
>
> Cannot convert GlobalWindow to
> apache_beam.utils.windowed_value._IntervalWindowBase [while running
> 'generatedPtransform-150']
>
> On 2.9.0 when running on the local runner, I receive the following output
> from the logging.info calls in format_message_element:
>
> INFO:root:{u'source': u'python'}
> INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
>
> I was expecting the messageId and publishTime as part of the object
> returned; but as you can see there's nothing there for those attributes.
>
> (The code does not quite map correctly to the BigQuery table so it fails
> inserts at that point, which I'm currently trying to resolve!)
>
> Kind regards,
>
> Matthew
>
>
>
> On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> Hi Matthew,
>
> Welcome to Beam!
>
> Looking at Python PubSub IO API, you should be able to access id and
> timestamp by setting `with_attributes=True` when using `ReadFromPubSub`
> PTransform, see [1,2].
>
> [1]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
> [2]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
>
> On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Good morning,
>
> I'm very new to Beam, and pretty new to Python so please first accept my
> apologies for any obvious misconceptions/mistakes in the following.
>
> I am currently trying to develop a sample pipeline in Python to pull
> messages from Pub/Sub and then write them to either files in cloud storage
> or to BigQuery. The ultimate goal will be to utilise the pipeline for real
> time streaming of event data to BigQuery (with various transformations) but
> also to store the raw messages long term in files in cloud storage.
>
> At the moment, I'm simply trying to parse the message to get the PubSub
> messageId and publishTime in order to be able to write them into the
> output. The json of my PubSub message looks like this:-
>
> [
> {
> "ackId":
> "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
> "message": {
> "attributes": {
> "source": "python"
> },
> "data": "eyJyb3dudW1iZXIiOiAyfQ==",
> "messageId": "619310330691403",
> "publishTime": "2019-07-12T08:27:58.522Z"
> }
> }
> ]
> According to the documentation
> <https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html>
> the PubSub message payload returns the *data* and *attributes*
> properties; is there simply no way of retrieving the messageId and
> publishTime, or are these exposed somewhere else? If not, will the
> inclusion of these be in the roadmap, and are they available if using Java
> (I have zero Java experience hence why reaching for Python first).
>
> Kind regards,
>
> Matthew
>
>
>
>
> ---------- Forwarded message ----------
> From: Domain postMaster address <po...@carfinance247.co.uk>
> To: "user@beam.apache.org" <us...@beam.apache.org>
> Cc:
> Bcc:
> Date: Mon, 15 Jul 2019 09:12:15 +0000
> Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
>
> [image: Logo]
> We removed a file from this message
>
> Your organization's email policy doesn't permit this type of file. If you
> need it, please contact your administrator.
>
>
> File Details
>
> *ApacheBeamDataflow-To-BigQuery2.py* (6559 bytes)
>
>
> © 2003 - 2019 Mimecast Services Limited.
>
>
>
>
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Matthew Darwin <Ma...@carfinance247.co.uk>.
Thanks Ahmet. Looking at the source code, with my untrained python eyes, I think if the intention is to include the message id and the publish time in the attributes attribute of the PubSubMessage type, then the protobuf mapping is missing something:-
@staticmethod
def _from_proto_str(proto_msg):
"""Construct from serialized form of ``PubsubMessage``.
Args:
proto_msg: String containing a serialized protobuf of type
https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
Returns:
A new PubsubMessage object.
"""
msg = pubsub.types.pubsub_pb2.PubsubMessage()
msg.ParseFromString(proto_msg)
# Convert ScalarMapContainer to dict.
attributes = dict((key, msg.attributes[key]) for key in msg.attributes)
return PubsubMessage(msg.data, attributes)
The protobuf definition is here:-
https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
and so it looks as if the message_id and publish_time are not being parsed as they are seperate from the attributes. Perhaps the PubsubMessage class needs expanding to include these as attributes, or they would need adding to the dictionary for attributes. This would only need doing for the _from_proto_str as obviously they would not need to be populated when transmitting a message to PubSub.
My python is not great, I'm assuming the latter option would need to look something like this?
attributes = dict((key, msg.attributes[key]) for key in msg.attributes)
attributes.update({'message_id': msg.message_id, 'publish_time': msg.publish_time})
return PubsubMessage(msg.data, attributes)
On Wed, 2019-07-24 at 10:01 -0700, Ahmet Altay wrote:
This message originated from outside your organization
________________________________
No, I do not see any issues with the code you posted on stackoverflow. It might very well be an issue with Beam.
+Udi Meiri<ma...@google.com> is the author of this code, might have an idea.
On Wed, Jul 24, 2019 at 9:47 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Thanks Ahmet, I'm aware of the documentation and what should be happening, however, as you can see in the example I posted on stackoverflow, this does not appear to be the case, hence my question here! If there are any glaring issues with my code, I'd be grateful if you could point them out.
On Wed, 2019-07-24 at 09:16 -0700, Ahmet Altay wrote:
This message originated from outside your organization
________________________________
When with_attributes is set to True, the elements will be of type PubsubMessage [1]. I could not find a test/example for this, but documentation suggests [2], PubsubMessage will have an attributes map including the system provided values. One of those keys will be message_id [3].
[1] https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L156
[2] https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L94
[3] https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
On Wed, Jul 24, 2019 at 6:23 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Thank you both for you replies; apologies for the delay in responding, I was away from my machine over the weekend.
I'm still a bit confused over how the message id from PubSub is exposed.
From the documentation, I was expecting the PubSub message id to be in the attributes property of the message; but when running this into BigQuery as a string, the only properties are those that I pass in to my PubSub message as attributes. None of the examples you've linked use the with_attributes flag, unfortunately.
I've posted the code and output onto stackoverflow:- https://stackoverflow.com/questions/57183876/how-do-you-access-the-message-id-from-google-pub-sub-using-apache-beam
Secondly, is the 2.14 release available for preview?
Release candidate is not out yet. This will happen soon (~this week) and you can watch dev@ for this. If it makes sense for your case, you can also build yourself from head and test it out.
Kind regards,
Matthew
On Fri, 2019-07-19 at 13:22 -0700, Valentyn Tymofieiev wrote:
This message originated from outside your organization
________________________________
Also, see https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/complete/game/leader_board.py which involves both PubSub and Bigquery IOs.
On Fri, Jul 19, 2019 at 12:31 PM Pablo Estrada <pa...@google.com>> wrote:
Beam 2.14.0 will include support for writing files in the fileio module (the support will include GCS, local files, HDFS). It will also support streaming. The transform is still marked as experimental, and is likely to receive improvements - but you can check it out for your pipelines, and see if it helps you : )
Best
-P.
On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>> wrote:
As of today, Beam Python streaming does not support writing to GCS yet, which explains https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python .
You are right - id_label and timestamp_attribute does not work on Direct runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I checked with a few folks and that seems to be the current status, but you can still give them a try on Dataflow runner.
You may also find the following examples helpful:
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py (streaming pipeline).
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch pipeline)
https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o.
On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Hi Valentyn,
Thank you for your reply. I'm already using the with_attributes=True option, however this returns the attributes property of the JSON, i.e :-
{
"attributes": {
"source": "python"
}
My pipeline currently looks like this (id_label is commented out when running directly, as it causes a not implemented error):-
messages = (p
| 'Read From Pub Sub' >> ReadFromPubSub(subscription=known_args.input_subscription,with_attributes=True)#,id_label='message_id')
| 'Parse JSON' >> beam.Map(format_message_element)
My function to parse the message looks like this:-
def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
messagedict = json.loads(message.data)
rownumber = messagedict['rownumber']
fullmessage = {'data' : json.dumps(message.data),
'rownumber' : int(rownumber),
'attributes' : json.dumps(message.attributes),
'timestamp' : float(timestamp)}
logging.info<http://logging.info>(message.attributes)
logging.info<http://logging.info>(message)
return (rownumber, fullmessage)
I'm aware there are the id_label and with_timestamp parameters for the ReadFromPubSub method, however, these don't seem to work with the direct runner, as per https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22 which makes testing somewhat difficult.
My full code is attached, when running above 2.9.0 of the SDK I can't get passed the windowing function, due to an issue that appears related to this https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python and this https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue as I was receiving the following error when running on 2.12.0:
Cannot convert GlobalWindow to apache_beam.utils.windowed_value._IntervalWindowBase [while running 'generatedPtransform-150']
On 2.9.0 when running on the local runner, I receive the following output from the logging.info<http://logging.info> calls in format_message_element:
INFO:root:{u'source': u'python'}
INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
I was expecting the messageId and publishTime as part of the object returned; but as you can see there's nothing there for those attributes.
(The code does not quite map correctly to the BigQuery table so it fails inserts at that point, which I'm currently trying to resolve!)
Kind regards,
Matthew
On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
This message originated from outside your organization
________________________________
Hi Matthew,
Welcome to Beam!
Looking at Python PubSub IO API, you should be able to access id and timestamp by setting `with_attributes=True` when using `ReadFromPubSub` PTransform, see [1,2].
[1] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
[2] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Good morning,
I'm very new to Beam, and pretty new to Python so please first accept my apologies for any obvious misconceptions/mistakes in the following.
I am currently trying to develop a sample pipeline in Python to pull messages from Pub/Sub and then write them to either files in cloud storage or to BigQuery. The ultimate goal will be to utilise the pipeline for real time streaming of event data to BigQuery (with various transformations) but also to store the raw messages long term in files in cloud storage.
At the moment, I'm simply trying to parse the message to get the PubSub messageId and publishTime in order to be able to write them into the output. The json of my PubSub message looks like this:-
[
{
"ackId": "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
"message": {
"attributes": {
"source": "python"
},
"data": "eyJyb3dudW1iZXIiOiAyfQ==",
"messageId": "619310330691403",
"publishTime": "2019-07-12T08:27:58.522Z"
}
}
]
According to the documentation<https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html> the PubSub message payload returns the data and attributes properties; is there simply no way of retrieving the messageId and publishTime, or are these exposed somewhere else? If not, will the inclusion of these be in the roadmap, and are they available if using Java (I have zero Java experience hence why reaching for Python first).
Kind regards,
Matthew
---------- Forwarded message ----------
From: Domain postMaster address <po...@carfinance247.co.uk>>
To: "user@beam.apache.org<ma...@beam.apache.org>" <us...@beam.apache.org>>
Cc:
Bcc:
Date: Mon, 15 Jul 2019 09:12:15 +0000
Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
[https://static-uk.mimecast.com/mimecast/resources/images/notifications/mimecast-logo-254x120.png]
We removed a file from this message
Your organization's email policy doesn't permit this type of file. If you need it, please contact your administrator.
File Details
ApacheBeamDataflow-To-BigQuery2.py (6559 bytes)
[https://static-uk.mimecast.com/mimecast/resources/images/notifications/powered-mimecast-logo-278x28.png]
© 2003 - 2019 Mimecast Services Limited.
Disclaimer
The information contained in this communication from the sender is confidential. It is intended solely for use by the recipient and others authorized to receive it. If you are not the recipient, you are hereby notified that any disclosure, copying, distribution or taking action in relation of the contents of this information is strictly prohibited and may be unlawful.
This email has been scanned for viruses and malware, and may have been automatically archived by Mimecast Ltd, an innovator in Software as a Service (SaaS) for business. Providing a safer and more useful place for your human generated data. Specializing in; Security, archiving and compliance. To find out more visit the Mimecast website.
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Ahmet Altay <al...@google.com>.
No, I do not see any issues with the code you posted on stackoverflow. It
might very well be an issue with Beam.
+Udi Meiri <eh...@google.com> is the author of this code, might have an
idea.
On Wed, Jul 24, 2019 at 9:47 AM Matthew Darwin <
Matthew.Darwin@carfinance247.co.uk> wrote:
> Thanks Ahmet, I'm aware of the documentation and what should be happening,
> however, as you can see in the example I posted on stackoverflow, this does
> not appear to be the case, hence my question here! If there are any glaring
> issues with my code, I'd be grateful if you could point them out.
>
> On Wed, 2019-07-24 at 09:16 -0700, Ahmet Altay wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> When with_attributes is set to True, the elements will be of type
> PubsubMessage [1]. I could not find a test/example for this, but
> documentation suggests [2], PubsubMessage will have an attributes map
> including the system provided values. One of those keys will be message_id
> [3].
>
> [1]
> https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L156
> [2]
> https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L94
> [3]
> https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
>
> On Wed, Jul 24, 2019 at 6:23 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Thank you both for you replies; apologies for the delay in responding, I
> was away from my machine over the weekend.
>
> I'm still a bit confused over how the message id from PubSub is exposed.
>
> From the documentation, I was expecting the PubSub message id to be in the
> attributes property of the message; but when running this into BigQuery as
> a string, the only properties are those that I pass in to my PubSub message
> as attributes. None of the examples you've linked use the with_attributes
> flag, unfortunately.
>
> I've posted the code and output onto stackoverflow:-
> https://stackoverflow.com/questions/57183876/how-do-you-access-the-message-id-from-google-pub-sub-using-apache-beam
>
> Secondly, is the 2.14 release available for preview?
>
>
> Release candidate is not out yet. This will happen soon (~this week) and
> you can watch dev@ for this. If it makes sense for your case, you can
> also build yourself from head and test it out.
>
>
>
> Kind regards,
>
> Matthew
>
>
> On Fri, 2019-07-19 at 13:22 -0700, Valentyn Tymofieiev wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> Also, see
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/complete/game/leader_board.py which
> involves both PubSub and Bigquery IOs.
>
> On Fri, Jul 19, 2019 at 12:31 PM Pablo Estrada <pa...@google.com> wrote:
>
> Beam 2.14.0 will include support for writing files in the fileio module
> (the support will include GCS, local files, HDFS). It will also support
> streaming. The transform is still marked as experimental, and is likely to
> receive improvements - but you can check it out for your pipelines, and see
> if it helps you : )
> Best
> -P.
>
> On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>
> wrote:
>
> As of today, Beam Python streaming does not support writing to GCS yet,
> which explains
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python .
>
>
> You are right - id_label and timestamp_attribute does not work on Direct
> runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I
> checked with a few folks and that seems to be the current status, but you
> can still give them a try on Dataflow runner.
>
> You may also find the following examples helpful:
>
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py
> (streaming pipeline).
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch
> pipeline)
>
> https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o
> .
>
> On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Hi Valentyn,
>
> Thank you for your reply. I'm already using the with_attributes=True
> option, however this returns the attributes property of the JSON, i.e :-
>
> {
> "attributes": {
> "source": "python"
> }
>
>
>
> My pipeline currently looks like this (id_label is commented out when
> running directly, as it causes a not implemented error):-
>
> messages = (p
> | 'Read From Pub Sub' >> ReadFromPubSub(subscription
> =known_args.input_subscription,with_attributes=True)
> #,id_label='message_id')
> | 'Parse JSON' >> beam.Map(format_message_element)
>
> My function to parse the message looks like this:-
>
> def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
> messagedict = json.loads(message.data)
> rownumber = messagedict['rownumber']
> fullmessage = {'data' : json.dumps(message.data),
> 'rownumber' : int(rownumber),
> 'attributes' : json.dumps(message.attributes),
> 'timestamp' : float(timestamp)}
>
> logging.info(message.attributes)
> logging.info(message)
>
> return (rownumber, fullmessage)
>
> I'm aware there are the id_label and with_timestamp parameters for the
> ReadFromPubSub method, however, these don't seem to work with the direct
> runner, as per
> https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22
> which makes testing somewhat difficult.
>
> My full code is attached, when running above 2.9.0 of the SDK I can't get
> passed the windowing function, due to an issue that appears related to this
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
> and this
> https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue
> as I was receiving the following error when running on 2.12.0:
>
> Cannot convert GlobalWindow to
> apache_beam.utils.windowed_value._IntervalWindowBase [while running
> 'generatedPtransform-150']
>
> On 2.9.0 when running on the local runner, I receive the following output
> from the logging.info calls in format_message_element:
>
> INFO:root:{u'source': u'python'}
> INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
>
> I was expecting the messageId and publishTime as part of the object
> returned; but as you can see there's nothing there for those attributes.
>
> (The code does not quite map correctly to the BigQuery table so it fails
> inserts at that point, which I'm currently trying to resolve!)
>
> Kind regards,
>
> Matthew
>
>
>
> On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> Hi Matthew,
>
> Welcome to Beam!
>
> Looking at Python PubSub IO API, you should be able to access id and
> timestamp by setting `with_attributes=True` when using `ReadFromPubSub`
> PTransform, see [1,2].
>
> [1]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
> [2]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
>
> On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Good morning,
>
> I'm very new to Beam, and pretty new to Python so please first accept my
> apologies for any obvious misconceptions/mistakes in the following.
>
> I am currently trying to develop a sample pipeline in Python to pull
> messages from Pub/Sub and then write them to either files in cloud storage
> or to BigQuery. The ultimate goal will be to utilise the pipeline for real
> time streaming of event data to BigQuery (with various transformations) but
> also to store the raw messages long term in files in cloud storage.
>
> At the moment, I'm simply trying to parse the message to get the PubSub
> messageId and publishTime in order to be able to write them into the
> output. The json of my PubSub message looks like this:-
>
> [
> {
> "ackId":
> "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
> "message": {
> "attributes": {
> "source": "python"
> },
> "data": "eyJyb3dudW1iZXIiOiAyfQ==",
> "messageId": "619310330691403",
> "publishTime": "2019-07-12T08:27:58.522Z"
> }
> }
> ]
> According to the documentation
> <https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html>
> the PubSub message payload returns the *data* and *attributes*
> properties; is there simply no way of retrieving the messageId and
> publishTime, or are these exposed somewhere else? If not, will the
> inclusion of these be in the roadmap, and are they available if using Java
> (I have zero Java experience hence why reaching for Python first).
>
> Kind regards,
>
> Matthew
>
>
>
>
> ---------- Forwarded message ----------
> From: Domain postMaster address <po...@carfinance247.co.uk>
> To: "user@beam.apache.org" <us...@beam.apache.org>
> Cc:
> Bcc:
> Date: Mon, 15 Jul 2019 09:12:15 +0000
> Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
>
> [image: Logo]
> We removed a file from this message
>
> Your organization's email policy doesn't permit this type of file. If you
> need it, please contact your administrator.
>
>
> File Details
>
> *ApacheBeamDataflow-To-BigQuery2.py* (6559 bytes)
>
>
> © 2003 - 2019 Mimecast Services Limited.
>
>
>
>
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Matthew Darwin <Ma...@carfinance247.co.uk>.
Thanks Ahmet, I'm aware of the documentation and what should be happening, however, as you can see in the example I posted on stackoverflow, this does not appear to be the case, hence my question here! If there are any glaring issues with my code, I'd be grateful if you could point them out.
On Wed, 2019-07-24 at 09:16 -0700, Ahmet Altay wrote:
This message originated from outside your organization
________________________________
When with_attributes is set to True, the elements will be of type PubsubMessage [1]. I could not find a test/example for this, but documentation suggests [2], PubsubMessage will have an attributes map including the system provided values. One of those keys will be message_id [3].
[1] https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L156
[2] https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L94
[3] https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
On Wed, Jul 24, 2019 at 6:23 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Thank you both for you replies; apologies for the delay in responding, I was away from my machine over the weekend.
I'm still a bit confused over how the message id from PubSub is exposed.
From the documentation, I was expecting the PubSub message id to be in the attributes property of the message; but when running this into BigQuery as a string, the only properties are those that I pass in to my PubSub message as attributes. None of the examples you've linked use the with_attributes flag, unfortunately.
I've posted the code and output onto stackoverflow:- https://stackoverflow.com/questions/57183876/how-do-you-access-the-message-id-from-google-pub-sub-using-apache-beam
Secondly, is the 2.14 release available for preview?
Release candidate is not out yet. This will happen soon (~this week) and you can watch dev@ for this. If it makes sense for your case, you can also build yourself from head and test it out.
Kind regards,
Matthew
On Fri, 2019-07-19 at 13:22 -0700, Valentyn Tymofieiev wrote:
This message originated from outside your organization
________________________________
Also, see https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/complete/game/leader_board.py which involves both PubSub and Bigquery IOs.
On Fri, Jul 19, 2019 at 12:31 PM Pablo Estrada <pa...@google.com>> wrote:
Beam 2.14.0 will include support for writing files in the fileio module (the support will include GCS, local files, HDFS). It will also support streaming. The transform is still marked as experimental, and is likely to receive improvements - but you can check it out for your pipelines, and see if it helps you : )
Best
-P.
On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>> wrote:
As of today, Beam Python streaming does not support writing to GCS yet, which explains https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python .
You are right - id_label and timestamp_attribute does not work on Direct runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I checked with a few folks and that seems to be the current status, but you can still give them a try on Dataflow runner.
You may also find the following examples helpful:
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py (streaming pipeline).
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch pipeline)
https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o.
On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Hi Valentyn,
Thank you for your reply. I'm already using the with_attributes=True option, however this returns the attributes property of the JSON, i.e :-
{
"attributes": {
"source": "python"
}
My pipeline currently looks like this (id_label is commented out when running directly, as it causes a not implemented error):-
messages = (p
| 'Read From Pub Sub' >> ReadFromPubSub(subscription=known_args.input_subscription,with_attributes=True)#,id_label='message_id')
| 'Parse JSON' >> beam.Map(format_message_element)
My function to parse the message looks like this:-
def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
messagedict = json.loads(message.data)
rownumber = messagedict['rownumber']
fullmessage = {'data' : json.dumps(message.data),
'rownumber' : int(rownumber),
'attributes' : json.dumps(message.attributes),
'timestamp' : float(timestamp)}
logging.info<http://logging.info>(message.attributes)
logging.info<http://logging.info>(message)
return (rownumber, fullmessage)
I'm aware there are the id_label and with_timestamp parameters for the ReadFromPubSub method, however, these don't seem to work with the direct runner, as per https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22 which makes testing somewhat difficult.
My full code is attached, when running above 2.9.0 of the SDK I can't get passed the windowing function, due to an issue that appears related to this https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python and this https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue as I was receiving the following error when running on 2.12.0:
Cannot convert GlobalWindow to apache_beam.utils.windowed_value._IntervalWindowBase [while running 'generatedPtransform-150']
On 2.9.0 when running on the local runner, I receive the following output from the logging.info<http://logging.info> calls in format_message_element:
INFO:root:{u'source': u'python'}
INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
I was expecting the messageId and publishTime as part of the object returned; but as you can see there's nothing there for those attributes.
(The code does not quite map correctly to the BigQuery table so it fails inserts at that point, which I'm currently trying to resolve!)
Kind regards,
Matthew
On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
This message originated from outside your organization
________________________________
Hi Matthew,
Welcome to Beam!
Looking at Python PubSub IO API, you should be able to access id and timestamp by setting `with_attributes=True` when using `ReadFromPubSub` PTransform, see [1,2].
[1] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
[2] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Good morning,
I'm very new to Beam, and pretty new to Python so please first accept my apologies for any obvious misconceptions/mistakes in the following.
I am currently trying to develop a sample pipeline in Python to pull messages from Pub/Sub and then write them to either files in cloud storage or to BigQuery. The ultimate goal will be to utilise the pipeline for real time streaming of event data to BigQuery (with various transformations) but also to store the raw messages long term in files in cloud storage.
At the moment, I'm simply trying to parse the message to get the PubSub messageId and publishTime in order to be able to write them into the output. The json of my PubSub message looks like this:-
[
{
"ackId": "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
"message": {
"attributes": {
"source": "python"
},
"data": "eyJyb3dudW1iZXIiOiAyfQ==",
"messageId": "619310330691403",
"publishTime": "2019-07-12T08:27:58.522Z"
}
}
]
According to the documentation<https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html> the PubSub message payload returns the data and attributes properties; is there simply no way of retrieving the messageId and publishTime, or are these exposed somewhere else? If not, will the inclusion of these be in the roadmap, and are they available if using Java (I have zero Java experience hence why reaching for Python first).
Kind regards,
Matthew
---------- Forwarded message ----------
From: Domain postMaster address <po...@carfinance247.co.uk>>
To: "user@beam.apache.org<ma...@beam.apache.org>" <us...@beam.apache.org>>
Cc:
Bcc:
Date: Mon, 15 Jul 2019 09:12:15 +0000
Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
[Logo]
We removed a file from this message
Your organization's email policy doesn't permit this type of file. If you need it, please contact your administrator.
File Details
ApacheBeamDataflow-To-BigQuery2.py (6559 bytes)
[https://static-uk.mimecast.com/mimecast/resources/images/notifications/powered-mimecast-logo-278x28.png]
© 2003 - 2019 Mimecast Services Limited.
Disclaimer
The information contained in this communication from the sender is confidential. It is intended solely for use by the recipient and others authorized to receive it. If you are not the recipient, you are hereby notified that any disclosure, copying, distribution or taking action in relation of the contents of this information is strictly prohibited and may be unlawful.
This email has been scanned for viruses and malware, and may have been automatically archived by Mimecast Ltd, an innovator in Software as a Service (SaaS) for business. Providing a safer and more useful place for your human generated data. Specializing in; Security, archiving and compliance. To find out more visit the Mimecast website.
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Ahmet Altay <al...@google.com>.
When with_attributes is set to True, the elements will be of type
PubsubMessage [1]. I could not find a test/example for this, but
documentation suggests [2], PubsubMessage will have an attributes map
including the system provided values. One of those keys will be message_id
[3].
[1]
https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L156
[2]
https://github.com/apache/beam/blob/a5ed104aeac91781492523808fa4d3df6f2e55f7/sdks/python/apache_beam/io/gcp/pubsub.py#L94
[3]
https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.PubsubMessage
On Wed, Jul 24, 2019 at 6:23 AM Matthew Darwin <
Matthew.Darwin@carfinance247.co.uk> wrote:
> Thank you both for you replies; apologies for the delay in responding, I
> was away from my machine over the weekend.
>
> I'm still a bit confused over how the message id from PubSub is exposed.
>
> From the documentation, I was expecting the PubSub message id to be in the
> attributes property of the message; but when running this into BigQuery as
> a string, the only properties are those that I pass in to my PubSub message
> as attributes. None of the examples you've linked use the with_attributes
> flag, unfortunately.
>
> I've posted the code and output onto stackoverflow:-
> https://stackoverflow.com/questions/57183876/how-do-you-access-the-message-id-from-google-pub-sub-using-apache-beam
>
> Secondly, is the 2.14 release available for preview?
>
Release candidate is not out yet. This will happen soon (~this week) and
you can watch dev@ for this. If it makes sense for your case, you can also
build yourself from head and test it out.
>
> Kind regards,
>
> Matthew
>
>
> On Fri, 2019-07-19 at 13:22 -0700, Valentyn Tymofieiev wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> Also, see
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/complete/game/leader_board.py which
> involves both PubSub and Bigquery IOs.
>
> On Fri, Jul 19, 2019 at 12:31 PM Pablo Estrada <pa...@google.com> wrote:
>
> Beam 2.14.0 will include support for writing files in the fileio module
> (the support will include GCS, local files, HDFS). It will also support
> streaming. The transform is still marked as experimental, and is likely to
> receive improvements - but you can check it out for your pipelines, and see
> if it helps you : )
> Best
> -P.
>
> On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>
> wrote:
>
> As of today, Beam Python streaming does not support writing to GCS yet,
> which explains
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
> .
>
> You are right - id_label and timestamp_attribute does not work on Direct
> runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I
> checked with a few folks and that seems to be the current status, but you
> can still give them a try on Dataflow runner.
>
> You may also find the following examples helpful:
>
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py
> (streaming pipeline).
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch
> pipeline)
>
> https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o
> .
>
> On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Hi Valentyn,
>
> Thank you for your reply. I'm already using the with_attributes=True
> option, however this returns the attributes property of the JSON, i.e :-
>
> {
> "attributes": {
> "source": "python"
> }
>
>
>
> My pipeline currently looks like this (id_label is commented out when
> running directly, as it causes a not implemented error):-
>
> messages = (p
> | 'Read From Pub Sub' >> ReadFromPubSub(subscription
> =known_args.input_subscription,with_attributes=True)
> #,id_label='message_id')
> | 'Parse JSON' >> beam.Map(format_message_element)
>
> My function to parse the message looks like this:-
>
> def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
> messagedict = json.loads(message.data)
> rownumber = messagedict['rownumber']
> fullmessage = {'data' : json.dumps(message.data),
> 'rownumber' : int(rownumber),
> 'attributes' : json.dumps(message.attributes),
> 'timestamp' : float(timestamp)}
>
> logging.info(message.attributes)
> logging.info(message)
>
> return (rownumber, fullmessage)
>
> I'm aware there are the id_label and with_timestamp parameters for the
> ReadFromPubSub method, however, these don't seem to work with the direct
> runner, as per
> https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22
> which makes testing somewhat difficult.
>
> My full code is attached, when running above 2.9.0 of the SDK I can't get
> passed the windowing function, due to an issue that appears related to this
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
> and this
> https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue
> as I was receiving the following error when running on 2.12.0:
>
> Cannot convert GlobalWindow to
> apache_beam.utils.windowed_value._IntervalWindowBase [while running
> 'generatedPtransform-150']
>
> On 2.9.0 when running on the local runner, I receive the following output
> from the logging.info calls in format_message_element:
>
> INFO:root:{u'source': u'python'}
> INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
>
> I was expecting the messageId and publishTime as part of the object
> returned; but as you can see there's nothing there for those attributes.
>
> (The code does not quite map correctly to the BigQuery table so it fails
> inserts at that point, which I'm currently trying to resolve!)
>
> Kind regards,
>
> Matthew
>
>
>
> On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> Hi Matthew,
>
> Welcome to Beam!
>
> Looking at Python PubSub IO API, you should be able to access id and
> timestamp by setting `with_attributes=True` when using `ReadFromPubSub`
> PTransform, see [1,2].
>
> [1]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
> [2]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
>
> On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Good morning,
>
> I'm very new to Beam, and pretty new to Python so please first accept my
> apologies for any obvious misconceptions/mistakes in the following.
>
> I am currently trying to develop a sample pipeline in Python to pull
> messages from Pub/Sub and then write them to either files in cloud storage
> or to BigQuery. The ultimate goal will be to utilise the pipeline for real
> time streaming of event data to BigQuery (with various transformations) but
> also to store the raw messages long term in files in cloud storage.
>
> At the moment, I'm simply trying to parse the message to get the PubSub
> messageId and publishTime in order to be able to write them into the
> output. The json of my PubSub message looks like this:-
>
> [
> {
> "ackId":
> "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
> "message": {
> "attributes": {
> "source": "python"
> },
> "data": "eyJyb3dudW1iZXIiOiAyfQ==",
> "messageId": "619310330691403",
> "publishTime": "2019-07-12T08:27:58.522Z"
> }
> }
> ]
> According to the documentation
> <https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html>
> the PubSub message payload returns the *data* and *attributes*
> properties; is there simply no way of retrieving the messageId and
> publishTime, or are these exposed somewhere else? If not, will the
> inclusion of these be in the roadmap, and are they available if using Java
> (I have zero Java experience hence why reaching for Python first).
>
> Kind regards,
>
> Matthew
>
>
>
>
> ---------- Forwarded message ----------
> From: Domain postMaster address <po...@carfinance247.co.uk>
> To: "user@beam.apache.org" <us...@beam.apache.org>
> Cc:
> Bcc:
> Date: Mon, 15 Jul 2019 09:12:15 +0000
> Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
>
> [image: Logo]
> We removed a file from this message
>
> Your organization's email policy doesn't permit this type of file. If you
> need it, please contact your administrator.
>
>
> File Details
>
> *ApacheBeamDataflow-To-BigQuery2.py* (6559 bytes)
>
>
> © 2003 - 2019 Mimecast Services Limited.
>
>
>
>
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Matthew Darwin <Ma...@carfinance247.co.uk>.
Thank you both for you replies; apologies for the delay in responding, I was away from my machine over the weekend.
I'm still a bit confused over how the message id from PubSub is exposed.
From the documentation, I was expecting the PubSub message id to be in the attributes property of the message; but when running this into BigQuery as a string, the only properties are those that I pass in to my PubSub message as attributes. None of the examples you've linked use the with_attributes flag, unfortunately.
I've posted the code and output onto stackoverflow:- https://stackoverflow.com/questions/57183876/how-do-you-access-the-message-id-from-google-pub-sub-using-apache-beam
Secondly, is the 2.14 release available for preview?
Kind regards,
Matthew
On Fri, 2019-07-19 at 13:22 -0700, Valentyn Tymofieiev wrote:
This message originated from outside your organization
________________________________
Also, see https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/complete/game/leader_board.py which involves both PubSub and Bigquery IOs.
On Fri, Jul 19, 2019 at 12:31 PM Pablo Estrada <pa...@google.com>> wrote:
Beam 2.14.0 will include support for writing files in the fileio module (the support will include GCS, local files, HDFS). It will also support streaming. The transform is still marked as experimental, and is likely to receive improvements - but you can check it out for your pipelines, and see if it helps you : )
Best
-P.
On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>> wrote:
As of today, Beam Python streaming does not support writing to GCS yet, which explains https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python .
You are right - id_label and timestamp_attribute does not work on Direct runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I checked with a few folks and that seems to be the current status, but you can still give them a try on Dataflow runner.
You may also find the following examples helpful:
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py (streaming pipeline).
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch pipeline)
https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o.
On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Hi Valentyn,
Thank you for your reply. I'm already using the with_attributes=True option, however this returns the attributes property of the JSON, i.e :-
{
"attributes": {
"source": "python"
}
My pipeline currently looks like this (id_label is commented out when running directly, as it causes a not implemented error):-
messages = (p
| 'Read From Pub Sub' >> ReadFromPubSub(subscription=known_args.input_subscription,with_attributes=True)#,id_label='message_id')
| 'Parse JSON' >> beam.Map(format_message_element)
My function to parse the message looks like this:-
def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
messagedict = json.loads(message.data)
rownumber = messagedict['rownumber']
fullmessage = {'data' : json.dumps(message.data),
'rownumber' : int(rownumber),
'attributes' : json.dumps(message.attributes),
'timestamp' : float(timestamp)}
logging.info<http://logging.info>(message.attributes)
logging.info<http://logging.info>(message)
return (rownumber, fullmessage)
I'm aware there are the id_label and with_timestamp parameters for the ReadFromPubSub method, however, these don't seem to work with the direct runner, as per https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22 which makes testing somewhat difficult.
My full code is attached, when running above 2.9.0 of the SDK I can't get passed the windowing function, due to an issue that appears related to this https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python and this https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue as I was receiving the following error when running on 2.12.0:
Cannot convert GlobalWindow to apache_beam.utils.windowed_value._IntervalWindowBase [while running 'generatedPtransform-150']
On 2.9.0 when running on the local runner, I receive the following output from the logging.info<http://logging.info> calls in format_message_element:
INFO:root:{u'source': u'python'}
INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
I was expecting the messageId and publishTime as part of the object returned; but as you can see there's nothing there for those attributes.
(The code does not quite map correctly to the BigQuery table so it fails inserts at that point, which I'm currently trying to resolve!)
Kind regards,
Matthew
On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
This message originated from outside your organization
________________________________
Hi Matthew,
Welcome to Beam!
Looking at Python PubSub IO API, you should be able to access id and timestamp by setting `with_attributes=True` when using `ReadFromPubSub` PTransform, see [1,2].
[1] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
[2] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Good morning,
I'm very new to Beam, and pretty new to Python so please first accept my apologies for any obvious misconceptions/mistakes in the following.
I am currently trying to develop a sample pipeline in Python to pull messages from Pub/Sub and then write them to either files in cloud storage or to BigQuery. The ultimate goal will be to utilise the pipeline for real time streaming of event data to BigQuery (with various transformations) but also to store the raw messages long term in files in cloud storage.
At the moment, I'm simply trying to parse the message to get the PubSub messageId and publishTime in order to be able to write them into the output. The json of my PubSub message looks like this:-
[
{
"ackId": "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
"message": {
"attributes": {
"source": "python"
},
"data": "eyJyb3dudW1iZXIiOiAyfQ==",
"messageId": "619310330691403",
"publishTime": "2019-07-12T08:27:58.522Z"
}
}
]
According to the documentation<https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html> the PubSub message payload returns the data and attributes properties; is there simply no way of retrieving the messageId and publishTime, or are these exposed somewhere else? If not, will the inclusion of these be in the roadmap, and are they available if using Java (I have zero Java experience hence why reaching for Python first).
Kind regards,
Matthew
---------- Forwarded message ----------
From: Domain postMaster address <po...@carfinance247.co.uk>>
To: "user@beam.apache.org<ma...@beam.apache.org>" <us...@beam.apache.org>>
Cc:
Bcc:
Date: Mon, 15 Jul 2019 09:12:15 +0000
Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
[https://static-uk.mimecast.com/mimecast/resources/images/notifications/mimecast-logo-254x120.png]
We removed a file from this message
Your organization's email policy doesn't permit this type of file. If you need it, please contact your administrator.
File Details
ApacheBeamDataflow-To-BigQuery2.py (6559 bytes)
[https://static-uk.mimecast.com/mimecast/resources/images/notifications/powered-mimecast-logo-278x28.png]
© 2003 - 2019 Mimecast Services Limited.
Disclaimer
The information contained in this communication from the sender is confidential. It is intended solely for use by the recipient and others authorized to receive it. If you are not the recipient, you are hereby notified that any disclosure, copying, distribution or taking action in relation of the contents of this information is strictly prohibited and may be unlawful.
This email has been scanned for viruses and malware, and may have been automatically archived by Mimecast Ltd, an innovator in Software as a Service (SaaS) for business. Providing a safer and more useful place for your human generated data. Specializing in; Security, archiving and compliance. To find out more visit the Mimecast website.
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Valentyn Tymofieiev <va...@google.com>.
Also, see
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/complete/game/leader_board.py
which
involves both PubSub and Bigquery IOs.
On Fri, Jul 19, 2019 at 12:31 PM Pablo Estrada <pa...@google.com> wrote:
> Beam 2.14.0 will include support for writing files in the fileio module
> (the support will include GCS, local files, HDFS). It will also support
> streaming. The transform is still marked as experimental, and is likely to
> receive improvements - but you can check it out for your pipelines, and see
> if it helps you : )
> Best
> -P.
>
> On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>
> wrote:
>
>> As of today, Beam Python streaming does not support writing to GCS yet,
>> which explains
>> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
>> .
>>
>> You are right - id_label and timestamp_attribute does not work on Direct
>> runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I
>> checked with a few folks and that seems to be the current status, but you
>> can still give them a try on Dataflow runner.
>>
>> You may also find the following examples helpful:
>>
>>
>> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py
>> (streaming pipeline).
>>
>> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch
>> pipeline)
>>
>> https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o
>> .
>>
>> On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <
>> Matthew.Darwin@carfinance247.co.uk> wrote:
>>
>>> Hi Valentyn,
>>>
>>> Thank you for your reply. I'm already using the with_attributes=True
>>> option, however this returns the attributes property of the JSON, i.e :-
>>>
>>> {
>>>> "attributes": {
>>>> "source": "python"
>>>> }
>>>>
>>>
>>>
>>> My pipeline currently looks like this (id_label is commented out when
>>> running directly, as it causes a not implemented error):-
>>>
>>> messages = (p
>>> | 'Read From Pub Sub' >> ReadFromPubSub(subscription
>>> =known_args.input_subscription,with_attributes=True)
>>> #,id_label='message_id')
>>> | 'Parse JSON' >> beam.Map(format_message_element)
>>>
>>> My function to parse the message looks like this:-
>>>
>>> def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
>>> messagedict = json.loads(message.data)
>>> rownumber = messagedict['rownumber']
>>> fullmessage = {'data' : json.dumps(message.data),
>>> 'rownumber' : int(rownumber),
>>> 'attributes' : json.dumps(message.attributes),
>>> 'timestamp' : float(timestamp)}
>>>
>>> logging.info(message.attributes)
>>> logging.info(message)
>>>
>>> return (rownumber, fullmessage)
>>>
>>> I'm aware there are the id_label and with_timestamp parameters for the
>>> ReadFromPubSub method, however, these don't seem to work with the direct
>>> runner, as per
>>> https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22
>>> which makes testing somewhat difficult.
>>>
>>> My full code is attached, when running above 2.9.0 of the SDK I can't
>>> get passed the windowing function, due to an issue that appears related to
>>> this
>>> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
>>> and this
>>> https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue
>>> as I was receiving the following error when running on 2.12.0:
>>>
>>> Cannot convert GlobalWindow to
>>> apache_beam.utils.windowed_value._IntervalWindowBase [while running
>>> 'generatedPtransform-150']
>>>
>>> On 2.9.0 when running on the local runner, I receive the following
>>> output from the logging.info calls in format_message_element:
>>>
>>> INFO:root:{u'source': u'python'}
>>> INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
>>>
>>> I was expecting the messageId and publishTime as part of the object
>>> returned; but as you can see there's nothing there for those attributes.
>>>
>>> (The code does not quite map correctly to the BigQuery table so it fails
>>> inserts at that point, which I'm currently trying to resolve!)
>>>
>>> Kind regards,
>>>
>>> Matthew
>>>
>>>
>>>
>>> On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
>>>
>>> *This message originated from outside your organization*
>>> ------------------------------
>>> Hi Matthew,
>>>
>>> Welcome to Beam!
>>>
>>> Looking at Python PubSub IO API, you should be able to access id and
>>> timestamp by setting `with_attributes=True` when using `ReadFromPubSub`
>>> PTransform, see [1,2].
>>>
>>> [1]
>>> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
>>> [2]
>>> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
>>>
>>> On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <
>>> Matthew.Darwin@carfinance247.co.uk> wrote:
>>>
>>> Good morning,
>>>
>>> I'm very new to Beam, and pretty new to Python so please first accept my
>>> apologies for any obvious misconceptions/mistakes in the following.
>>>
>>> I am currently trying to develop a sample pipeline in Python to pull
>>> messages from Pub/Sub and then write them to either files in cloud storage
>>> or to BigQuery. The ultimate goal will be to utilise the pipeline for real
>>> time streaming of event data to BigQuery (with various transformations) but
>>> also to store the raw messages long term in files in cloud storage.
>>>
>>> At the moment, I'm simply trying to parse the message to get the PubSub
>>> messageId and publishTime in order to be able to write them into the
>>> output. The json of my PubSub message looks like this:-
>>>
>>> [
>>> {
>>> "ackId":
>>> "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
>>> "message": {
>>> "attributes": {
>>> "source": "python"
>>> },
>>> "data": "eyJyb3dudW1iZXIiOiAyfQ==",
>>> "messageId": "619310330691403",
>>> "publishTime": "2019-07-12T08:27:58.522Z"
>>> }
>>> }
>>> ]
>>> According to the documentation
>>> <https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html>
>>> the PubSub message payload returns the *data* and *attributes*
>>> properties; is there simply no way of retrieving the messageId and
>>> publishTime, or are these exposed somewhere else? If not, will the
>>> inclusion of these be in the roadmap, and are they available if using Java
>>> (I have zero Java experience hence why reaching for Python first).
>>>
>>> Kind regards,
>>>
>>> Matthew
>>>
>>>
>>>
>>>
>>> ---------- Forwarded message ----------
>>> From: Domain postMaster address <po...@carfinance247.co.uk>
>>> To: "user@beam.apache.org" <us...@beam.apache.org>
>>> Cc:
>>> Bcc:
>>> Date: Mon, 15 Jul 2019 09:12:15 +0000
>>> Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
>>>
>>> [image: Logo]
>>> We removed a file from this message
>>>
>>> Your organization's email policy doesn't permit this type of file. If
>>> you need it, please contact your administrator.
>>>
>>>
>>> File Details
>>>
>>> *ApacheBeamDataflow-To-BigQuery2.py* (6559 bytes)
>>>
>>>
>>> © 2003 - 2019 Mimecast Services Limited.
>>>
>>>
>>>
>>
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Pablo Estrada <pa...@google.com>.
Beam 2.14.0 will include support for writing files in the fileio module
(the support will include GCS, local files, HDFS). It will also support
streaming. The transform is still marked as experimental, and is likely to
receive improvements - but you can check it out for your pipelines, and see
if it helps you : )
Best
-P.
On Fri, Jul 19, 2019 at 12:24 PM Valentyn Tymofieiev <va...@google.com>
wrote:
> As of today, Beam Python streaming does not support writing to GCS yet,
> which explains
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
> .
>
> You are right - id_label and timestamp_attribute does not work on Direct
> runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I
> checked with a few folks and that seems to be the current status, but you
> can still give them a try on Dataflow runner.
>
> You may also find the following examples helpful:
>
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py
> (streaming pipeline).
>
> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py (batch
> pipeline)
>
> https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o
> .
>
> On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
>> Hi Valentyn,
>>
>> Thank you for your reply. I'm already using the with_attributes=True
>> option, however this returns the attributes property of the JSON, i.e :-
>>
>> {
>>> "attributes": {
>>> "source": "python"
>>> }
>>>
>>
>>
>> My pipeline currently looks like this (id_label is commented out when
>> running directly, as it causes a not implemented error):-
>>
>> messages = (p
>> | 'Read From Pub Sub' >> ReadFromPubSub(subscription
>> =known_args.input_subscription,with_attributes=True)
>> #,id_label='message_id')
>> | 'Parse JSON' >> beam.Map(format_message_element)
>>
>> My function to parse the message looks like this:-
>>
>> def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
>> messagedict = json.loads(message.data)
>> rownumber = messagedict['rownumber']
>> fullmessage = {'data' : json.dumps(message.data),
>> 'rownumber' : int(rownumber),
>> 'attributes' : json.dumps(message.attributes),
>> 'timestamp' : float(timestamp)}
>>
>> logging.info(message.attributes)
>> logging.info(message)
>>
>> return (rownumber, fullmessage)
>>
>> I'm aware there are the id_label and with_timestamp parameters for the
>> ReadFromPubSub method, however, these don't seem to work with the direct
>> runner, as per
>> https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22
>> which makes testing somewhat difficult.
>>
>> My full code is attached, when running above 2.9.0 of the SDK I can't get
>> passed the windowing function, due to an issue that appears related to this
>> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
>> and this
>> https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue
>> as I was receiving the following error when running on 2.12.0:
>>
>> Cannot convert GlobalWindow to
>> apache_beam.utils.windowed_value._IntervalWindowBase [while running
>> 'generatedPtransform-150']
>>
>> On 2.9.0 when running on the local runner, I receive the following output
>> from the logging.info calls in format_message_element:
>>
>> INFO:root:{u'source': u'python'}
>> INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
>>
>> I was expecting the messageId and publishTime as part of the object
>> returned; but as you can see there's nothing there for those attributes.
>>
>> (The code does not quite map correctly to the BigQuery table so it fails
>> inserts at that point, which I'm currently trying to resolve!)
>>
>> Kind regards,
>>
>> Matthew
>>
>>
>>
>> On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
>>
>> *This message originated from outside your organization*
>> ------------------------------
>> Hi Matthew,
>>
>> Welcome to Beam!
>>
>> Looking at Python PubSub IO API, you should be able to access id and
>> timestamp by setting `with_attributes=True` when using `ReadFromPubSub`
>> PTransform, see [1,2].
>>
>> [1]
>> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
>> [2]
>> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
>>
>> On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <
>> Matthew.Darwin@carfinance247.co.uk> wrote:
>>
>> Good morning,
>>
>> I'm very new to Beam, and pretty new to Python so please first accept my
>> apologies for any obvious misconceptions/mistakes in the following.
>>
>> I am currently trying to develop a sample pipeline in Python to pull
>> messages from Pub/Sub and then write them to either files in cloud storage
>> or to BigQuery. The ultimate goal will be to utilise the pipeline for real
>> time streaming of event data to BigQuery (with various transformations) but
>> also to store the raw messages long term in files in cloud storage.
>>
>> At the moment, I'm simply trying to parse the message to get the PubSub
>> messageId and publishTime in order to be able to write them into the
>> output. The json of my PubSub message looks like this:-
>>
>> [
>> {
>> "ackId":
>> "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
>> "message": {
>> "attributes": {
>> "source": "python"
>> },
>> "data": "eyJyb3dudW1iZXIiOiAyfQ==",
>> "messageId": "619310330691403",
>> "publishTime": "2019-07-12T08:27:58.522Z"
>> }
>> }
>> ]
>> According to the documentation
>> <https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html>
>> the PubSub message payload returns the *data* and *attributes*
>> properties; is there simply no way of retrieving the messageId and
>> publishTime, or are these exposed somewhere else? If not, will the
>> inclusion of these be in the roadmap, and are they available if using Java
>> (I have zero Java experience hence why reaching for Python first).
>>
>> Kind regards,
>>
>> Matthew
>>
>>
>>
>>
>> ---------- Forwarded message ----------
>> From: Domain postMaster address <po...@carfinance247.co.uk>
>> To: "user@beam.apache.org" <us...@beam.apache.org>
>> Cc:
>> Bcc:
>> Date: Mon, 15 Jul 2019 09:12:15 +0000
>> Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
>>
>> [image: Logo]
>> We removed a file from this message
>>
>> Your organization's email policy doesn't permit this type of file. If you
>> need it, please contact your administrator.
>>
>>
>> File Details
>>
>> *ApacheBeamDataflow-To-BigQuery2.py* (6559 bytes)
>>
>>
>> © 2003 - 2019 Mimecast Services Limited.
>>
>>
>>
>
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Valentyn Tymofieiev <va...@google.com>.
As of today, Beam Python streaming does not support writing to GCS yet,
which explains
https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
.
You are right - id_label and timestamp_attribute does not work on Direct
runner yet as per https://issues.apache.org/jira/browse/BEAM-4275, I
checked with a few folks and that seems to be the current status, but you
can still give them a try on Dataflow runner.
You may also find the following examples helpful:
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py
(streaming pipeline).
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/bigquery_tornadoes.py
(batch
pipeline)
https://stackoverflow.com/questions/46854167/dataflow-streaming-using-python-sdk-transform-for-pubsub-messages-to-bigquery-o
.
On Mon, Jul 15, 2019 at 2:12 AM Matthew Darwin <
Matthew.Darwin@carfinance247.co.uk> wrote:
> Hi Valentyn,
>
> Thank you for your reply. I'm already using the with_attributes=True
> option, however this returns the attributes property of the JSON, i.e :-
>
> {
>> "attributes": {
>> "source": "python"
>> }
>>
>
>
> My pipeline currently looks like this (id_label is commented out when
> running directly, as it causes a not implemented error):-
>
> messages = (p
> | 'Read From Pub Sub' >> ReadFromPubSub(subscription
> =known_args.input_subscription,with_attributes=True)
> #,id_label='message_id')
> | 'Parse JSON' >> beam.Map(format_message_element)
>
> My function to parse the message looks like this:-
>
> def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
> messagedict = json.loads(message.data)
> rownumber = messagedict['rownumber']
> fullmessage = {'data' : json.dumps(message.data),
> 'rownumber' : int(rownumber),
> 'attributes' : json.dumps(message.attributes),
> 'timestamp' : float(timestamp)}
>
> logging.info(message.attributes)
> logging.info(message)
>
> return (rownumber, fullmessage)
>
> I'm aware there are the id_label and with_timestamp parameters for the
> ReadFromPubSub method, however, these don't seem to work with the direct
> runner, as per
> https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22
> which makes testing somewhat difficult.
>
> My full code is attached, when running above 2.9.0 of the SDK I can't get
> passed the windowing function, due to an issue that appears related to this
> https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python
> and this
> https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue
> as I was receiving the following error when running on 2.12.0:
>
> Cannot convert GlobalWindow to
> apache_beam.utils.windowed_value._IntervalWindowBase [while running
> 'generatedPtransform-150']
>
> On 2.9.0 when running on the local runner, I receive the following output
> from the logging.info calls in format_message_element:
>
> INFO:root:{u'source': u'python'}
> INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
>
> I was expecting the messageId and publishTime as part of the object
> returned; but as you can see there's nothing there for those attributes.
>
> (The code does not quite map correctly to the BigQuery table so it fails
> inserts at that point, which I'm currently trying to resolve!)
>
> Kind regards,
>
> Matthew
>
>
>
> On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
>
> *This message originated from outside your organization*
> ------------------------------
> Hi Matthew,
>
> Welcome to Beam!
>
> Looking at Python PubSub IO API, you should be able to access id and
> timestamp by setting `with_attributes=True` when using `ReadFromPubSub`
> PTransform, see [1,2].
>
> [1]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
> [2]
> https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
>
> On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <
> Matthew.Darwin@carfinance247.co.uk> wrote:
>
> Good morning,
>
> I'm very new to Beam, and pretty new to Python so please first accept my
> apologies for any obvious misconceptions/mistakes in the following.
>
> I am currently trying to develop a sample pipeline in Python to pull
> messages from Pub/Sub and then write them to either files in cloud storage
> or to BigQuery. The ultimate goal will be to utilise the pipeline for real
> time streaming of event data to BigQuery (with various transformations) but
> also to store the raw messages long term in files in cloud storage.
>
> At the moment, I'm simply trying to parse the message to get the PubSub
> messageId and publishTime in order to be able to write them into the
> output. The json of my PubSub message looks like this:-
>
> [
> {
> "ackId":
> "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
> "message": {
> "attributes": {
> "source": "python"
> },
> "data": "eyJyb3dudW1iZXIiOiAyfQ==",
> "messageId": "619310330691403",
> "publishTime": "2019-07-12T08:27:58.522Z"
> }
> }
> ]
> According to the documentation
> <https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html>
> the PubSub message payload returns the *data* and *attributes*
> properties; is there simply no way of retrieving the messageId and
> publishTime, or are these exposed somewhere else? If not, will the
> inclusion of these be in the roadmap, and are they available if using Java
> (I have zero Java experience hence why reaching for Python first).
>
> Kind regards,
>
> Matthew
>
>
>
>
> ---------- Forwarded message ----------
> From: Domain postMaster address <po...@carfinance247.co.uk>
> To: "user@beam.apache.org" <us...@beam.apache.org>
> Cc:
> Bcc:
> Date: Mon, 15 Jul 2019 09:12:15 +0000
> Subject: ApacheBeamDataflow-To-BigQuery2_py was removed from this message
>
> [image: Logo]
> We removed a file from this message
>
> Your organization's email policy doesn't permit this type of file. If you
> need it, please contact your administrator.
>
>
> File Details
>
> *ApacheBeamDataflow-To-BigQuery2.py* (6559 bytes)
>
>
> © 2003 - 2019 Mimecast Services Limited.
>
>
>
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Matthew Darwin <Ma...@carfinance247.co.uk>.
Hi Valentyn,
Thank you for your reply. I'm already using the with_attributes=True option, however this returns the attributes property of the JSON, i.e :-
{
"attributes": {
"source": "python"
}
My pipeline currently looks like this (id_label is commented out when running directly, as it causes a not implemented error):-
messages = (p
| 'Read From Pub Sub' >> ReadFromPubSub(subscription=known_args.input_subscription,with_attributes=True)#,id_label='message_id')
| 'Parse JSON' >> beam.Map(format_message_element)
My function to parse the message looks like this:-
def format_message_element(message, timestamp=beam.DoFn.TimestampParam):
messagedict = json.loads(message.data)
rownumber = messagedict['rownumber']
fullmessage = {'data' : json.dumps(message.data),
'rownumber' : int(rownumber),
'attributes' : json.dumps(message.attributes),
'timestamp' : float(timestamp)}
logging.info(message.attributes)
logging.info(message)
return (rownumber, fullmessage)
I'm aware there are the id_label and with_timestamp parameters for the ReadFromPubSub method, however, these don't seem to work with the direct runner, as per https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22 which makes testing somewhat difficult.
My full code is attached, when running above 2.9.0 of the SDK I can't get passed the windowing function, due to an issue that appears related to this https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python and this https://stackoverflow.com/questions/55109403/apache-beam-python-sdk-upgrade-issue as I was receiving the following error when running on 2.12.0:
Cannot convert GlobalWindow to apache_beam.utils.windowed_value._IntervalWindowBase [while running 'generatedPtransform-150']
On 2.9.0 when running on the local runner, I receive the following output from the logging.info calls in format_message_element:
INFO:root:{u'source': u'python'}
INFO:root:PubsubMessage({"rownumber": 1}, {u'source': u'python'})
I was expecting the messageId and publishTime as part of the object returned; but as you can see there's nothing there for those attributes.
(The code does not quite map correctly to the BigQuery table so it fails inserts at that point, which I'm currently trying to resolve!)
Kind regards,
Matthew
On Fri, 2019-07-12 at 09:13 -0700, Valentyn Tymofieiev wrote:
This message originated from outside your organization
________________________________
Hi Matthew,
Welcome to Beam!
Looking at Python PubSub IO API, you should be able to access id and timestamp by setting `with_attributes=True` when using `ReadFromPubSub` PTransform, see [1,2].
[1] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
[2] https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <Ma...@carfinance247.co.uk>> wrote:
Good morning,
I'm very new to Beam, and pretty new to Python so please first accept my apologies for any obvious misconceptions/mistakes in the following.
I am currently trying to develop a sample pipeline in Python to pull messages from Pub/Sub and then write them to either files in cloud storage or to BigQuery. The ultimate goal will be to utilise the pipeline for real time streaming of event data to BigQuery (with various transformations) but also to store the raw messages long term in files in cloud storage.
At the moment, I'm simply trying to parse the message to get the PubSub messageId and publishTime in order to be able to write them into the output. The json of my PubSub message looks like this:-
[
{
"ackId": "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
"message": {
"attributes": {
"source": "python"
},
"data": "eyJyb3dudW1iZXIiOiAyfQ==",
"messageId": "619310330691403",
"publishTime": "2019-07-12T08:27:58.522Z"
}
}
]
According to the documentation<https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html> the PubSub message payload returns the data and attributes properties; is there simply no way of retrieving the messageId and publishTime, or are these exposed somewhere else? If not, will the inclusion of these be in the roadmap, and are they available if using Java (I have zero Java experience hence why reaching for Python first).
Kind regards,
Matthew
Disclaimer
The information contained in this communication from the sender is confidential. It is intended solely for use by the recipient and others authorized to receive it. If you are not the recipient, you are hereby notified that any disclosure, copying, distribution or taking action in relation of the contents of this information is strictly prohibited and may be unlawful.
This email has been scanned for viruses and malware, and may have been automatically archived by Mimecast Ltd, an innovator in Software as a Service (SaaS) for business. Providing a safer and more useful place for your human generated data. Specializing in; Security, archiving and compliance. To find out more visit the Mimecast website.
Re: [python SDK] Returning Pub/Sub message_id and timestamp
Posted by Valentyn Tymofieiev <va...@google.com>.
Hi Matthew,
Welcome to Beam!
Looking at Python PubSub IO API, you should be able to access id and
timestamp by setting `with_attributes=True` when using `ReadFromPubSub`
PTransform, see [1,2].
[1]
https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L61
[2]
https://github.com/apache/beam/blob/0fce2b88660f52dae638697e1472aa108c982ae6/sdks/python/apache_beam/io/gcp/pubsub.py#L138
On Fri, Jul 12, 2019 at 1:36 AM Matthew Darwin <
Matthew.Darwin@carfinance247.co.uk> wrote:
> Good morning,
>
> I'm very new to Beam, and pretty new to Python so please first accept my
> apologies for any obvious misconceptions/mistakes in the following.
>
> I am currently trying to develop a sample pipeline in Python to pull
> messages from Pub/Sub and then write them to either files in cloud storage
> or to BigQuery. The ultimate goal will be to utilise the pipeline for real
> time streaming of event data to BigQuery (with various transformations) but
> also to store the raw messages long term in files in cloud storage.
>
> At the moment, I'm simply trying to parse the message to get the PubSub
> messageId and publishTime in order to be able to write them into the
> output. The json of my PubSub message looks like this:-
>
> [
> {
> "ackId":
> "BCEhPjA-RVNEUAYWLF1GSFE3GQhoUQ5PXiM_NSAoRRIICBQFfH1xU1t1Xl8aB1ENGXJ8Zyc_XxcIB0BTeFVaEQx6bVxXOFcMEHF8YXZpWhUIA0FTfXeq5cveluzJNksxIbvE8KxfeqqmgfhiZho9XxJLLD5-PT5FQV5AEkw2C0RJUytDCypYEU4",
> "message": {
> "attributes": {
> "source": "python"
> },
> "data": "eyJyb3dudW1iZXIiOiAyfQ==",
> "messageId": "619310330691403",
> "publishTime": "2019-07-12T08:27:58.522Z"
> }
> }
> ]
> According to the documentation
> <https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html>
> the PubSub message payload returns the *data* and *attributes*
> properties; is there simply no way of retrieving the messageId and
> publishTime, or are these exposed somewhere else? If not, will the
> inclusion of these be in the roadmap, and are they available if using Java
> (I have zero Java experience hence why reaching for Python first).
>
> Kind regards,
>
> Matthew
>
>