中国民生银行大数据团队的Flume实践

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 转载自:AI前线 中国民生银行服务器的操作系统种类众多,除 Linux 外,部分生产系统仍采用 AIX 和 HP-UNIX 操作系统,由于在 AIX 和 HP-UNIX 无法使用 Logstash 作为日志采集端,在大数据基础平台产品团队经过一系列选型后,采用 Flume 作为 AIX 和 HP-UNIX 操作系统上日志采集端。

转载自:AI前线


一. Flume 简介

Apache Flume 是 Cloudera 公司开源的一款分布式、可靠、可用的服务,可用于从多种不同数据源收集、聚集、移动大量日志数据到集中数据存储中;它通过事务机制提供了可靠的消息传输支持,并自带负载均衡机制来支撑水平扩展。尤其近几年随着 Flume 的不断被完善以及升级版本的逐一推出,特别是 flume-ng 的推出,以及 Flume 内部的各种组件不断丰富,用户在开发的过程中使用的便利性得到很大的改善,现已成为 Apache 顶级社区项目之一。


二. 中国民生银行 Flume 实践

中国民生银行服务器的操作系统种类众多,除 Linux 外,部分生产系统仍采用 AIX 和 HP-UNIX 操作系统,由于在 AIX 和 HP-UNIX 无法使用 Logstash 作为日志采集端,在大数据基础平台产品团队经过一系列选型后,采用 Flume 作为 AIX 和 HP-UNIX 操作系统上日志采集端。

2016 年我们在测试环境进行试验,使用的版本是 Apache Flume 1.6,在使用 Taildir Source 组件和核心组件的过程中,发现其无法完全满足我们的需求,例如:

  1. 若 filegroup 路径中包含正则表达式,则无法获取文件的完整路径,在日志入到 Elasticsearch 后无法定位日志的路径;
  2. Taildir Source 不支持将多行合并为一个 event,只能一行一行读取文件;
  3. filegroup 配置中不支持目录包含正则表达式,不便配置包含多个日期并且日期自动增长的目录,例如 /app/logs/yyyymmdd/appLog.log;
  4. 在使用 Host Interceptor 时,发现只能保留主机名或者是 IP,二者无法同时保留。

在研究 Flume 源码之后,我们在源码上扩展开发。截至目前,我们为开源社区贡献了 4 个 Patch,其中 FLUME-2955 已被社区 Merge 并在 1.7 版本中发布,另外我们在 Github 上开放了一个版本,将 FLUME-2960/2961/3187 三个 Patch 合并到 Flume 1.7 上,欢迎大家下载使用,

Github 地址:

github.com/tinawenqiao/,分支名 trunk-cmbc。

接下来本文将对每个 Issue 进行详细介绍:


三. FLUME-2955

3.1 问题和需求

为了采集后缀为 log 的日志文件,filegroups 设置如下:

agent.sources.s1.type = org.apache.flume.source.taildir.TaildirSource
agent.sources.s1.filegroups = f1 
agent.sources.s1.filegroups.f1 = /app/logs/.*.log

注:安卓手机端读者查看代码时可左右滑动阅读完整代码

若 /app/logs 目录中存在 a.log、b.log、c.log 三个文件,在 Flume 1.6 版本中,虽然可以通过 headers.\.\在 event 的 header 里放入自定义的 key 和 value,但是由于正则表达式匹配上了目录中多个文件,所以无法通过该方法设置,这样导致日志数据入到 Elasticsearch 后,用户从 Kibana 从查询时无法定位到数据所在的日志文件路径。

3.2 解决办法

增加 fileHeader 和 fileHeaderKey 两个参数,两个参数含义分别是:

修改类 ReliableTaildirEventReader 中 readEvents() 方法,根据配置文件的值,选择是否在 event 的 header 里加入文件的路径,主要代码如下:

Map<String, String> headers = currentFile.getHeaders();
if (annotateFileName || (headers != null && !headers.isEmpty())) {
  for (Event event : events) {
    if (headers != null && !headers.isEmpty()) {
      event.getHeaders().putAll(headers);
    }
    if (annotateFileName) {
      event.getHeaders().put(fileNameHeader, currentFile.getPath());
    }
  }
}

3.3 相关配置示例

agent.sources.s1.type = org.apache.flume.source.taildir.TaildirSource
agent.sources.s1.filegroups = f1 
agent.sources.s1.filegroups.f1 = /app/logs/.*.log
agent.sources.s1.fileHeader = true
agent.sources.s1.fileHeaderKey = path


四. FLUME-2960

4.1 问题和需求

