日志采集框架Flume 和 工作流调度器azkaban

• • 1.1 Flume介绍
  • • 1.1.1 概述
    • 1.1.2 运行机制
    • 1.1.3 Flume采集系统结构图
  • 1.2 Flume实战案例
  • • 1.2.1 Flume的安装部署
    • • 第一步：下载解压修改配置文件
      • 第二步：开发配置文件
      • 第三步：启动配置文件
      • 第四步：安装telent准备测试
    • 1.2.2 采集案例
    • • 1、采集目录到HDFS
      • 2、 采集文件到HDFS
      • 3、两个agent级联
  • 1.3 更多source和sink组件
  • 1.4 高可用Flum-NG配置案例failover
  • • 1.4.1、角色分配
    • 1.4.2、node01安装配置flume与拷贝文件脚本
    • 1.4.4、顺序启动命令
    • 1.4.5、 FAILOVER测试
  • 1.5、flume的负载均衡load balancer
• 2. 工作流调度器azkaban
• • 2.1 概述
  • 2.1.1为什么需要工作流调度系统
  • • 2.1.2 工作流调度实现方式
    • 2.1.3 常见工作流调度系统
    • 2.1.4 各种调度工具特性对比
    • 2.1.5 Azkaban与Oozie对比
  • 2.2 Azkaban介绍
  • 2. 3 Azkaban安装部署
  • • 2.3.1、azkaban的编译
    • 2.3.2、azkaban单服务模式安装与使用
    • 2.3.3、azkaban两个服务模式安装
    • • 1、确认所需软件：
      • 2、数据库准备
      • 3、解压软件安装包
      • 4、安装SSL安全认证
      • 5、azkaban web server安装
      • 6、azkaban executor server 安装
      • 7、启动服务
      • 修改linux的时区问题
  • 2.4 Azkaban实战

1.1 Flume介绍

1.1.1 概述

• Flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。
• Flume可以采集文件，socket数据包、文件、文件夹、kafka等各种形式源数据，又可以将采集到的数据(下沉sink)输出到HDFS、hbase、hive、kafka等众多外部存储系统中
• 一般的采集需求，通过对flume的简单配置即可实现
• Flume针对特殊场景也具备良好的自定义扩展能力，
  因此，flume可以适用于大部分的日常数据采集场景

1.1.2 运行机制

1、 Flume分布式系统中最核心的角色是agent，flume采集系统就是由一个个agent所连接起来形成
2、 每一个agent相当于一个数据传递员 ，内部有三个组件：
a) Source：采集组件，用于跟数据源对接，以获取数据
b) Sink：下沉组件，用于往下一级agent传递数据或者往最终存储系统传递数据
c) Channel：传输通道组件，用于从source将数据传递到sink

1.1.3 Flume采集系统结构图

1. 简单结构
   单个agent采集数据
   
2. 复杂结构
   多级agent之间串联
   

1.2 Flume实战案例

1.2.1 Flume的安装部署

案例：使用网络telent命令向一台机器发送一些网络数据，然后通过flume采集网络端口数据

第一步：下载解压修改配置文件

Flume的安装非常简单，只需要解压即可，当然，前提是已有hadoop环境
上传安装包到数据源所在节点上
这里我们采用在第三台机器来进行安装
tar -zxvf flume-ng-1.6.0-cdh5.14.0.tar.gz -C /export/servers/
cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
cp flume-env.sh.template flume-env.sh
vim flume-env.sh
export JAVA_HOME=/export/servers/jdk1.8.0_141

第二步：开发配置文件

根据数据采集的需求配置采集方案，描述在配置文件中(文件名可任意自定义)

配置我们的网络收集的配置文件
在flume的conf目录下新建一个配置文件（采集方案）

vim   /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf/netcat-logger.conf

• 定义这个agent中各组件的名字
  a1.sources = r1
  a1.sinks = k1
  a1.channels = c1

• 描述和配置source组件：r1
  a1.sources.r1.type = netcat
  a1.sources.r1.bind = 192.168.52.120
  a1.sources.r1.port = 44444

• 描述和配置sink组件：k1
  a1.sinks.k1.type = logger

• 描述和配置channel组件，此处使用是内存缓存的方式
  a1.channels.c1.type = memory
  a1.channels.c1.capacity = 1000
  a1.channels.c1.transactionCapacity = 100

• 描述和配置source channel sink之间的连接关系
  a1.sources.r1.channels = c1
  a1.sinks.k1.channel = c1

第三步：启动配置文件

指定采集方案配置文件，在相应的节点上启动flume agent

先用一个最简单的例子来测试一下程序环境是否正常
启动agent去采集数据

