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
<[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
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.
--- Begin Message ---
====================================================
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
Filename: ApacheBeamDataflow-To-BigQuery2.py
Size: 6559 bytes
–––––––––––––––––––––––––––––––––––––––––––––––––––
Powered by Mimecast
© 2003 - 2019 Mimecast Services Limited.
--- End Message ---
