创建表
create table bas-ingress (
time bigint NOT NULL,
app varchar(200) NOT NULL,
code varchar(50) NULL,
instance varchar(200) NULL,
level varchar(50) NULL,
message varchar(1048000) NULL,
origin varchar(200) NULL,
segid varchar(200) NULL,
spid varchar(200) NULL,
tid varchar(300) NULL,
ctime varchar(200) NULL,
host_ip varchar(100) NULL,
host_name varchar(200) NULL,
log_file_path varchar(500) NULL,
contents JSON NULL,
tags JSON Null
) PRIMARY KEY (time, app)
DISTRIBUTED BY HASH(time,app)
ORDER BY(time)
PROPERTIES (
“replication_num” = “1”,
“enable_persistent_index” = “true”
);
创建routin load
CREATE ROUTINE LOAD load_kafka2basingress ON bas-ingress
COLUMNS(time, app, code, instance, level, message, origin, segid, spid, tid, contents, tags)
PROPERTIES
(
“desired_concurrent_number” = “3”,
“format” = “json”,
“jsonpaths” = “[”$.time","$.contents.app","$.contents.code","$.contents.instance","$.contents.level","$.contents.message","$.contents.origin","$.contents.segid","$.contents.spid","$.contents.tid","$.contents","$.tags"]"
)
FROM KAFKA
(
“kafka_broker_list” = “119.3.222.168:9092”,
“kafka_topic” = “bas-ingress”,
“kafka_partitions” = “0”,
“property.kafka_default_offsets” = “OFFSET_BEGINNING”,
“property.group.id” = “starrocks_bas_ingress_0”
);
kafka的json数据结构
{
“contents”: {
“app”: “basight”,
“code”: “?”,
“instance”: “bas-insight”,
“level”: "INFO ",
“message”: “fetch trace list controller entered!”,
“origin”: “http-nio-8082-exec-8”,
“segid”: “8cda13ad68ce46d2b65841a89a7f1600.65.17054007001061062”,
“spid”: “0”,
“tid”: “1359d87acb11435cab385862975a8a4c.54.17054007001024983”,
“time”: “2024-01-16 18:25:00.107”
},
“tags”: {
“host.ip”: “10.0.16.5”,
“host.name”: “VM-16-5-centos”,
“log.file.path”: “/home/itmonitor/docker/insight/app/logs/imonitor_api_sw.log”
},
“time”: 1705400700
}
be日志错误信息：
I0131 20:41:10.993237 1770573 plan_fragment_executor.cpp:176] Open(): fragment_instance_id=2d903d63-3742-4425-87a2-084a33a486f7
I0131 20:41:10.993273 1772857 data_consumer.cpp:181] start kafka consumer: e14fc80db33d2903-7c9e72c2b16e15b2, grp: 2e47e2de5ead1bfe-377a425164bef7a6, max running time(ms): 15000
I0131 20:41:11.227263 1770840 fragment_executor.cpp:161] Prepare(): query_id=035cbbee-c036-11ee-827b-1e04cbd100aa fragment_instance_id=035cbbee-c036-11ee-827b-1e04cbd100ac backend_num=0
I0131 20:41:11.228293 1770836 fragment_executor.cpp:161] Prepare(): query_id=035cbbee-c036-11ee-827b-1e04cbd100aa fragment_instance_id=035cbbee-c036-11ee-827b-1e04cbd100ab backend_num=1
I0131 20:41:11.229513 1772858 pipeline_driver_executor.cpp:225] [Driver] Fail to report exec state due to query not found: fragment_instance_id=035cbbee-c036-11ee-827b-1e04cbd100ac
I0131 20:41:11.229606 1770590 pipeline_driver_executor.cpp:225] [Driver] Fail to report exec state due to query not found: fragment_instance_id=035cbbee-c036-11ee-827b-1e04cbd100ab
W0131 20:41:11.269294 1770477 stack_util.cpp:350] 2024-01-31 20:41:11.269265, query_id=00000000-0000-0000-0000-000000000000, fragment_instance_id=00000000-0000-0000-0000-000000000000 throws exception: std::length_error, trace:
@ 0x2b7ccf1 std::__throw_length_error()
@ 0x813134c std::__cxx11::basic_string<>::_M_create()
@ 0x5121616 std::__uninitialized_copy<>::__uninit_copy<>()
@ 0x5122146 std::vector<>::_M_realloc_insert<>()
@ 0x5122419 starrocks::vectorized::ObjectColumn<>::append()
@ 0x41c81a5 starrocks::vectorized::add_native_json_column()
@ 0x41b69bb starrocks::vectorized::add_adpative_nullable_column()
@ 0x41b7fa3 starrocks::vectorized::add_adaptive_nullable_column()
@ 0x2f45b93 starrocks::vectorized::JsonReader::_construct_row_with_jsonpath()
@ 0x2f4af78 starrocks::vectorized::JsonReader::_read_rows<>()
@ 0x2f461a9 starrocks::vectorized::JsonReader::read_chunk()
@ 0x2f463e8 starrocks::vectorized::JsonScanner::get_next()
@ 0x515ae41 starrocks::connector::FileDataSource::get_next()
@ 0x304730d starrocks::vectorized::ConnectorScanNode::_scanner_thread()
@ 0x4a0f8e0 starrocks::PriorityThreadPool::work_thread()
@ 0x5bdf3e7 thread_proxy
@ 0x7ff1a601617a start_thread
@ 0x7ff1a55b7dc3 __GI___clone
@ (nil) (unknown)