bin/flume-ng agent -c conf -f conf/netcat-logger.conf -n a1  -Dflume.root.logger=INFO,console

-c conf 指定flume自身的配置文件所在目录
-f conf/netcat-logger.con 指定我们所描述的采集方案
-n a1 指定我们这个agent的名字

第四步：安装telent准备测试

在node02机器上面安装telnet客户端，用于模拟数据的发送
yum -y install telnet
telnet node03 44444 # 使用telnet模拟数据发送

1.2.2 采集案例

1、采集目录到HDFS

• 需求分析
  结构示意图：
  
  采集需求：某服务器的某特定目录下，会不断产生新的文件，每当有新文件出现，就需要把文件采集到HDFS中去
  根据需求，首先定义以下3大要素

1. 数据源组件，即source ——监控文件目录 : spooldir
   spooldir特性：
   1、监视一个目录，只要目录中出现新文件，就会采集文件中的内容
   2、采集完成的文件，会被agent自动添加一个后缀：COMPLETED
   3、所监视的目录中不允许重复出现相同文件名的文件
2. 下沉组件，即sink——HDFS文件系统 : hdfs sink
3. 通道组件，即channel——可用file channel 也可以用内存channel

• flume配置文件开发
  配置文件编写：

cd  /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
mkdir -p /export/servers/dirfile
vim spooldir.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
##注意：不能往监控目中重复丢同名文件
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /export/servers/dirfile
a1.sources.r1.fileHeader = true
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = hdfs://node01:8020/spooldir/files/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本
a1.sinks.k1.hdfs.fileType = DataStream
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

Channel参数解释：
capacity：默认该通道中最大的可以存储的event数量
trasactionCapacity：每次最大可以从source中拿到或者送到sink中的event数量
keep-alive：event添加到通道中或者移出的允许时间

• 启动flume

bin/flume-ng agent -c ./conf -f ./conf/spooldir.conf -n a1 -Dflume.root.logger=INFO,console

• 上传文件到指定目录
  将不同的文件上传到下面目录里面去，注意文件不能重名

cd /export/servers/dirfile

2、 采集文件到HDFS

• 需求分析：
  采集需求：比如业务系统使用log4j生成的日志，日志内容不断增加，需要把追加到日志文件中的数据实时采集到hdfs

根据需求，首先定义以下3大要素

1. 采集源，即source——监控文件内容更新 : exec ‘tail -F file’
2. 下沉目标，即sink——HDFS文件系统 : hdfs sink
3. Source和sink之间的传递通道——channel，可用file channel 也可以用 内存channel

定义flume的配置文件
node03开发配置文件

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim tail-file.conf

配置文件内容

agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1

# Describe/configure tail -F source1
agent1.sources.source1.type = exec
agent1.sources.source1.command = tail -F /export/servers/taillogs/access_log
agent1.sources.source1.channels = channel1

#configure host for source
#agent1.sources.source1.interceptors = i1
#agent1.sources.source1.interceptors.i1.type = host
#agent1.sources.source1.interceptors.i1.hostHeader = hostname

# Describe sink1
agent1.sinks.sink1.type = hdfs
#a1.sinks.k1.channel = c1
agent1.sinks.sink1.hdfs.path = hdfs://node01:8020/weblog/flume-collection/%y-%m-%d/%H-%M
agent1.sinks.sink1.hdfs.filePrefix = access_log
agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
agent1.sinks.sink1.hdfs.batchSize= 100
agent1.sinks.sink1.hdfs.fileType = DataStream
agent1.sinks.sink1.hdfs.writeFormat =Text
agent1.sinks.sink1.hdfs.rollSize = 102400
agent1.sinks.sink1.hdfs.rollCount = 1000000
agent1.sinks.sink1.hdfs.rollInterval = 60
agent1.sinks.sink1.hdfs.round = true
agent1.sinks.sink1.hdfs.roundValue = 10
agent1.sinks.sink1.hdfs.roundUnit = minute
agent1.sinks.sink1.hdfs.useLocalTimeStamp = true

# Use a channel which buffers events in memory
agent1.channels.channel1.type = memory
agent1.channels.channel1.keep-alive = 120
agent1.channels.channel1.capacity = 500000
agent1.channels.channel1.transactionCapacity = 600

# Bind the source and sink to the channel
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1

• 启动flume

cd  /export/servers/apache-flume-1.6.0-cdh5.14.0-bin

bin/flume-ng agent -c conf -f conf/tail-file.conf -n agent1  -Dflume.root.logger=INFO,console

• 开发shell脚本定时追加文件内容

mkdir -p /export/servers/shells/
cd  /export/servers/shells/
vim tail-file.sh

