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<mailto:[email protected]> is the author of this code, might have an idea. On Wed, Jul 24, 2019 at 9:47 AM Matthew Darwin <[email protected]<mailto:[email protected]>> 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 <[email protected]<mailto:[email protected]>> 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 <[email protected]<mailto:[email protected]>> 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 <[email protected]<mailto:[email protected]>> 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 <[email protected]<mailto:[email protected]>> 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 <[email protected]<mailto:[email protected]>> 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 <[email protected]<mailto:[email protected]>> To: "[email protected]<mailto:[email protected]>" <[email protected]<mailto:[email protected]>> 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.
