CREATE TABLE IF NOT EXISTS table_1(key1 varchar(32),key2 varchar(32),key3 varchar(32),value1 int,value2 varchar(128),value3 Decimal(20, 6),data_deal_datetime DateTime COMMENT '数据处理时间',data_status INT COMMENT '数据是否删除，1表示正常，-1表示数据已经删除') ENGINE=OLAPUNIQUE KEY(`key1`,`key2`,`key3`)COMMENT "xxx"DISTRIBUTED BY HASH(`key2`) BUCKETS 32PROPERTIES ("storage_type"="column","replication_num" = "3","function_column.sequence_type" = 'DateTime');

CREATE ROUTINE LOAD database.table1 ON table1COLUMNS(key1,key2,key3,value1,value2,value3,data_deal_datetime,data_status),ORDER BY data_deal_datetimePROPERTIES("desired_concurrent_number"="3","max_batch_interval" = "10","max_batch_rows" = "500000","max_batch_size" = "209715200","format" = "json","json_root" = "$.data","jsonpaths"="[\"$.key1\",\"$.key2\",\"$.key3\",\"$.value1\",\"$.value2\",            \"$.value3\",\"$.data_deal_datetime\",\"$.data_status\"]")FROM KAFKA("kafka_broker_list"="broker1_ip:port1,broker2_ip:port2,broker3_ip:port3","kafka_topic"="topic_name","property.group.id"="group_id","property.kafka_default_offsets"="OFFSET_BEGINNING");

import pymysql  #导入 pymysqlimport requests,json

#打开数据库连接db= pymysql.connect(host="host",user="user",                    password="passwd",db="database",port=port)
# 使用cursor()方法获取操作游标cur = db.cursor()
#1.查询操作# 编写sql 查询语句 sql = "show routine load"cur.execute(sql)        #执行sql语句results = cur.fetchall()        #获取查询的所有记录for row in results :    name = row[1]    state = row[7]    if state != 'RUNNING':        err_log_urls = row[16]        reason_state_changed = row[15]        msg = "doris 数据导入任务异常:\n name=%s \n state=%s \n reason_state_changed=%s \n err_log_urls=%s \n即将自动恢复，请检查错误信息" % (name, state,reason_state_changed, err_log_urls)        payload_message = {    "msg_type": "text",    "content": {        "text": msg    }}        url = 'lark 报警url'        s = json.dumps(payload_message)        r = requests.post(url, data=s)        cur.execute("resume routine load for " + name)
cur.close()db.close()

CREATE TABLE tmp_table_1    (        user_id varchar(64) COMMENT "用户id",        channel varchar(64) COMMENT "用户来源渠道",        city_code varchar(64) COMMENT "用户所在城市编码",        last_visit_date DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "用户最后一次访问时间",        total_cost BIGINT SUM DEFAULT "0" COMMENT "用户总消费"    )ENGINE=OLAPAGGREGATE KEY(user_id, channel, city_code)DISTRIBUTED BY HASH(user_id) BUCKETS 6    PROPERTIES("storage_type"="column","replication_num" = "1"):

insert into tmp_table_1 values('suh_001','JD','001','2022-01-01 00:00:01','57');insert into tmp_table_1 values('suh_001','JD','001','2022-02-01 00:00:01','76');insert into tmp_table_1 values('suh_001','JD','001','2022-03-01 00:00:01','107');

CREATE TABLE tmp_table_2    (        user_id varchar(64) COMMENT "用户id",        channel varchar(64) COMMENT "用户来源渠道",        city_code varchar(64) COMMENT "用户所在城市编码",        visit_date DATETIME COMMENT "用户登陆时间",cost BIGINT COMMENT "用户消费金额"    )ENGINE=OLAPDUPLICATE KEY(user_id, channel, city_code)DISTRIBUTED BY HASH(user_id) BUCKETS 6    PROPERTIES("storage_type"="column","replication_num" = "1");

insert into tmp_table_2 values('suh_001','JD','001','2022-01-01 00:00:01','57');insert into tmp_table_2 values('suh_001','JD','001','2022-02-01 00:00:01','76');insert into tmp_table_2 values('suh_001','JD','001','2022-03-01 00:00:01','107');