#!/bin/bash
while true
do
 date >> /export/servers/taillogs/access_log;
  sleep 0.5;
done

创建文件夹

mkdir -p /export/servers/taillogs

启动脚本

sh /export/servers/shells/tail-file.sh

3、两个agent级联

• 需求分析：
  第一个agent负责收集文件当中的数据，通过网络发送到第二个agent当中去，第二个agent负责接收第一个agent发送的数据，并将数据保存到hdfs上面去

• 第一步：node02安装flume
  将node03机器上面解压后的flume文件夹拷贝到node02机器上面去
  cd /export/servers
  scp -r apache-flume-1.6.0-cdh5.14.0-bin/ node02:$PWD
  

• 第二步：node02配置flume配置文件
  在node02机器配置我们的flume

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim tail-avro-avro-logger.conf

##################
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/servers/taillogs/access_log
a1.sources.r1.channels = c1
# Describe the sink
##sink端的avro是一个数据发送者
a1.sinks = k1
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = 192.168.52.120
a1.sinks.k1.port = 4141
a1.sinks.k1.batch-size = 10
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
第三步：node02开发脚本文件往文件写入数据
直接将node03下面的脚本和数据拷贝到node02即可，node03机器上执行以下命令
cd  /export/servers
scp -r shells/ taillogs/ node02:$PWD
 
第五步：node03开发flume配置文件
在node03机器上开发flume的配置文件
cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim avro-hdfs.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
##source中的avro组件是一个接收者服务
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 192.168.52.120
a1.sources.r1.port = 4141
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://node01:8020/avro/hdfs/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本
a1.sinks.k1.hdfs.fileType = DataStream
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

第六步：顺序启动

node03机器启动flume进程
cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -c conf -f conf/avro-hdfs.conf -n a1  -Dflume.root.logger=INFO,console   

node02机器启动flume进程

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/
bin/flume-ng agent -c conf -f conf/tail-avro-avro-logger.conf -n a1  -Dflume.root.logger=INFO,console    

node02机器启shell脚本生成文件

cd  /export/servers/shells
sh tail-file.sh

1.3 更多source和sink组件

Flume支持众多的source和sink类型，详细手册可参考官方文档
http://archive.cloudera.com/cdh5/cdh/5/flume-ng-1.6.0-cdh5.14.0/FlumeUserGuide.html

1.4 高可用Flum-NG配置案例failover

在完成单点的Flume NG搭建后，下面我们搭建一个高可用的Flume NG集群，架构图如下所示：

图中，我们可以看出，Flume的存储可以支持多种，这里只列举了HDFS和Kafka（如：存储最新的一周日志，并给Storm系统提供实时日志流）。

1.4.1、角色分配

Flume的Agent和Collector分布如下表所示：

名称	HOST	角色
Agent1	node01	Web Server
Collector1	node02	AgentMstr1
Collector2	node03	AgentMstr2

图中所示，Agent1数据分别流入到Collector1和Collector2，Flume NG本身提供了Failover机制，可以自动切换和恢复。在上图中，有3个产生日志服务器分布在不同的机房，要把所有的日志都收集到一个集群中存储。下 面我们开发配置Flume NG集群

1.4.2、node01安装配置flume与拷贝文件脚本

将node03机器上面的flume安装包以及文件生产的两个目录拷贝到node01机器上面去

node03机器执行以下命令

cd /export/servers
scp -r apache-flume-1.6.0-cdh5.14.0-bin/ node01:$PWD
scp -r shells/ taillogs/ node01:$PWD

node01机器配置agent的配置文件

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim agent.conf

#agent1 name
agent1.channels = c1
agent1.sources = r1
agent1.sinks = k1 k2
#
##set gruop
agent1.sinkgroups = g1
#
##set channel
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 1000
agent1.channels.c1.transactionCapacity = 100
#
agent1.sources.r1.channels = c1
agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F /export/servers/taillogs/access_log
#
agent1.sources.r1.interceptors = i1 i2
agent1.sources.r1.interceptors.i1.type = static
agent1.sources.r1.interceptors.i1.key = Type
agent1.sources.r1.interceptors.i1.value = LOGIN
agent1.sources.r1.interceptors.i2.type = timestamp
#
## set sink1
agent1.sinks.k1.channel = c1
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = node02
agent1.sinks.k1.port = 52020
#
## set sink2
agent1.sinks.k2.channel = c1
agent1.sinks.k2.type = avro
agent1.sinks.k2.hostname = node03
agent1.sinks.k2.port = 52020
#
##set sink group
agent1.sinkgroups.g1.sinks = k1 k2
#
##set failover
agent1.sinkgroups.g1.processor.type = failover
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 1
agent1.sinkgroups.g1.processor.maxpenalty = 10000
#




