谷歌云数据流-有可能定义一个从BigQuery读取数据并写入本地数据库的管道吗？

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
from apache_beam.io.gcp.bigquery import ReadFromBigQuery
import mysql.connector

class WriteToMySQL(beam.DoFn):
    def __init__(self, host, user, password, database):
        self.host = host
        self.user = user
        self.password = password
        self.database = database

    def start_bundle(self):
        self.conn = mysql.connector.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database
        )
        self.cursor = self.conn.cursor()

    def process(self, element):
        # 假设 BigQuery 表有 'id' 和 'name' 列
        query = "INSERT INTO your_table (id, name) VALUES (%s, %s)"
        self.cursor.execute(query, (element['id'], element['name']))
        self.conn.commit()

    def finish_bundle(self):
        self.cursor.close()
        self.conn.close()

def run():
    options = PipelineOptions()
    gcp_options = options.view_as(GoogleCloudOptions)
    gcp_options.project = 'YOUR_PROJECT_ID'
    gcp_options.region = 'YOUR_REGION'
    gcp_options.job_name = 'bigquery-to-mysql'
    gcp_options.staging_location = 'gs://YOUR_BUCKET_NAME/staging'
    gcp_options.temp_location = 'gs://YOUR_BUCKET_NAME/temp'
    options.view_as(PipelineOptions).runner = 'DataflowRunner'

    with beam.Pipeline(options=options) as p:
        (p
         | 'ReadFromBigQuery' >> ReadFromBigQuery(query='SELECT id, name FROM `YOUR_PROJECT_ID.YOUR_DATASET.YOUR_TABLE`')
         | 'WriteToMySQL' >> beam.ParDo(WriteToMySQL(host='localhost', user='your_user', password='your_password', database='your_database'))
        )

if __name__ == '__main__':
    run()

python dataflow_bigquery_to_mysql.py