[Python] How to define nullable fields for SqlTransform

I'm using Beam SqlTransform in python, trying to define/pass nullable fields.

This code works just fine:

with beam.Pipeline(options=options) as p:
# ...

# Use beam.Row to create a schema-aware PCollection
| "Create beam Row" >> beam.Map(lambda x: beam.Row(
user_id=int(x['user_id']),
user_name=str(x['user_name])
))
| 'SQL' >> SqlTransform("SELECT user_id, COUNT(*) AS msg_count FROM PCOLLECTION GROUP BY user_id")


However, I am not able to create nullable fields with this approach.

Without the direct cast, I'm getting a decoding Field error.

user_id = json.get('user_id')


throws:

Failed to decode Schema due to an error decoding Field proto:

name: "user_id"
type {
nullable: true
logical_type {
urn: "beam:logicalythonsdk_any:v1"
}
}


Without using beam.Row, any other object, throws a missing schema error.

Cannot call getSchema when there is no schema


What is the proper way to define nullable fields?

Continue reading...