1.4.3、node02与node03配置flumecollection
node02机器修改配置文件
cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim collector.conf

#set Agent name
a1.sources = r1
a1.channels = c1
a1.sinks = k1
#
##set channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
#
## other node,nna to nns
a1.sources.r1.type = avro
a1.sources.r1.bind = node02
a1.sources.r1.port = 52020
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = Collector
a1.sources.r1.interceptors.i1.value = node02
a1.sources.r1.channels = c1
#
##set sink to hdfs
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path= hdfs://node01:8020/flume/failover/
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
a1.sinks.k1.hdfs.rollInterval=10
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d
#


node03机器修改配置文件
cd  /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim collector.conf

#set Agent name
a1.sources = r1
a1.channels = c1
a1.sinks = k1
#
##set channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
#
## other node,nna to nns
a1.sources.r1.type = avro
a1.sources.r1.bind = node03
a1.sources.r1.port = 52020
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = Collector
a1.sources.r1.interceptors.i1.value = node03
a1.sources.r1.channels = c1
#
##set sink to hdfs
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path= hdfs://node01:8020/flume/failover/
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
a1.sinks.k1.hdfs.rollInterval=10
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

1.4.4、顺序启动命令

node03机器上面启动flume

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/collector.conf -Dflume.root.logger=DEBUG,console

node02机器上面启动flume

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/collector.conf -Dflume.root.logger=DEBUG,console

node01机器上面启动flume

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -n agent1 -c conf -f conf/agent.conf -Dflume.root.logger=DEBUG,console

node01机器启动文件产生脚本

cd  /export/servers/shells
sh tail-file.sh

1.4.5、 FAILOVER测试

下面我们来测试下Flume NG集群的高可用（故障转移）。场景如下：我们在Agent1节点上传文件，由于我们配置Collector1的权重比Collector2大，所以 Collector1优先采集并上传到存储系统。然后我们kill掉Collector1，此时有Collector2负责日志的采集上传工作，之后，我 们手动恢复Collector1节点的Flume服务，再次在Agent1上次文件，发现Collector1恢复优先级别的采集工作。具体截图如下所 示：

Collector1优先上传

HDFS集群中上传的log内容预览

Collector1宕机，Collector2获取优先上传权限

重启Collector1服务，Collector1重新获得优先上传的权限

1.5、flume的负载均衡load balancer

负载均衡是用于解决一台机器(一个进程)无法解决所有请求而产生的一种算法。Load balancing Sink Processor 能够实现 load balance 功能，如下图Agent1 是一个路由节点，负责将 Channel 暂存的 Event 均衡到对应的多个
Sink组件上，而每个 Sink 组件分别连接到一个独立的 Agent 上，示例配置，如下所示：

在此处我们通过三台机器来进行模拟flume的负载均衡
三台机器规划如下：
node01：采集数据，发送到node02和node03机器上去
node02：接收node01的部分数据
node03：接收node01的部分数据

• 第一步：开发node01服务器的flume配置

node01服务器配置：

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim load_banlancer_client.conf

#agent name
a1.channels = c1
a1.sources = r1
a1.sinks = k1 k2

#set gruop
a1.sinkgroups = g1

#set channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

a1.sources.r1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/servers/taillogs/access_log

# set sink1
a1.sinks.k1.channel = c1
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = node02
a1.sinks.k1.port = 52020

# set sink2
a1.sinks.k2.channel = c1
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = node03
a1.sinks.k2.port = 52020

#set sink group
a1.sinkgroups.g1.sinks = k1 k2

#set failover
a1.sinkgroups.g1.processor.type = load_balance
a1.sinkgroups.g1.processor.backoff = true
a1.sinkgroups.g1.processor.selector = round_robin
a1.sinkgroups.g1.processor.selector.maxTimeOut=10000

• 第二步：开发node02服务器的flume配置

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim load_banlancer_server.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = node02
a1.sources.r1.port = 52020

# Describe the sink
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

• 第三步：开发node03服务器flume配置

node03服务器配置

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim load_banlancer_server.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = node03
a1.sources.r1.port = 52020

# Describe the sink
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

• 第四步：准备启动flume服务
  启动node03的flume服务

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/load_banlancer_server.conf -Dflume.root.logger=DEBUG,console

启动node02的flume服务

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/load_banlancer_server.conf -Dflume.root.logger=DEBUG,console

启动node01的flume服务

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/load_banlancer_client.conf -Dflume.root.logger=DEBUG,console

第五步：node01服务器运行脚本产生数据