在实际应用写日志时,很多系统是根据日期生成日期目录,每个日期目录中包含一个或多个日志文件,因此存在:

/app/logs/20170101/、/app/logs/20170102/、/app/logs/20170103/

等多个目录,且 /app/logs/ 目录下每天会自动生成新的日期目录,但是根据 Taildir Source 中 filegroups.\的描述,只支持文件名带正则,因此 1.6 版本的 Taildir Source 无法满足该需求。

4.2 解决办法

增加 filegroups.\.parentDir 和 filegroups.\.filePattern 两个参数,两个参数含义分别是:

修改类 TaildirMatcher 中匹配文件的方法,相关代码如下:

private List<File> getMatchingFilesNoCache() {
  final List<File> result = Lists.newArrayList();
  try {
    Set options = EnumSet.of(FOLLOW_LINKS);
    Files.walkFileTree(Paths.get(parentDir.toString()), options, Integer.MAX_VALUE,
            new SimpleFileVisitor<Path>() {
        @Override
        public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
          if (fileMatcher.matches(file.toAbsolutePath())) {
            result.add(file.toFile());
          }
          return FileVisitResult.CONTINUE;
        }
        @Override
        public FileVisitResult visitFileFailed(Path file, IOException exc) {
          return FileVisitResult.CONTINUE;
        }
        });
  } 
  ...
}

另外进行了配置参数的兼容性处理,用户仍可保留以前的 filegroups 配置,不需单独配置 parentDir 和 filePattern,程序会将 filegroups 中的文件的目录赋值给 parentDir,文件名赋值给 filePattern。

需要注意的是:在 Taildir Source 中有个参数 cachePatternMatching,默认值是 true,其作用是缓存正则匹配的文件列表和消费文件的顺序,若目录中文件较多时,使用正则匹配比较耗时,设置该参数可提高性能,当发现文件的目录修改后会刷新缓存列表。由于 filePattern 中可包含目录,若 cachePatternMatching 设为 true,在 filePattern 的子目录中新增文件,parentDir 的修改时间不变,此时新增的日志文件不能被跟踪到,因此,建议在 filePattern 包含目录的情况下,将 cachePatternMatching 设置为 false

4.3 相关配置示例

agent.sources.s2.type = org.apache.flume.source.taildir.TaildirSource
agent.sources.s2.filegroups = f1 f2
agent.sources.s2.filegroups.f1.parentDir = /app/log/
agent.sources.s2.filegroups.f1.filePattern = /APP.log.\\d{8}
agent.sources.s2.filegroups.f2.parentDir = /app/log/
agent.sources.s2.filegroups.f2.filePattern = /\\w/.*log
agent.sources.s2.cachePatternMatching = false


五. FLUME-2961

5.1 问题和需求

Taildir Source 按行读取日志,把每一行作为内容放入 flume event 的 body 中,对于以下这种每行就可以结束的日志处理没有问题:

13 七月 2016 23:37:30,580 INFO  [lifecycleSupervisor-1-0] (org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start:62)  - Configuration provider starting
13 七月 2016 23:37:30,585 INFO  [conf-file-poller-0] (org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run:134)  - Reloading configuration file:conf/taildir.conf
13 七月 2016 23:37:30,592 INFO  [conf-file-poller-0] (org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty:1013)  - Processing:s1

但是对于类似 Java Stacktrace 的日志,如果按上述处理,以下日志被截断成 9 个 flume event(一共 9 行)输出,而我们希望这样的日志记录,要作为 1 个 flume event,而不是 9 个输出:

13 七月 2016 23:37:41,942 ERROR [SinkRunner-PollingRunner-DefaultSinkProcessor] (org.apache.flume.sink.kafka.KafkaSink.process:229)  - Failed to publish events
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.RecordTooLargeException: The message is 2000067 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
    at org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:686)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:449)
    at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:200)
    at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67)
    at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 2000067 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.

5.2 解决办法

设计一个 buffer event 缓存多行内容,仿照 Logstash 的 codec/mulitline 插件配置,增加了如下参数:

主要修改了类 TailFile 里的 readEvents() 方法,相关代码如下:

