Flink读取StarRocks数据写入另外一个StarRocks

张

张建站

2026/4/18 12:10:09

10分钟阅读

-- 创建StarRocks-Sink CREATE TABLE IF NOT EXISTS source_df ( bike_no string NOT NULL, bik_show_no string NULL, belong_city_guid string NULL, belong_service_area_guid string NULL, business_type bigint NULL, bike_status bigint NULL, pro_code string NULL, pro_code_desc string NULL, asset_model string NULL, asset_model_name string NULL, tenant_id string NULL, pre_belong_city_guid string NULL, this_put_date string NULL, fst_put_date string NULL, is_kfk_zero_opt_useable_veh bigint NULL, is_opt_useable_veh bigint NULL, is_put_urban_veh bigint NULL, is_put_veh bigint NULL, is_ge1d_idle_veh bigint NULL, is_ge3d_idle_veh bigint NULL, is_ge7d_idle_veh bigint NULL, is_ge15d_idle_veh bigint NULL, is_ge30d_idle_veh bigint NULL, is_lt_idle_veh bigint NULL, is_lt_no_ride_veh bigint NULL, is_ext_site_veh bigint NULL, is_over_area_veh bigint NULL, is_zero_pwr_veh bigint NULL, is_low_btry_veh bigint NULL, is_btry_sep_veh bigint NULL, is_fut_miss_veh bigint NULL, is_miss_veh bigint NULL, is_pending_veh bigint NULL, is_mark_road_rep_veh bigint NULL, is_mark_rcl_veh bigint NULL, is_usr_rpt_flt_veh bigint NULL, is_unlock_fail_veh bigint NULL, is_susp_unlock_veh bigint NULL, is_hel_susp_unlock_veh bigint NULL, is_timeout_unlock_veh bigint NULL, is_hel_pwr_shtg_veh bigint NULL, is_flt_btry_veh_veh bigint NULL, is_sml_btry_low_pwr_veh bigint NULL, ge1d_idle_atime string NULL, ge3d_idle_atime string NULL, ge7d_idle_atime string NULL, ge15d_idle_atime string NULL, ge30d_idle_atime string NULL, lt_idle_atime string NULL, lt_no_ride_atime string NULL, ext_site_atime string NULL, over_area_atime string NULL, zero_pwr_atime string NULL, low_btry_atime string NULL, btry_sep_atime string NULL, fut_miss_atime string NULL, miss_atime string NULL, pending_atime string NULL, mark_road_rep_atime string NULL, mark_rcl_atime string NULL, usr_rpt_flt_atime string NULL, unlock_fail_atime string NULL, susp_unlock_atime string NULL, hel_susp_unlock_atime string NULL, timeout_unlock_atime string NULL, hel_pwr_shtg_atime string NULL, flt_btry_veh_atime string NULL, sml_btry_low_pwr_atime string NULL, scan_cnt bigint NULL, last_logtime string NULL, lack_power_scan_cnt bigint NULL, lack_power_last_logtime string NULL, pt date NULL, PRIMARY KEY (bike_no) NOT ENFORCED ) WITH ( connector starrocks, jdbc-url jdbc:mysql:loadbalance://192.168.0.1:9030, scan-url192.168.0.1:8030, database-name test, table-name status_df, username root, password 123456 ); -- 创建StarRocks-Sink CREATE TABLE IF NOT EXISTS sink_df ( bike_no string NOT NULL, bik_show_no string NULL, belong_city_guid string NULL, belong_service_area_guid string NULL, business_type bigint NULL, bike_status bigint NULL, pro_code string NULL, pro_code_desc string NULL, asset_model string NULL, asset_model_name string NULL, tenant_id string NULL, pre_belong_city_guid string NULL, this_put_date string NULL, fst_put_date string NULL, is_kfk_zero_opt_useable_veh bigint NULL, is_opt_useable_veh bigint NULL, is_put_urban_veh bigint NULL, is_put_veh bigint NULL, is_ge1d_idle_veh bigint NULL, is_ge3d_idle_veh bigint NULL, is_ge7d_idle_veh bigint NULL, is_ge15d_idle_veh bigint NULL, is_ge30d_idle_veh bigint NULL, is_lt_idle_veh bigint NULL, is_lt_no_ride_veh bigint NULL, is_ext_site_veh bigint NULL, is_over_area_veh bigint NULL, is_zero_pwr_veh bigint NULL, is_low_btry_veh bigint NULL, is_btry_sep_veh bigint NULL, is_fut_miss_veh bigint NULL, is_miss_veh bigint NULL, is_pending_veh bigint NULL, is_mark_road_rep_veh bigint NULL, is_mark_rcl_veh bigint NULL, is_usr_rpt_flt_veh bigint NULL, is_unlock_fail_veh bigint NULL, is_susp_unlock_veh bigint NULL, is_hel_susp_unlock_veh bigint NULL, is_timeout_unlock_veh bigint NULL, is_hel_pwr_shtg_veh bigint NULL, is_flt_btry_veh_veh bigint NULL, is_sml_btry_low_pwr_veh bigint NULL, ge1d_idle_atime string NULL, ge3d_idle_atime string NULL, ge7d_idle_atime string NULL, ge15d_idle_atime string NULL, ge30d_idle_atime string NULL, lt_idle_atime string NULL, lt_no_ride_atime string NULL, ext_site_atime string NULL, over_area_atime string NULL, zero_pwr_atime string NULL, low_btry_atime string NULL, btry_sep_atime string NULL, fut_miss_atime string NULL, miss_atime string NULL, pending_atime string NULL, mark_road_rep_atime string NULL, mark_rcl_atime string NULL, usr_rpt_flt_atime string NULL, unlock_fail_atime string NULL, susp_unlock_atime string NULL, hel_susp_unlock_atime string NULL, timeout_unlock_atime string NULL, hel_pwr_shtg_atime string NULL, flt_btry_veh_atime string NULL, sml_btry_low_pwr_atime string NULL, scan_cnt bigint NULL, last_logtime string NULL, lack_power_scan_cnt bigint NULL, lack_power_last_logtime string NULL, pt date NULL, PRIMARY KEY (bike_no) NOT ENFORCED ) WITH ( connector starrocks, jdbc-url jdbc:mysql:loadbalance://192.168.0.2:9030, load-url192.168.0.2:8030, database-name test, table-name status_df, username root, password 123456 ); insert into default_catalog.default_database.sink_df select bike_no, bik_show_no, belong_city_guid, belong_service_area_guid, business_type, bike_status, pro_code, pro_code_desc, asset_model, asset_model_name, tenant_id, pre_belong_city_guid, this_put_date, fst_put_date, is_kfk_zero_opt_useable_veh, is_opt_useable_veh, is_put_urban_veh, is_put_veh, is_ge1d_idle_veh, is_ge3d_idle_veh, is_ge7d_idle_veh, is_ge15d_idle_veh, is_ge30d_idle_veh, is_lt_idle_veh, is_lt_no_ride_veh, is_ext_site_veh, is_over_area_veh, is_zero_pwr_veh, is_low_btry_veh, is_btry_sep_veh, is_fut_miss_veh, is_miss_veh, is_pending_veh, is_mark_road_rep_veh, is_mark_rcl_veh, is_usr_rpt_flt_veh, is_unlock_fail_veh, is_susp_unlock_veh, is_hel_susp_unlock_veh, is_timeout_unlock_veh, is_hel_pwr_shtg_veh, is_flt_btry_veh_veh, is_sml_btry_low_pwr_veh, ge1d_idle_atime, ge3d_idle_atime, ge7d_idle_atime, ge15d_idle_atime, ge30d_idle_atime, lt_idle_atime, lt_no_ride_atime, ext_site_atime, over_area_atime, zero_pwr_atime, low_btry_atime, btry_sep_atime, fut_miss_atime, miss_atime, pending_atime, mark_road_rep_atime, mark_rcl_atime, usr_rpt_flt_atime, unlock_fail_atime, susp_unlock_atime, hel_susp_unlock_atime, timeout_unlock_atime, hel_pwr_shtg_atime, flt_btry_veh_atime, sml_btry_low_pwr_atime, scan_cnt, last_logtime, lack_power_scan_cnt, lack_power_last_logtime, pt from default_catalog.default_database.source_df where pt TO_DATE(20251110, yyyyMMdd) ;