cd /export/servers/shells
sh tail-file.sh

2. 工作流调度器azkaban

2.1 概述

azkaban官网：https://azkaban.github.io/

2.1.1为什么需要工作流调度系统

1. 一个完整的数据分析系统通常都是由大量任务单元组成：
   shell脚本程序，java程序，mapreduce程序、hive脚本等
2. 各任务单元之间存在时间先后及前后依赖关系
3. 为了很好地组织起这样的复杂执行计划，需要一个工作流调度系统来调度执行；

例如，我们可能有这样一个需求，某个业务系统每天产生20G原始数据，我们每天都要对其进行处理，处理步骤如下所示：

• 1、 通过Hadoop先将原始数据同步到HDFS上；
• 2、 借助MapReduce计算框架对原始数据进行转换，生成的数据以分区表的形式存储到多张Hive表中；
• 3、 需要对Hive中多个表的数据进行JOIN处理，得到一个明细数据Hive大表；
• 4、 将明细数据进行各种统计分析，得到结果报表信息；
• 5、 需要将统计分析得到的结果数据同步到业务系统中，供业务调用使用。

2.1.2 工作流调度实现方式

简单的任务调度：直接使用linux的crontab来定义；
复杂的任务调度：开发调度平台或使用现成的开源调度系统，比如ooize、azkaban、airflow等

2.1.3 常见工作流调度系统

市面上目前有许多工作流调度器
在hadoop领域，常见的工作流调度器有Oozie, Azkaban,Cascading,Hamake等

2.1.4 各种调度工具特性对比

下面的表格对上述四种hadoop工作流调度器的关键特性进行了比较，尽管这些工作流调度器能够解决的需求场景基本一致，但在设计理念，目标用户，应用场景等方面还是存在显著的区别，在做技术选型的时候，可以提供参考

2.1.5 Azkaban与Oozie对比

对市面上最流行的两种调度器，给出以下详细对比，以供技术选型参考。总体来说，ooize相比azkaban是一个重量级的任务调度系统，功能全面，但配置使用也更复杂。如果可以不在意某些功能的缺失，轻量级调度器azkaban是很不错的候选对象。
详情如下：

1. 功能
   两者均可以调度mapreduce,pig,java,脚本工作流任务
   两者均可以定时执行工作流任务

2. 工作流定义
   Azkaban使用Properties文件定义工作流
   Oozie使用XML文件定义工作流

3. 工作流传参
   Azkaban支持直接传参，例如 i n p u t O o z i e 支 持 参 数 和 E L 表 达 式 ， 例 如 {input} Oozie支持参数和EL表达式，例如 inputOozie支持参数和EL表达式，例如{fs:dirSize(myInputDir)}

4. 定时执行
   Azkaban的定时执行任务是基于时间的
   Oozie的定时执行任务基于时间和输入数据

5. 资源管理
   Azkaban有较严格的权限控制，如用户对工作流进行读/写/执行等操作
   Oozie暂无严格的权限控制

6. 工作流执行
   Azkaban有两种运行模式，分别是solo server mode(executor server和web server部署在同一台节点)和multi server mode(executor server和web server可以部署在不同节点)
   Oozie作为工作流服务器运行，支持多用户和多工作流

7. 工作流管理
   Azkaban支持浏览器以及ajax方式操作工作流
   Oozie支持命令行、HTTP REST、Java API、浏览器操作工作流

2.2 Azkaban介绍

Azkaban是由Linkedin开源的一个批量工作流任务调度器。用于在一个工作流内以一个特定的顺序运行一组工作和流程。
Azkaban定义了一种KV文件(properties)格式来建立任务之间的依赖关系，并提供一个易于使用的web用户界面维护和跟踪你的工作流。
它有如下功能特点：

 Web用户界面
 方便上传工作流
 方便设置任务之间的关系
 调度工作流
 认证/授权(权限的工作)
 能够杀死并重新启动工作流
 模块化和可插拔的插件机制
 项目工作区
 工作流和任务的日志记录和审计

2. 3 Azkaban安装部署

2.3.1、azkaban的编译

我们这里选用azkaban3.51.0这个版本自己进行重新编译，编译完成之后得到我们需要的安装包进行安装

注意：我们这里编译需要使用jdk1.8的版本来进行编译，如果编译服务器使用的jdk版本是1.7的，记得切换成jdk1.8，我们这里使用的是jdk8u141这个版本来进行编译

cd /export/softwares/
wget https://github.com/azkaban/azkaban/archive/3.51.0.tar.gz
tar -zxvf 3.51.0.tar.gz -C ../servers/
cd /export/servers/azkaban-3.51.0/
yum -y install git
yum -y install gcc-c++
./gradlew build installDist -x test

