Wanted to dynamically name and create tables by filtering events in a pipeline using Apache Beam?

I had a use-case where I was listening from pub/sub in an event driven architecture and wanted to store and insert data into tables dynamically. I wanted to make new table if a new eventName is noticed in the stream, for example if I have a streaming data of user creation with a name in attributes (which is a python dict), as CREATED, so what should happen is this should automatically create a table in that specific dataset which is used for the use case and start writing data into that newly made table of that specific table. I've tried the following thing and it hasn't worked for me, please help:

class FilterEvents(beam.DoFn):
    def process(self, element):
        events = []
        event_name = element['event_name']
        for i in event_name:
            res.extend(list(i.values()))
        res=list(set(res))   
        if event_name in events:
            yield element

filtered_events = events | beam.ParDo(FilterEvents())

# Step 4: Use the Partition transform to split the filtered events into multiple PCollections based on the event name.
def partition_fn(element):
    event_name = element['event_name']
    if event_name == 'event1':
        return 0
    elif event_name == 'event2':
        return 1
    elif event_name == 'event3':
        return 2

partitioned_events = filtered_events | beam.Partition(partition_fn, 3)

# Step 5: Write the events in each partition to a separate table.
def write_to_table(elements, table_name):
    elements | beam.GroupByKey() | beam.io.WriteToBigQuery(
        table=table_name,
        # schema=<table_schema>,
        write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)

partitioned_events[0] | beam.Map(lambda element: (element['event_name'], element)) | beam.WindowInto(window.FixedWindows(60)) | beam.ParDo(lambda x: write_to_table(x[1], "event1_table"))
partitioned_events[1] | beam.Map(lambda element: (element['event_name'], element)) | beam.WindowInto(window.FixedWindows(60)) | beam.ParDo(lambda x: write_to_table(x[1], "event2_table"))
partitioned_events[2] | beam.Map(lambda element: (element['event_name'], element)) | beam.WindowInto(window.FixedWindows(60)) | beam.ParDo(lambda x: write_to_table(x[1], "event3_table"))

pipeline.run()

This can be done using the standard WriteToBigQuery sink of Apache beam (see docs).

In particular, you can provide a custom function in the table argument, which has as an argument the element you want to write to BigQuery and returns a string representing the table name. So you can skip the whole partition and just do something like this:

filtered_events
| beam.io.gcp.bigquery.WriteToBigQuery(
 table=lambda element: f"{element['event_name']}_table",  # <- replace here with custom logic
 write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
 create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
)

where you may replace the simple lambda with a more sophisticated custom method, which suits your case. It is even possible to provide a side input which you can access in this method as well (see here).