使用谷歌DataFlow直接将数据流式传输到云SQL的简单查询

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
from apache_beam.io.gcp.pubsub import ReadFromPubSub
from apache_beam.io.jdbc import WriteToJdbc
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms.trigger import AfterWatermark, AfterProcessingTime, AccumulationMode

class ParseMessage(beam.DoFn):
    def process(self, element):
        # 解析Pub/Sub消息并进行任何必要的转换
        yield parsed_element

def run():
    options = PipelineOptions()
    gcp_options = options.view_as(GoogleCloudOptions)
    gcp_options.project = 'your-gcp-project-id'
    gcp_options.region = 'your-gcp-region'
    gcp_options.job_name = 'your-job-name'
    gcp_options.staging_location = 'gs://your-bucket/staging'
    gcp_options.temp_location = 'gs://your-bucket/temp'

    with beam.Pipeline(options=options) as p:
        (
            p
            | 'Read from PubSub' >> ReadFromPubSub(topic='projects/your-gcp-project-id/topics/your-topic')
            | 'Parse Message' >> beam.ParDo(ParseMessage())
            | 'Windowing' >> FixedWindows(size=60)  # 窗口大小为60秒
            | 'Write to Cloud SQL' >> WriteToJdbc(
                driver_class_name='com.mysql.jdbc.Driver',
                jdbc_url='jdbc:mysql://google/your-database?cloudSqlInstance=your-project-id:your-region:your-instance-name&socketFactory=com.google.cloud.sql.mysql.SocketFactory',
                username='your-username',
                password='your-password',
                statement='INSERT INTO your_table (column1, column2) VALUES (?, ?)',
                parameters=[beam.DoFn.Element(), beam.DoFn.Element()],
                write_batch_size=100,  # 每批写入的行数
                max_retries=5,
                retry_on_timeout=True
            )
        )

if __name__ == '__main__':
    run()

python your_pipeline_script.py