编译之后需要的安装文件列表如下

• azkaban-exec-server

编译完成之后得到我们需要的安装包在以下目录下即可获取得到
azkaban-exec-server存放目录

/export/servers/azkaban-3.51.0/azkaban-exec-server/build/distributions 

• azkaban-web-server
  azkaban-web-server存放目录

/export/servers/azkaban-3.51.0/azkaban-web-server/build/distributions

• azkaban-solo-server
  azkaban-solo-server存放目录

/export/servers/azkaban-3.51.0/azkaban-solo-server/build/distributions

• execute-as-user.c
  azkaban two server模式下需要的C程序在这个路径下面

/export/servers/azkaban-3.51.0/az-exec-util/src/main/c

• 数据库脚本文件
  数据库脚本文件在这个路径下面

/export/servers/azkaban-3.51.0/azkaban-db/build/install/azkaban-db

2.3.2、azkaban单服务模式安装与使用

所需软件
azkaban-solo-server
单服务模式安装

• 第一步：解压
  azkaban 的solo server使用的是一个单节点的模式来进行启动服务的，只需要一个
  azkaban-solo-server-0.1.0-SNAPSHOT.tar.gz的安装包即可启动，所有的数据信息都是保存在H2这个azkaban默认的数据当中，
  上传我们的压缩包，然后修改配置文件启动即可

cd /export/softwares
tar -zxvf azkaban-solo-server-0.1.0-SNAPSHOT.tar.gz -C ../servers/

• 第二步：修改两个配置文件
  修改时区配置文件

cd /export/servers/azkaban-solo-server-0.1.0-SNAPSHOT/conf
vim azkaban.properties

default.timezone.id=Asia/Shanghai

修改commonprivate.properties配置文件

cd /export/servers/azkaban-solo-server-0.1.0-SNAPSHOT/plugins/jobtypes
vim commonprivate.properties

execute.as.user=false
memCheck.enabled=false

• 第三步：启动solo-server

启动azkaban-solo-server

cd  /export/servers/azkaban-solo-server-0.1.0-SNAPSHOT
bin/start-solo.sh

• 第四步：浏览器页面访问
  浏览器页面访问 http://node03:8081/

• 单服务模式使用
  需求：使用azkaban调度我们的shell脚本，执行linux的shell命令

创建普通文本文件 foo.job，文件内容如下

type=command
command=echo "hello world"

然后将这个文件打包为压缩文件，如下：

azkaban上传我们的压缩包

2.3.3、azkaban两个服务模式安装

1、确认所需软件：

• Azkaban Web服务安装包
  azkaban-web-server-0.1.0-SNAPSHOT.tar.gz

• Azkaban执行服务安装包
  azkaban-exec-server-0.1.0-SNAPSHOT.tar.gz

• 编译之后的sql脚本
  create-all-sql-0.1.0-SNAPSHOT.sql
  

• C程序文件脚本
  execute-as-user.c程序

2、数据库准备

进入mysql的客户端执行以下命令

mysql  -uroot -p

执行以下命令：

CREATE DATABASE azkaban;
CREATE USER 'azkaban'@'%' IDENTIFIED BY 'azkaban';    
GRANT all privileges ON azkaban.* to 'azkaban'@'%' identified by 'azkaban' WITH GRANT OPTION; 
flush privileges;
use azkaban; 
source /export/softwares/create-all-sql-0.1.0-SNAPSHOT.sql;

3、解压软件安装包

解压azkaban-web-server

cd /export/softwares
tar -zxvf azkaban-web-server-0.1.0-SNAPSHOT.tar.gz -C ../servers/
cd /export/servers
mv azkaban-web-server-0.1.0-SNAPSHOT/ azkaban-web-server-3.51.0

解压azkaban-exec-server

cd /export/softwares
tar -zxvf azkaban-exec-server-0.1.0-SNAPSHOT.tar.gz -C ../servers/
cd /export/servers
mv azkaban-exec-server-0.1.0-SNAPSHOT/ azkaban-exec-server-3.51.0

4、安装SSL安全认证

安装ssl安全认证，允许我们使用https的方式访问我们的azkaban的web服务
密码一定要一个个的字母输入，或者粘贴也行

cd /export/servers/azkaban-web-server-3.51.0
keytool -keystore keystore -alias jetty -genkey -keyalg RSA

密码一直

输入azkaban

5、azkaban web server安装

修改azkaban-web-server的配置文件

cd /export/servers/azkaban-web-server-3.51.0/conf
vim azkaban.properties

