????????????????????????????????flink??????????????g_unit(????????)??????????kafka??????????????g_unit??????????????g_summary??????????????????g_line????????????????????????????????????????????
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic, CheckpointingMode from pyflink.table import StreamTableEnvironment, EnvironmentSettings env = StreamExecutionEnvironment.get_execution_environment() env.set_stream_time_characteristic(TimeCharacteristic.EventTime) env.set_parallelism(1) env_settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build() t_env = StreamTableEnvironment.create(env, environment_settings=env_settings) kafka_source_ddl = """ CREATE TABLE kafka_source_tab ( id VARCHAR, alarm_id VARCHAR, trck_id VARCHAR ) WITH ( 'connector' = 'kafka', 'topic' = 'gg', 'scan.startup.mode' = 'specific-offsets', 'scan.startup.specific-offsets'='partition:1,offset:0', 'properties.bootstrap.servers' = '****', 'format' = 'json' ) """ g_unit_sink_ddl = """ CREATE TABLE g_sink_unit ( alarm_id VARCHAR, trck_id VARCHAR ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.2.2.70:3306/bdoa?useSSL=false', 'table-name' = 'g_unit', 'username' = 'root', 'password' = 'root', 'sink.buffer-flush.interval' = '1s' ) """ g_summary_ddl = """ CREATE TABLE g_summary_base( alarm_id VARCHAR, trck_id VARCHAR ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.2.2.70:3306/bdoa?useSSL=false', 'table-name' = 'g_summary', 'username' = 'root', 'password' = 'root', 'sink.buffer-flush.interval' = '1s' ) """ t_env.execute_sql(kafka_source_ddl) t_env.execute_sql(g_unit_sink_ddl) t_env.execute_sql(g_summary_ddl) sql1='''Insert into g_unit_sink_ddl select alarm_id,trck_id from kafka_source_tab''' sql2='''Insert into g_summary_ddl select alarm_id,trck_id from g_unit_sink_ddl''' stmt_set = t_env.create_statement_set() stmt_set.add_insert_sql(sql1) stmt_set.add_insert_sql(sql2) stmt_set.execute().get_job_client().get_job_execution_result().result()
