当前位置:   article > 正文

基于Doris的日志存储分析平台(同步Kafka日志数据)_doris kafka

doris kafka
Kafka日志写入Doris
  • 实现方式
    • 采用 Routine Load 官方文档:Routine Load - Apache Doris
    • 采用 SparkStreaming实时消费Kafka后写入Doris,在日志记录场景(只写不删,不覆盖,不聚合...),我们使们前者;
    • Doris 可以通过 Routine Load 导入方式持续消费 Kafka Topic 中的数据。在提交 Routine Load 作业后,Doris 会持续运行该导入作业,实时生成导入任务不断消费 Kakfa 集群中指定 Topic 中的消息。
  • 我们的日志数据由log4net推入Kafka后是一串JSON格式,大致如下格式:
  1. [
  2. {
  3. "partition": 0,
  4. "offset": 49,
  5. "msg": "{\"log_timestamp\":1715752796717,\"business\":\"example.business.project\",\"service\":\"example.service\",\"host_ip\":\"YFB-CTO\",\"level\":\"INFO\",\"logger_name\":\"Web.Controllers.RoleController\",\"message\":\"########### \\u8FD9\\u662F\\u624B\\u52A8\\u8F93\\u51FA\\u7684\\u4E2D\\u6587\\u65E5\\u5FD7\\u4FE1\\u606F ###########\",\"request_path\":null,\"request_parameter\":null,\"request_method\":null,\"request_header\":null,\"status_code\":null,\"request_response_time\":0,\"exception\":null}",
  6. "timespan": 1715752796716,
  7. "date": "2024-05-15 13:59:56"
  8. }
  9. ]
  • 创建 Routine Load 导入作业
  • 语法
    1. CREATE ROUTINE LOAD [db.]job_name [ON tbl_name]
    2. [merge_type]
    3. [load_properties]
    4. [job_properties]
    5. FROM data_source [data_source_properties]
    6. [COMMENT "comment"]

  • 在 Doris 中,使用 CREATE ROUTINE LOAD 命令,创建导入作业
    1. CREATE ROUTINE LOAD biz_log_db.ods_example_service_log_routine_load_json ON ods_example_service_log
    2. COLUMNS(log_timestamp,business,service,host_ip,level,logger_name,message,request_path,request_parameter,request_method,request_header,response_content,status_code,request_response_time,exception,log_time=from_millisecond(log_timestamp))
    3. PROPERTIES(
    4. "desired_concurrent_number"="3",
    5. "max_batch_interval" = "20",
    6. "max_batch_rows" = "300000",
    7. "max_batch_size" = "209715200",
    8. "strict_mode" = "false",
    9. "format"="json",
    10. "jsonpaths"="[\"$.log_timestamp\",\"$.business\",\"$.service\",\"$.host_ip\",\"$.level\",\"$.logger_name\",\"$.message\",\"$.request_path\",\"$.request_parameter\",\"$.request_method\",\"$.request_header\",\"$.response_content\",\"$.status_code\",\"$.request_response_time\",\"$.exception\"]"
    11. )
    12. FROM KAFKA(
    13. "kafka_broker_list" = "192.168.2.111:9092,192.168.2.184:9092,192.168.2.156:9092",
    14. "kafka_topic" = "example-service-log",
    15. "property.kafka_default_offsets" = "OFFSET_BEGINNING",
    16. "property.deserializer.encoding" = "UTF-8"
    17. )
    18. COMMENT "示例服务日志记录同步 KAFKA ROUTINE LOAD 配置";

  • 说明
    • 请参考官方文档:CREATE-ROUTINE-LOAD - Apache Doris
    • 示例中通过 json_root 抽取根节点的元素进行解析,以简化配置。
    • OFFSET_BEGINNING: 表示从有数据的位置开始订阅。
    • 上述配置表示:自动默认消费所有分区,且从有数据的位置(OFFSET_BEGINNING)开始订阅
    • 查看导入运行任务
    • 查看导入运行作业
      • 创建 routine load 任务之后,可以通过show routine load命令查看运行状态的例行任务,如果在show routine load中没有找到对应的例行任务,则可能因为例行任务失败或者错误数过多被停止或者暂停,使用show all routine load查看所有状态的例行任务。
    • 暂停导入作业
    • 恢复(重启)导入作业
    • 修改导入作业
    • 取消(停止)导入作业
    • 问题排查
      • 先执行“SHOW ALL ROUTINE LOAD;”,找出有问题的ROUTINE-LOAD,拖到最右边,看到“ErrorLogUrls”,如下:
      • 复制ErrorLogUrls,在浏览器里打开,查看错误信息,如下:
      • 看一下数据
      • 最后:经过实测,已经取消的ROUTINE LOAD是没办法手动删除的,Doris会自动删除,不过这个过程等了好几天,过了一个周未才发现自动删除了已经取消的ROUTINE LOAD,如下:
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/木道寻08/article/detail/877543
推荐阅读
相关标签
  

闽ICP备14008679号