# Azkaban Personalization Settings
azkaban.name=Azkaban
azkaban.label=My Azkaban
azkaban.color=#FF3601
azkaban.default.servlet.path=/index
web.resource.dir=web/
default.timezone.id=Asia/Shanghai
# Azkaban UserManager class
user.manager.class=azkaban.user.XmlUserManager
user.manager.xml.file=conf/azkaban-users.xml
# Loader for projects
executor.global.properties=conf/global.properties
azkaban.project.dir=projects
# Velocity dev mode
velocity.dev.mode=false
# Azkaban Jetty server properties.
jetty.use.ssl=true
jetty.maxThreads=25
jetty.port=8081

jetty.ssl.port=8443
jetty.keystore=/export/servers/azkaban-web-server-3.51.0/keystore
jetty.password=azkaban
jetty.keypassword=azkaban
jetty.truststore=/export/servers/azkaban-web-server-3.51.0/keystore
jetty.trustpassword=azkaban


# Azkaban Executor settings
# mail settings
mail.sender=
mail.host=
# User facing web server configurations used to construct the user facing server URLs. They are useful when there is a reverse proxy between Azkaban web servers and users.
# enduser -> myazkabanhost:443 -> proxy -> localhost:8081
# when this parameters set then these parameters are used to generate email links.
# if these parameters are not set then jetty.hostname, and jetty.port(if ssl configured jetty.ssl.port) are used.
# azkaban.webserver.external_hostname=myazkabanhost.com
# azkaban.webserver.external_ssl_port=443
# azkaban.webserver.external_port=8081
job.failure.email=
job.success.email=
lockdown.create.projects=false
cache.directory=cache
# JMX stats
jetty.connector.stats=true
executor.connector.stats=true
# Azkaban mysql settings by default. Users should configure their own username and password.
database.type=mysql
mysql.port=3306
mysql.host=node03
mysql.database=azkaban
mysql.user=azkaban
mysql.password=azkaban
mysql.numconnections=100
#Multiple Executor
azkaban.use.multiple.executors=true
#azkaban.executorselector.filters=StaticRemainingFlowSize,MinimumFreeMemory,CpuStatus
azkaban.executorselector.comparator.NumberOfAssignedFlowComparator=1
azkaban.executorselector.comparator.Memory=1
azkaban.executorselector.comparator.LastDispatched=1
azkaban.executorselector.comparator.CpuUsage=1

azkaban.activeexecutor.refresh.milisecinterval=10000
azkaban.queueprocessing.enabled=true
azkaban.activeexecutor.refresh.flowinterval=10
azkaban.executorinfo.refresh.maxThreads=10

6、azkaban executor server 安装

• 第一步：修改azkaban-exex-server配置文件
  修改azkaban-exec-server的配置文件

cd /export/servers/azkaban-exec-server-3.51.0/conf
vim azkaban.properties

# Azkaban Personalization Settings
azkaban.name=Azkaban
azkaban.label=My Azkaban
azkaban.color=#FF3601
azkaban.default.servlet.path=/index
web.resource.dir=web/
default.timezone.id=Asia/Shanghai
# Azkaban UserManager class
user.manager.class=azkaban.user.XmlUserManager
user.manager.xml.file=conf/azkaban-users.xml
# Loader for projects
executor.global.properties=conf/global.properties
azkaban.project.dir=projects
# Velocity dev mode
velocity.dev.mode=false
# Azkaban Jetty server properties.
jetty.use.ssl=true
jetty.maxThreads=25
jetty.port=8081


jetty.keystore=/export/servers/azkaban-web-server-3.51.0/keystore
jetty.password=azkaban
jetty.keypassword=azkaban
jetty.truststore=/export/servers/azkaban-web-server-3.51.0/keystore
jetty.trustpassword=azkaban


# Where the Azkaban web server is located
azkaban.webserver.url=https://node03:8443
# mail settings
mail.sender=
mail.host=
# User facing web server configurations used to construct the user facing server URLs. They are useful when there is a reverse proxy between Azkaban web servers and users.
# enduser -> myazkabanhost:443 -> proxy -> localhost:8081
# when this parameters set then these parameters are used to generate email links.
# if these parameters are not set then jetty.hostname, and jetty.port(if ssl configured jetty.ssl.port) are used.
# azkaban.webserver.external_hostname=myazkabanhost.com
# azkaban.webserver.external_ssl_port=443
# azkaban.webserver.external_port=8081
job.failure.email=
job.success.email=
lockdown.create.projects=false
cache.directory=cache
# JMX stats
jetty.connector.stats=true
executor.connector.stats=true
# Azkaban plugin settings
azkaban.jobtype.plugin.dir=plugins/jobtypes
# Azkaban mysql settings by default. Users should configure their own username and password.
database.type=mysql
mysql.port=3306
mysql.host=node03
mysql.database=azkaban
mysql.user=azkaban
mysql.password=azkaban
mysql.numconnections=100
# Azkaban Executor settings
executor.maxThreads=50
executor.flow.threads=30