if (this.multiline) {
  if (raf != null) { // when file has not closed yet
    boolean match = this.multilinePatternMatched;
    while (events.size() < numEvents) {
      LineResult line = readLine();
      if (line == null) {
        break;
      }
      Event event = null;
      logger.debug("TailFile.readEvents: Current line = " + new String(line.line) +
               ". Current time : " + new Timestamp(System.currentTimeMillis()) +
               ". Pos:" + pos +
               ". LineReadPos:" + lineReadPos + ",raf.getPointer:" + raf.getFilePointer());
      switch (this.multilinePatternBelong) {
        case "next":
          event = readMultilineEventNext(line, match);
          break;
        case "previous":
          event = readMultilineEventPre(line, match);
          break;
        default:
          break;
      }
      if (event != null) {
        events.add(event);
      }
      if (bufferEvent != null) {
        if (bufferEvent.getBody().length >= multilineMaxBytes
                || Integer.parseInt(bufferEvent.getHeaders().get("lineCount")) == multilineMaxLines) {
          flushBufferEvent(events);
        }
      }
    }
  }
  if (needFlushTimeoutEvent()) {
    flushBufferEvent(events);
  }
}

合并多行处理的方法代码如下:

private Event readMultilineEventPre(LineResult line, boolean match)
          throws IOException {
  Event event = null;
  Matcher m = multilinePattern.matcher(new String(line.line));
  boolean find = m.find();
  match = (find && match) || (!find && !match);
  byte[] lineBytes = toOriginBytes(line);
  if (match) {
    /** If matched, merge it to the buffer event. */
    mergeEvent(line);
  } else {
    /**
     * If not matched, this line is not part of previous event when the buffer event is not null.
     * Then create a new event with buffer event's message and put the current line into the
     * cleared buffer event.
     */
    if (bufferEvent != null) {
      event = EventBuilder.withBody(bufferEvent.getBody());
    }
    bufferEvent = null;
    bufferEvent = EventBuilder.withBody(lineBytes);
    if (line.lineSepInclude) {
      bufferEvent.getHeaders().put("lineCount", "1");
    } else {
      bufferEvent.getHeaders().put("lineCount", "0");
    }
    long now = System.currentTimeMillis();
    bufferEvent.getHeaders().put("time", Long.toString(now));
  }
  return event;
}

private Event readMultilineEventNext(LineResult line, boolean match)
        throws IOException {
  Event event = null;
  Matcher m = multilinePattern.matcher(new String(line.line));
  boolean find = m.find();
  match = (find && match) || (!find && !match);
  if (match) {
    /** If matched, merge it to the buffer event. */
    mergeEvent(line);
  } else {
    /**
     * If not matched, this line is not part of next event. Then merge the current line into the
     * buffer event and create a new event with the merged message.
     */
    mergeEvent(line);
    event = EventBuilder.withBody(bufferEvent.getBody());
    bufferEvent = null;
  }
  return event;
}

3.3 相关配置示例

agent.sources.s3.multiline = true
agent.sources.s3.multilinePattern = ^AGENT_IP:
agent.sources.s3.multilinePatternBelong = previous
agent.sources.s3.multilineMatched = false
agent.sources.s3.multilineEventTimeoutSeconds = 120
agent.sources.s3.multilineMaxBytes = 3145728
agent.sources.s3.multilineMaxLines = 3000


六. FLUME-3187

6.1 问题和需求

为了获取 Flume agent 所在机器的主机名或 IP,我们使用了主机名拦截器 (Host Interceptor),但是根据主机名拦截器的定义,只能保留主机名和 IP 中的一种,无法同时保留主机名和 IP。

Host Interceptor
This interceptor inserts the hostname or IP address of the host that this agent is running on. It inserts a header with key host or a configured key whose value is the hostname or IP address of the host, based on configuration.

6.2 解决办法

将原来的 useIP 参数扩展,增加一个参数 useHostname,若同时设置为 true,可同时保留主机名和 IP;另外支持自定义主机名和 IP 地址在 event header 里的 key,参数如下:

修改了类 HostInterceptor 中的构造方法和拦截方法,相关代码如下:

addr = InetAddress.getLocalHost();
if (useIP) {
  ip = addr.getHostAddress();
}
if (useHostname) {
  hostname = addr.getCanonicalHostName();
}

6.3 相关配置示例

agent.sources.s4.interceptors = i1
agent.sources.s4.interceptors.i1.type = host
agent.sources.s4.interceptors.i1.useIP = true
agent.sources.s4.interceptors.i1.useHostname = true
agent.sources.s4.interceptors.i1.ip = ip
agent.sources.s4.interceptors.i1.hostname = hostname


总结

目前上述 4 个 Patch 在我行 A 类和 B 类生产系统已实际运行使用,“拥抱开源,回馈开源”,我们用的是开源软件,我们希望也能对开源软件做出贡献。后续我们将分享我行 ELK 日志平台架构演进的详细细节,敬请大家关注!


