It seems like the main issue is caused by not converting the decoded string to a dictionary befor...

It seems like the main issue is caused by not converting the decoded string to a dictionary before passing it to MySchema. Since you mentioned that the output is in JSON format, you can use the json.loads() function to convert the decoded string to a dictionary.

Here's the modified code with the necessary changes:

⎗
✓
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
import json
import typing

class MySchema(typing.NamedTuple):
    # your schema definition

TOPIC_PATH = "projects/nectar-259905/topics/events"

def run(pubsub_topic):
    options = PipelineOptions(
        streaming=True
    )
    runner = 'DirectRunner'

    print("I reached before pipeline")

    with beam.Pipeline(runner, options=options) as pipeline:
        message=(
            pipeline
            | "Read from Pub/Sub topic" >> beam.io.ReadFromPubSub(subscription='projects/triple-nectar-259905/subscriptions/bq_subscribe')
            | 'UTF-8 bytes to string' >> beam.Map(lambda msg: msg.decode('utf-8'))
            | 'String to JSON' >> beam.Map(lambda msg: json.loads(msg))
            | 'Map to MySchema' >> beam.Map(lambda msg: MySchema(**msg)).with_output_types(MySchema)
            | "Writing to console" >> beam.Map(print))

        print("I reached after pipeline")
        result = message.run()
        result.wait_until_finish()

run(TOPIC_PATH)

In this code, I've added the 'String to JSON' step to convert the decoded string to a dictionary using json.loads(). This should fix the TypeError you encountered.

Regarding data masking and running SQL within Apache Beam, you can use the beam.Map() and beam.SqlTransform() functions. Here's an example of how to mask the user_id field and run a simple SQL query to filter records with a specific event_id:

⎗
✓
def mask_user_id(record: MySchema) -> MySchema:
    masked_user_id = "MASKED-" + record.user_id[-4:]
    return record._replace(user_id=masked_user_id)

with beam.Pipeline(runner, options=options) as pipeline:
    message=(
        pipeline
        # ... previous steps ...
        | 'Map to MySchema' >> beam.Map(lambda msg: MySchema(**msg)).with_output_types(MySchema)
        | 'Mask user_id' >> beam.Map(mask_user_id)
        | 'Filter specific event_id' >> beam.SqlTransform("""
            SELECT * FROM PCOLLECTION
            WHERE event_id = 'Game_Request_Declined'
        """)
        | "Writing to console" >> beam.Map(print))

In this example, I've added a mask_user_id function to mask the user_id field, a Mask user_id step to apply the data masking, and a Filter specific event_id step to filter records using SQL.