• 第二步：添加插件
  将我们编译后的C文件execute-as-user.c
  上传到这个目录来/export/servers/azkaban-exec-server-3.51.0/plugins/jobtypes
  或者直接将我们/export/softwares下面的文件拷贝过来也行

cp /export/softwares/execute-as-user.c /export/servers/azkaban-exec-server-3.51.0/plugins/jobtypes/

然后执行以下命令生成execute-as-user

yum -y install gcc-c++
cd /export/servers/azkaban-exec-server-3.51.0/plugins/jobtypes
gcc execute-as-user.c -o execute-as-user 
chown root execute-as-user
chmod 6050 execute-as-user

• 第三步：修改配置文件
  修改配置文件

cd  /export/servers/azkaban-exec-server-3.47.0/plugins/jobtypes
vim commonprivate.properties
execute.as.user=false
memCheck.enabled=false
azkaban.native.lib=/export/servers/azkaban-exec-server-3.51.0/plugins/jobtypes

最终生成如下

7、启动服务

• 第一步：启动azkaban exec server
  cd /export/servers/azkaban-exec-server-3.51.0
  bin/start-exec.sh

• 第二步：激活我们的exec-server
  node03机器任意目录下执行以下命令
  curl -G “node03:$(<./executor.port)/executor?action=activate” && echo

• 第三步：启动azkaban-web-server
  cd /export/servers/azkaban-web-server-3.51.0/
  bin/start-web.sh

访问地址： https://node03:8443

修改linux的时区问题

由于先前做好了时钟同步，所以不用担心时区问题，不需要修改时区了
注：先配置好服务器节点上的时区
1、先生成时区配置文件Asia/Shanghai，用交互式命令 tzselect 即可
2、拷贝该时区文件，覆盖系统本地时区配置

cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime  

2.4 Azkaban实战

Azkaba内置的任务类型支持command、java

• Command类型单一job示例
• 创建job描述文件
• 创建文本文件，更改名称为mycommand.job
  注意后缀.txt一定不要带上，保存为格式为UFT-8 without bom
  内容如下

type=command
command=echo 'hello world'

• 将job资源文件打包成zip文件

• 创建project并上传压缩包
  通过azkaban的web管理平台创建project并上传job压缩包
  首先创建project
  
  上传zip包
  

启动执行job

• Command类型多job工作流flow

1、创建有依赖关系的多个job描述
第一个job：foo.job
type=command
command=echo ‘foo’
第二个job：bar.job依赖foo.job
type=command
dependencies=foo
command=echo ‘bar’

2、将所有job资源文件打到一个zip包中

3、在azkaban的web管理界面创建工程并上传zip包
4、启动工作流flow

• HDFS操作任务

1、创建job描述文件fs.job

type=command
command=/export/servers/hadoop-2.6.0-cdh5.14.0/bin/hadoop fs -mkdir /azkaban

2、将job资源文件打包成zip文件
3、通过azkaban的web管理平台创建project并上传job压缩包
4、启动执行该job

• MAPREDUCE任务
  Mr任务依然可以使用command的job类型来执行

1、创建job描述文件，及mr程序jar包（示例中直接使用hadoop自带的example jar）
type=command
command=/export/servers/hadoop-2.6.0-cdh5.14.0/bin/hadoop jar hadoop-mapreduce-examples-2.6.0-cdh5.14.0.jar pi 3 5

2、将所有job资源文件打到一个zip包中
3、在azkaban的web管理界面创建工程并上传zip包
4、启动job

• HIVE脚本任务

  创建job描述文件和hive脚本
  Hive脚本： hive.sql

create database if not exists azhive;
use azhive;
create table if not exists aztest(id string,name string) row format delimited fields terminated by '\t';

Job描述文件：hive.job

type=command
command=/export/servers/hive-1.1.0-cdh5.14.0/bin/hive -f 'hive.sql'

将所有job资源文件打到一个zip包中

在azkaban的web管理界面创建工程并上传zip包
启动job

• zkaban的定时任务
  使用azkaban的scheduler功能可以实现对我们的作业任务进行定时调度功能
  
  
  */1 * ? * * 每分钟执行一次定时调度任务
  0 1 ? * * 每天晚上凌晨一点钟执行这个任务
  0 */2 ? * * 每隔两个小时定时执行这个任务
  30 21 ? * * 每天晚上九点半定时执行这个任务