相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
24天前
|
资源调度 安全 Java
Java 大数据在智能教育在线实验室设备管理与实验资源优化配置中的应用实践
本文探讨Java大数据技术在智能教育在线实验室设备管理与资源优化中的应用。通过统一接入异构设备、构建四层实时处理管道及安全防护双体系,显著提升设备利用率与实验效率。某“双一流”高校实践显示,设备利用率从41%升至89%,等待时间缩短78%。该方案降低管理成本,为教育数字化转型提供技术支持。
49 0
|
机器学习/深度学习 数据采集 算法
Java 大视界 -- Java 大数据机器学习模型在金融衍生品定价中的创新方法与实践(166)
本文围绕 Java 大数据机器学习模型在金融衍生品定价中的应用展开,分析定价现状与挑战,阐述技术原理与应用,结合真实案例与代码给出实操方案,助力提升金融衍生品定价的准确性与效率。
Java 大视界 -- Java 大数据机器学习模型在金融衍生品定价中的创新方法与实践(166)
|
2月前
|
Cloud Native 大数据 Java
大数据新视界--大数据大厂之大数据时代的璀璨导航星:Eureka 原理与实践深度探秘
本文深入剖析 Eureka 在大数据时代分布式系统中的关键作用。涵盖其原理,包括服务注册、续约、发现及自我保护机制;详述搭建步骤、两面性;展示在大数据等多领域的应用场景、实战案例及代码演示。Eureka 如璀璨导航星,为分布式系统高效协作指引方向。
|
3月前
|
存储 SQL 运维
中国联通网络资源湖仓一体应用实践
本文分享了中国联通技术专家李晓昱在Flink Forward Asia 2024上的演讲,介绍如何借助Flink+Paimon湖仓一体架构解决传统数仓处理百亿级数据的瓶颈。内容涵盖网络资源中心概况、现有挑战、新架构设计及实施效果。新方案实现了数据一致性100%,同步延迟从3小时降至3分钟,存储成本降低50%,为通信行业提供了高效的数据管理范例。未来将深化流式数仓与智能运维融合,推动数字化升级。
147 0
中国联通网络资源湖仓一体应用实践
|
4月前
|
存储 安全 数据挖掘
天翼云:Apache Doris + Iceberg 超大规模湖仓一体实践
天翼云基于 Apache Doris 成功落地项目已超 20 个,整体集群规模超 50 套,部署节点超 3000 个,存储容量超 15PB
天翼云:Apache Doris + Iceberg 超大规模湖仓一体实践
|
4月前
|
SQL 存储 消息中间件
vivo基于Paimon的湖仓一体落地实践
本文整理自vivo互联网大数据专家徐昱在Flink Forward Asia 2024的分享,基于实际案例探讨了构建现代化数据湖仓的关键决策和技术实践。内容涵盖组件选型、架构设计、离线加速、流批链路统一、消息组件替代、样本拼接、查询提速、元数据监控、数据迁移及未来展望等方面。通过这些探索,展示了如何优化性能、降低成本并提升数据处理效率,为相关领域提供了宝贵的经验和参考。
656 3
vivo基于Paimon的湖仓一体落地实践
|
4月前
|
数据采集 机器学习/深度学习 数据可视化
探索大数据分析的无限可能:R语言的应用与实践
探索大数据分析的无限可能:R语言的应用与实践
192 9
|
4月前
|
SQL 分布式计算 运维
StarRocks 在爱奇艺大数据场景的实践
本文介绍了爱奇艺大数据OLAP服务负责人林豪在StarRocks年度峰会上的分享,重点讲述了爱奇艺OLAP引擎的演进及引入StarRocks后的显著效果。在广告业务中,StarRocks替换Impala+Kudu后,接口性能提升400%,P90查询延迟缩短4.6倍;在“魔镜”数据分析平台中,StarRocks替代Spark达67%,P50查询速度提升33倍,P90提升15倍,节省4.6个人天。未来,爱奇艺计划进一步优化存算一体和存算分离架构,提升整体数据处理效率。
StarRocks 在爱奇艺大数据场景的实践
|
4月前
|
SQL 分布式计算 数据挖掘
从湖仓分离到湖仓一体,四川航空基于 SelectDB 的多源数据联邦分析实践
川航选择引入 SelectDB 建设湖仓一体大数据分析引擎,取得了数据导入效率提升 3-6 倍,查询分析性能提升 10-18 倍、实时性提升至 5 秒内等收益。
从湖仓分离到湖仓一体,四川航空基于 SelectDB 的多源数据联邦分析实践
|
4月前
|
运维 自然语言处理 算法
云栖实录 | 大模型在大数据智能运维的应用实践
云栖实录 | 大模型在大数据智能运维的应用实践
497 3