基于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格式，大致如下格式：

[
  {
      "partition": 0,
      "offset": 49,
      "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}",
      "timespan": 1715752796716,
      "date": "2024-05-15 13:59:56"
    }
]

创建 routine load 导入作业
语法

 create routine load [db.]job_name [on tbl_name]
  [merge_type]
  [load_properties]
  [job_properties]
  from data_source [data_source_properties]
  [comment "comment"]

在 doris 中，使用 create routine load 命令，创建导入作业

create routine load biz_log_db.ods_example_service_log_routine_load_json on ods_example_service_log
    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))
    properties(
        "desired_concurrent_number"="3",
        "max_batch_interval" = "20",
        "max_batch_rows" = "300000",
        "max_batch_size" = "209715200",
        "strict_mode" = "false",
        "format"="json",
        "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\"]"
    )
    from kafka(
        "kafka_broker_list" = "192.168.2.111:9092,192.168.2.184:9092,192.168.2.156:9092",
        "kafka_topic" = "example-service-log",
        "property.kafka_default_offsets" = "offset_beginning",
        "property.deserializer.encoding" = "utf-8"
    )
    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查看所有状态的例行任务。
- 暂停导入作业
  - pause-routine-load - apache doris
- 恢复（重启）导入作业
  - resume-routine-load - apache doris
- 修改导入作业
  - alter-routine-load - apache doris
- 取消(停止)导入作业
  - stop-routine-load - apache doris
- 问题排查
  - 先执行“show all routine load;”，找出有问题的routine-load，拖到最右边，看到“errorlogurls”，如下：
  - 复制errorlogurls，在浏览器里打开，查看错误信息，如下：
  - 看一下数据
  - 最后：经过实测，已经取消的routine load是没办法手动删除的，doris会自动删除，不过这个过程等了好几天，过了一个周未才发现自动删除了已经取消的routine load，如下：