EMR Flume支持多种服务启动方式,本文介绍通过E-MapReduce(简称EMR)控制台修改Flume配置并启动Flume Agent,继而实时同步HDFS Audit日志至HDFS。

前提条件

已创建DataLake集群,并且选择了Flume服务,详情请参见创建集群

操作步骤

  1. 进入集群服务页面。
    1. 登录EMR on ECS控制台
    2. 在顶部菜单栏处,根据实际情况选择地域和资源组
    3. 集群管理页面,单击目标集群操作列的集群服务
  2. 集群服务页面,单击FLUME服务区域的配置
  3. 配置core-1-1节点的Flume Agent并保存配置。
    以下配置项请遵循开源Flume内容,详情请参见Flume
    1. 配置页面,单击flume-conf.properties页签。
    2. 在上方的下拉列表中选择独立节点配置core-1-1
    3. 请根据实际情况修改参数flume-conf.properties中的值。
      default-agent.sinks = default-sink
      default-agent.sources = default-source
      default-agent.channels = default-channel
      default-agent.sinks.default-sink.type = hdfs
      default-agent.sinks.default-sink.channel =  default-channel
      default-agent.channels.default-channel.type = file
      default-agent.sources.default-source.type = avro
      default-agent.sinks.default-sink.hdfs.path = hdfs://master-1-1:9000/path
      default-agent.sinks.default-sink.hdfs.fileType = DataStream
      default-agent.sinks.default-sink.hdfs.rollSize = 0
      default-agent.sinks.default-sink.hdfs.rollCount = 0
      default-agent.sinks.default-sink.hdfs.rollInterval = 86400
      default-agent.sinks.default-sink.hdfs.batchSize = 51200
      default-agent.sources.default-source.bind = 0.0.0.0
      default-agent.sources.default-source.port = ****
      default-agent.sources.default-source.channels =  default-channel
      default-agent.channels.default-channel.transactionCapacity = 10000
      default-agent.channels.default-channel.dataDirs = ****
      default-agent.channels.default-channel.checkpointDir = ****
      default-agent.channels.default-channel.capacity = 1000000
      参数 描述
      default-agent.sinks 指定所有Sinks名称。例如,default-sink。
      default-agent.sources 指定所有Sources名称。例如,default-source。
      default-agent.channels 指定所有Channels名称。例如,default-channel。
      default-agent.sinks.default-sink.hdfs.path HDFS路径。
      • 高可用集群:例如hdfs://emr-cluster/path
      • 非高可用集群:例如hdfs://master-1-1:9000/path
      default-agent.sinks.default-sink.hdfs.fileType 固定值为DataStream。
      default-agent.sinks.default-sink.hdfs.rollSize 当临时文件达到该参数值时,滚动成目标文件,单位:byte。

      该值设置为0,则表示文件不根据文件大小滚动生成。

      default-agent.sinks.default-sink.hdfs.rollCount 当Event数据达到该数量时,将临时文件滚动生成目标文件。

      该值设置为0,则表示文件不根据Event数滚动生成。

      default-agent.sinks.default-sink.hdfs.rollInterval 设置多久生成一个新的文件,单位为秒。例如,86400。
      default-agent.sinks.default-sink.hdfs.batchSize 设置每满多少Event,刷新一次到HDFS。例如,51200。
      default-agent.sinks.default-sink.channel 指定default-sink的Channel名称。
      default-agent.sources.default-source.bind 绑定的端口。设置为0.0.0.0表示绑定机器所有的接口。
      default-agent.sources.default-source.port 监听的端口。根据实际情况设置。
      default-agent.sources.default-source.channels 指定default-source的Channel名称。
      default-agent.channels.default-channel.transactionCapacity Channel在一次事务中传递Event的最大数量。默认值为10000。
      default-agent.channels.default-channel.dataDirs Channel存储Event数据的路径。

      可选参数,默认路径~/.flume/file-channel/data

      default-agent.channels.default-channel.checkpointDir 存储Checkpoint的路径。

      可选参数,默认路径~/.flume/file-channel/checkpoint

      default-agent.channels.default-channel.capacity 根据HDFS Roll设置。

      可选参数,默认值为1000000。

    4. 保存配置。
      1. 单击下方的保存
      2. 在弹出的对话框中,输入执行原因,单击确定
  4. 请参见步骤3配置core-1-2节点的Flume Agent并保存配置。
  5. 配置Master节点组的Flume Agent并保存配置。
    以下配置项请遵循开源Flume内容,详情请参见Flume
    1. 在上方的下拉列表中选择独立节点配置master-1-1
    2. 请根据实际情况修改参数flume-conf.properties中的值。
      default-agent.sinks = default-sink k1
      default-agent.sources = default-source
      default-agent.channels = default-channel
      default-agent.sources.default-source.type = taildir
      default-agent.sinks.default-sink.type = avro
      default-agent.sinks.default-sink.channel =  default-channel
      default-agent.channels.default-channel.type = file
      default-agent.sources.default-source.filegroups = f1
      default-agent.sources.default-source.filegroups.f1 = /mnt/disk1/log/hadoop-hdfs/hdfs-audit.log.*
      default-agent.sources.default-source.positionFile = ~/.flume/taildir_position.json
      default-agent.sources.default-source.channels =  default-channel
      default-agent.sources.default-source.batchSize = 2000
      default-agent.sources.default-source.ignoreRenameWhenMultiMatching = true
      default-agent.channels.default-channel.checkpointDir = ****
      default-agent.channels.default-channel.dataDirs = ****
      default-agent.channels.default-channel.capacity = ****
      default-agent.channels.default-channel.transactionCapacity = 2000
      default-agent.sinkgroups = g1
      default-agent.sinkgroups.g1.sinks = default-sink k1
      default-agent.sinkgroups.g1.processor.type = failover
      default-agent.sinkgroups.g1.processor.priority.default-sink = 10
      default-agent.sinkgroups.g1.processor.priority.k1 = 5
      default-agent.sinks.default-sink.hostname = ****
      default-agent.sinks.default-sink.port = ****
      default-agent.sinks.k1.hostname = ****
      default-agent.sinks.k1.port = ****
      default-agent.sinks.default-sink.batch-size = 2000
      default-agent.sinks.k1.batch-size = 2000
      default-agent.sinks.k1.type = avro
      default-agent.sinks.k1.channel = default-channel
      参数 描述
      default-agent.sinks 指定所有Sinks名称。例如,default-sink k1。
      default-agent.sources 指定所有Sources名称。例如,default-source。
      default-agent.channels 指定所有Channels名称。例如,default-channel。
      default-agent.sources.default-source.filegroups.f1 生成日志的路径。默认路径为/mnt/disk1/log/hadoop-hdfs/hdfs-audit.log.*
      default-agent.sources.default-source.positionFile 存储Position File的路径。

      可选参数,默认路径为~/.flume/taildir_position.json

      default-agent.channels.default-channel.checkpointDir 存储Checkpoint的路径。
      default-agent.channels.default-channel.dataDirs 存储Event数据的路径。
      default-agent.channels.default-channel.capacity 根据HDFS Roll设置。
      default-agent.sources.default-source.batchSize 在一个批处理中写入通道的最大消息数。例如,2000。
      default-agent.channels.default-channel.transactionCapacity 每次Channel从Source获取事件或推送给Sink的最大事件数。例如,2000。
      default-agent.sources.default-source.ignoreRenameWhenMultiMatching Flume的taildir source在filegroups使用通配符匹配log4j的滚动日志时会有数据重复的问题,设置为true可以避免该问题。
      default-agent.sinkgroups 指定所有sinkgroups名称。例如,g1。
      default-agent.sinkgroups.g1.sinks 指定sinkgroup g1中所有sink名称。例如,default-sink k1。
      default-agent.sinkgroups.g1.processor.type 指定sinkgroup g1的sink组逻辑处理类型,取值为:
      • default:默认一对一。
      • failover:故障转移。
      • load_balance:负载均衡。
      default-agent.sinkgroups.g1.processor.priority.default-sink 指定sinkgroup g1组内sink,default-sink的权重值,数值越大会被优先使用。例如,设置为10。
      default-agent.sinkgroups.g1.processor.priority.k1 指定sinkgroup g1组内sink,k1的权重值,数值越大会被优先使用。例如,设置为5。
      default-agent.sinks.default-sink.hostname core-1-1节点的IP地址。
      default-agent.sinks.default-sink.port core-1-1节点Flume Agent的Port。
      default-agent.sinks.k1.hostname core-1-2节点的IP地址。
      default-agent.sinks.k1.port core-1-2节点Flume Agent的Port。
      default-agent.sinks.default-sink.batch-size default-sink每次批量发送的Event数。例如,2000。
      default-agent.sinks.k1.batch-size k1每次批量发送的Event数。例如,2000。
      default-agent.sinks.k1.type 指定Sink类型。例如,avro。
      default-agent.sinks.k1.channel 配置Sink。例如,default-channel。
    3. 保存配置。
      1. 单击下方的保存
      2. 在弹出的对话框中,输入执行原因,单击确定
  6. 启动Flume Agent。
    1. 在右上角选择更多操作 > 重启
    2. 在弹出的对话框中,输入执行原因,单击确定
    3. 确认对话框中,单击确定
      执行完成后,日志就从HDFS Audit同步至HDFS了。

      Flume Agent日志的存放路径为/var/log/emr/flume/default-agent/flume.log