返回 登录
27

大数据实时处理实战

作者:武智晖,北京移动网络运行维护中心大数据系统架构师,北京邮电大学软件工程硕士,高级工程师。多年从事系统架构设计,软件开发,运营商大数据分析挖掘工作。

随着互联网时代的发展,运营商作为内容传送的管道服务商,在数据领域具有巨大的优势,如何将这些数据转化为价值,越来越被运营商所重视。

运营商的大数据具有体量大,种类多的特点,如各类话单、信令等,通常一种话单每天的数据量就有上百亿条。随着业务分析需求对数据处理实时性的要求越来越高,也给我们的大数据处理架构带来了巨大的挑战,参照网络上可查的例子,运用到实际处理架构上,经常会因为实时数据流量大,造成系统运行不稳定及各种异常。从大数据实时处理架构开发到上线,耗时近2个月时间,经过大量优化,我们的系统才趋于稳定。最终我们使用10台服务器的集群,实时处理每天上百亿条的数据,这里每条数据的字段数量有100个,最长的字段内容超过1000字节。

下面就来分享一下我们在实时大数据处理大体量数据的过程中,总结出来的酸甜苦辣。

  • 项目目标

在有限服务器集群数量的基础上,实现对每天超过百亿条、体量超过20T的某话单进行实时处理。具体需求是FTP收集多台话单服务器上的详单,进行实时处理后将数据存储到Hbase数据库供用户即时详单查询,同时将话单存储到Hdfs供离线分析使用。

  • 硬件资源

10台x86服务器,单机配置16盒CPU,128G内存,2T硬盘*10,300G硬盘*2(系统盘)。

  • 系统架构

10台服务器组成hadoop集群,其中NameNode节点同时作为采集机安装FTP和Flume,选取其他5台服务器安装Kafka,Zookeeper和Storm实现大数据实时流处理架构,为了充分利用集群计算资源,这5台服务器也配置了少量的Yarn计算资源,参与日常的离线数据分析需求。剩下的4台服务器我们安装了Hbase满足大数据下的秒级查询需求,系统拓扑图如下:

图片描述
图一 系统拓扑图

  • 项目实施

1、使用的相关技术

我们先来回顾一下相关的大数据架构和开源技术,大数据处理分离线分析架构和实时处理架构。离线分析架构(如Hive,Map/Reduce,Spark Sql等)可以满足数据后分析,数据挖掘的应用需求。对于实时性要求高的应用,如用户即时详单查询,业务量监控等,需要应用实时处理架构。目前大数据开源实时处理架构最常见的是Storm和Spark Streaming,相比Spark Streaming准实时批处理系统,Strom是更纯粹的实时处理系统,即来一条事件就处理一条,具有更高的实时性。

Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统。Flume支持单机也支持集群,支持多种数据源,如不断写入的文件、Socket、不断生成新文件的文件夹等,支持多种输出,如Hdfs、Kafka、Mysql数据库等。Flume使用时仅需实现简单配置,无需开发程序。

Kafka是一种高吞吐量的分布式发布订阅消息系统,类似一个大数据量的缓存池,支持一份数据多用户消费。ZooKeeper是一个分布式的,开源的分布式应用程序协调服务,负责存储集群间部分组件的状态同步信息。Storm分布式实时计算系统,包含Nimbus主节点和Supervisor从节点(从storm1.0以后,增加了Nimbus备份节点),节点之间需要依靠Zookeeper做状态同步。Storm集群组件:

  • Nimbus:是Storm集群的master节点,负责资源分配和任务调度。
  • Supervisor:是Storm集群的slave节点,负责接受nimbus分配的任务,启动和停止属于自己管理的worker进程,是真正意义上的分布式计算节点。

图片描述
图二 Storm集群组件

Storm应用涉及到Java程序的开发,编程模型中涉及的概念:

  • Topology:Storm中运行的一个实时应用程序,各个组件间的消息流动形成逻辑上的一个拓扑结构,Topology一旦启动,就会常驻内存并占用worker资源。
  • Spout:在一个Topology中产生源数据流的组件。通常情况下Spout会从外部数据源中读取数据,然后转换为Topology内部的源数据。
  • Bolt:在一个Topology中接受数据然后执行处理的组件。Bolt可以执行过滤、函数操作、合并、写数据库等任何操作。
  • Tuple:一次消息传递的基本单元。

2、开源组件安装及配置

a)Flume安装及配置

http://flume.apache.org/下载flume的安装包,解压缩;如果使用Cloudera Manager或者Ambari安装,仅需通过相应的管理页面安装配置。我们仅安装了单机的Flume,未安装Flume集群,单机Flume处理效率非常高,完全能够满足我们每天处理上百亿条数据的需求,但需要说明一点的是Flume鲁棒性非常差,经常出现进程在、但数据不处理的进程卡死状态,使用Flume时要注意以下几点:

  • flume监控目录中不能含有目录;

  • flume正在处理的文件,其他进程不能更改(如FTP正在传送中的文件,需要设置过滤条件,避免flume处理)。建议flume监控目录与FTP实时传送目录分开,避免flume处理FTP传送中的文件,导致异常,也可以设置正则表达式忽略正在传送的文件:

a1.sources.r1.ignorePattern = ^(.)*\\.tmp$
  • flume处理的文件中可能含有特殊字符,导致flume进程卡死。设置遇到不能识别的字符忽略跳过:
a1.sources.r1.decodeErrorPolicy = IGNORE
  • flume运行过程中出现GC over的内存溢出错误,配置flume-env.sh中内存配置(默认值很小);
export JAVA_OPTS="-Xms1024m -Xmx2048m -Dcom.sun.management.jmxremote"
  • flume启动时-c后面要给全到详细flume配置文件目录,否则flume-env.sh中的配置不会加载,会使用默认配置,例如下面启动命令给全配置文件目录:
/hadoop/apache-flume-1.6.0-bin/bin/flume-ng agent -c /hadoop/apache-flume-1.6.0-bin/conf/
  • 如果使用内存队列,请注意内存队列消息数的配置,设置transactionCapacity队列大小必须大于等于batchSize;
a1.channels.c1.transactionCapacity = 2000
a1.sinks.k1.batchSize = 2000
  • 增加batchSize可以提升flume处理速度,原理是flume处理的event都保存在transaction队列中,直到满足了batchSize的数量条件,才一次性批量向sink发送。但是要注意实际数据量的大小,如果实际数据量很小,batchSize就不能配置过大,否则数据达不到batchSize的数量条件,会长时间积压在transaction队列中,后面的实时处理程序反而得不到数据,导致实时性变差;

  • flume中读取的一条记录长度超过2048字符,也就是4096字节就会被截断,可以在配置文件中增加如下配置项解决:

producer.sources.s.deserializer.maxLineLength=65535
  • flume字符转换异常问题,java.nio.charset.MalformedInputException: Input length = 1,可以在配置文件中增加如下配置项解决:
a1.sources.r1.inputCharset = ISO8859-1
  • flume遇到乱码停止,报异常:java.nio.charset.MalformedInputException,可以在配置文件中增加如下配置,忽略错误数据(默认是FAIL,抛异常报错,flume会停止)解决;
producer.sources.s.decodeErrorPolicy=IGNORE
  • 默认情况下,Flume处理完成的文件会增加.completed后缀,在数据量很大的情况下,会很快撑满采集机硬盘,可以在配置文件中增加如下配置,让flume处理完后自动删除该数据文件解决。
a1.sources.r1.deletePolicy = immediate

Flume配置:

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.channels = c1
a1.sources.r1.spoolDir = /ftpdata/xdr/HTTP_tmp
a1.sources.r1.ignorePattern = ^(.)*\\.tmp$
a1.sources.r1.fileHeader = false
a1.sources.r1.deletePolicy = immediate
a1.sources.r1.inputCharset = ISO8859-1
a1.sources.r1.deserializer.maxLineLength = 8192
a1.sources.r1.decodeErrorPolicy = IGNORE

# Describe the sink
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.batchSize = 10000
a1.sinks.k1.brokerList = stormmaster:9092,storm01:9092,storm02:9092,storm03:9092,storm04:9092
a1.sinks.k1.serializer.class = kafka.serializer.StringEncoder
a1.sinks.k1.requiredAcks = 0
a1.sinks.k1.producer.type = async
a1.sinks.k1.topic = sighttpnew

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 80000
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.keep-alive = 30

Flume-env.sh配置:

# Enviroment variables can be set here.
export JAVA_HOME=/usr/java/jdk1.7.0_80
export FLUME_HOME=/hadoop/apache-flume-1.6.0-bin
# Give Flume more memory and pre-allocate, enable remote monitoring via JMX
export JAVA_OPTS="-Xms1024m -Xmx2048m -Dcom.sun.management.jmxremote"
# Note that the Flume conf directory is always included in the classpath.
export FLUME_CLASSPATH="/hadoop/apache-flume-1.6.0-bin/lib"

Flume启动命令:

/hadoop/apache-flume-1.6.0-bin/bin/flume-ng agent -c /hadoop/apache-flume-1.6.0-bin/conf/ -f /hadoop/apache-flume-1.6.0-bin/conf/viewdata.conf -n producer –Dflume.root.logger=ERROR &

注意一定要给全Flume配置文件的路径,否则启动Flume不能正确加载Flume-env.sh的配置。

b)Kafka集群安装及配置

http://kafka.apache.org/下载kafka安装包:kafka_*.tgz,解压后,配置server.properties文件。

server.properties配置:

#本机在kafka集群中的id
broker.id=48
#服务端口
port=9092
#主机名
host.name=storm01
# The number of threads handling network requests
num.network.threads=3
# The number of threads doing disk I/O
num.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400
# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
#kafka数据存储位置(数据量大时,需要存储的目录大小也要充分)
log.dirs=/data1/kafka-logs
#默认topic创建partition的数量
num.partitions=1
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1
#kafka事件只有flash到硬盘才能被后续消费者消费,因此要配置flash时间参数,避免小数据量情况下数据刷新时间过久
log.flush.interval.messages=10000
log.flush.interval.ms=1000
# 数据在kafka中保存的时间,单位小时,超时的数据kafka会自动删除
log.retention.hours=48
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according 
to the retention policies
log.retention.check.interval.ms=300000
# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
log.cleaner.enable=false
# zookeeper集群配置
zookeeper.connect=master:2181,storm01:2181,storm02:2181,storm03:2181,storm04:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000
#是否能够删除topic的配置,默认false不能删除topic
delete.topic.enable=true

Kafka服务启动:jps命令可以看到kafka的进程名,说明kafka已经成功启动。

nohup kafka-server-start.sh /home/hadoop/kafka_2.9.1-0.8.2.1/config/server.properties &

创建topic:创建复制因子2,有24个partition的topic,创建多个partition的目的是增加并行性,复制因子的目的是数据安全冗余。

kafka-topics.sh --create --zookeeper master:2181,storm01:2181,storm02:2181,storm03:2181,storm04:2181 --replication-factor 2 --partitions 24 --topic sighttp

kafka数据存储方式:在kafka数据存储目录下,可以看到以每个-方式命名的文件夹,例如sighttp-19表示topic:sighttp,partition:19,如下图所示:

图片描述
图三

进入topic-partition目录,可以看到很多.index和.log结尾的文件。其中.log是数据文件,其中存储的是kafka缓存池中的数据,.index是索引文件,数据文件和索引文件成对出现,文件名为一串数字,标识了该文件中存储数据的起始序列号,如下:

图片描述
图四

kafka数据消费状态查询:消费者从kafka消费数据状态是记录在zookeeper中的,使用zkCli.sh命令可以查看,如下图查询了消费topic:sighttp,partition:0的状态,offset表明已经处理到49259227840行,如下图所示:

图片描述
图五

经验:通过消费到的行数与存储到的行数,可以判断数据处理程序的速度是否满足数据生成速度的需求。

kafka消费典型异常:

[2016-10-27 16:15:42,536] ERROR [Replica Manager on Broker 51]: Error when processing fetch request for partition [sighttp,3] offset 6535061966 from consumer with correlation id 0. Possible cause: Request for offset 6535061966 but we only have log segments in the range 6580106664 to 6797636149. (kafka.server.ReplicaManager)

异常原因:kafka中由于消息过期已经把序号是6535061966的消息删除了,目前kafka中只有范围是6580106664到6797636149的日志,但是消费者还要处理过期删除的消息,那就会出现此异常消息(通常是由于数据处理速度慢,无法满足数据生成速度的要求,导致消息积压,积压的消息到达kafka配置的过期时间,被kafka删除)。

c)Storm集群安装及配置

http://storm.apache.org/下载Storm安装包,建议使用Storm 0.10.0 released以上版本,因为最新版本修正了很多bug,特别是STORM-935的问题(拓扑启动后会占用大量系统资源,导致Topology运行不稳定)。

storm.yaml文件配置:

#zookeeper集群服务器配置
storm.zookeeper.servers:
    - "master"
    - "storm01"
    - "storm02"
    - "storm03"
    - "storm04"
#storm主节点
nimbus.host: "master"
#strom管理页面服务端口
ui.port: 8081
#storm从节点服务端口配置,默认6700-6703共4个端口,意味着每台服务器可以提供4个worker插槽,这里增加了6704和6705端口,即为单台服务器增加了2个worker插槽,worker数增加意味着storm集群可以提供更多的计算资源。
supervisor.slots.ports:
 - 6700
 - 6701
 - 6702
 - 6703
 - 6704
 - 6705
#状态信息存储位置,避免使用/tmp
storm.local.dir: "/home/hadoop/apache-storm-0.10.0/workdir"
#主节点的内存
nimbus.childopts: "-Xmx3072m"
#从节点的内存
supervisor.childopts: "-Xmx3072m"
#worker的内存,增加内存可以减少GC overload的问题
worker.childopts: "-Xmx3072m"
#默认为30,增加netty超时时长等参数,降低因Netty通信问题,造成worker不稳定
storm.messaging.netty.max_retries:60
#增加storm.messaging.netty.max_wait_ms设置,默认为1000
storm.messaging.netty.max_wait_ms:2000

启动服务:

  • 主节点:(启动主节点服务和管理页面)
    nohup storm nimbus &
    nohup storm ui &
  • 从节点:
    nohup storm supervisor &

Storm管理页面:

浏览器输入Storm UI所在服务器地址+8081端口号,打开Strom管理页面如下图:

图片描述
图六

从图六Cluster Summary中可以看出Storm集群共有4个Supervisor节点,因每台Supervisor提供6个slot(如果在storm.yaml配置文件中不配置supervisor.slots.ports属性,则每个Supervisor默认提供4个slot),因此共有4*6=24个slot,已使用22个,还有2个空闲。需要注意的是每个拓扑一旦发布,将长久占用slot,如果没有足够的slot,最新发布的拓扑只会占用空闲的slot,不会抢占其他已经被占用的slot资源;如果没有slot,将无法发布新的拓扑,此时需要挖潜Storm集群服务器,通过配置文件增加slot资源或增加新的服务器。

从图六Topology Summary中可以看出,集群上已经发布了7个Topology,每个Topology占用的worker资源,启动的executor线程数,具体资源占用多少是在Storm Topology开发程序中指定的。

d)Kafka+Storm+Hdfs+Hbase拓扑开发

我们使用Eclipse创建MAVEN工程,在pom.xml配置文件中添加Storm及Hdfs的相关依赖,本例是Storm从Kafka中消费数据,经过ETL处理后存储到Hdfs和Hbase中,因此需要添加Storm-Kafka、Storm-Hdfs、Storm-Hbase等依赖,注意依赖包版本要与集群一致。

抽取过程继承BaseRichBolt类:

public class splitBolt extends BaseRichBolt {
    private static final String TAB = ",";
    private OutputCollector collector;
    public void prepare(Map config,TopologyContext context,OutputCollector collector){
        this.collector=collector;
    }
    public void execute(Tuple input){
            String line=input.getString(0);
            String[] words=line.split(TAB);
            if (words.length>74)
            {
                String Account;
                if (words[0].length()>0) Account=words[0];
                else Account="NULL";
                String LocalIPv4;
                if (words[1].length()>0) LocalIPv4=words[1];
                else LocalIPv4="NULL";
                String RemoteIPv4;
                if (words[2].length()>0) RemoteIPv4=words[2];
                else RemoteIPv4="NULL";
                String newline=Account+"|"+LocalIPv4+"|"+RemoteIPv4;
                collector.emit(input,new Values(newline));
            }
            collector.ack(input);
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer){
        declarer.declare(new Fields("newline"));
    }
}

写Hbase需要实现HBaseMapper类:

public class myHbaseMapper implements HBaseMapper {
    public ColumnList columns(Tuple tuple) {
        String line=tuple.getString(0);
        String[] words=line.split("\\|");
        ColumnList cols = new ColumnList();
        //参数依次是列族名,列名,值
        if (words[1].length()>0) cols.addColumn("content".getBytes(), "LocalIPv4".getBytes(), words[1].getBytes());
        if (words[2].length()>0) cols.addColumn("content".getBytes(), "RemoteIPv4".getBytes(), words[2].getBytes());
        return cols;
    }
    public byte[] rowKey(Tuple tuple) {
        String line=tuple.getString(0);
        String[] words=line.split("\\|");
        String key;
        //rowkey设置成Account的反字符串,便于hbase表内分区的数据均衡
        key=new StringBuilder(words[0]).reverse().toString();
        return key.getBytes();
    }
}

main函数:

public static void main(String[] args)
{  
    String zks = "master:2181,storm01:2181,storm02:2181 "; //zookeeper集群
    String topic = "topicname"; //kafka中topic名称
    String zkRoot = "/storm";//zookeeper中存储状态信息的根目录
    String id = "kafkatopicname";//zookeeper中存储本拓扑状态信息的子目录
    FileNameFormat fileNameFormat = new DefaultFileNameFormat()
    .withPath("/storm/tmp/").withPrefix("tmp_").withExtension(".dat");
    RecordFormat format = new DelimitedRecordFormat()
    .withFieldDelimiter("|"); //写到hdfs的目录文件名以’tmp_’开头,’.dat’结尾
    //每10分钟重写一个hdfs的新文件
    FileRotationPolicy rotationPolicy = new TimedRotationPolicy(10.0f, TimeUnit.MINUTES);
    BrokerHosts brokerHosts = new ZkHosts(zks);
    //配置storm拓扑的spout
    SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id);
    spoutConf.scheme = new SchemeAsMultiScheme(new MessageScheme());  
    spoutConf.zkServers = Arrays.asList(new String[] {"master", "storm01","storm02"});  
    spoutConf.zkPort = 2181;
    spoutConf.ignoreZkOffsets = false;//重启拓扑时,需要从zookeeper中读取偏移量
    //如果偏移量中的数据已经从kafka中删除,则从kafka中保存的最早数据开始处理。
    spoutConf.startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
    spoutConf.useStartOffsetTimeIfOffsetOutOfRange = true;
    //配置hdfs bolt
    HdfsBolt hdfsBolt = new HdfsBolt()
    .withFsUrl("hdfs://hdfsmaster:9000")
    .withFileNameFormat(fileNameFormat)
    .withRecordFormat(format)
    .withRotationPolicy(rotationPolicy)
    //hdfs数据文件写完后,move到新目录
    .addRotationAction(new MoveFileAction().toDestination("/storm/http/")); 
    //实例化HBaseMapper
    HBaseMapper mapper = new myHbaseMapper();
    //实例化HBaseBolt,指定hbase中的表名
    HBaseBolt hBolt = new HBaseBolt("hbasetable", mapper).withConfigKey("hbase.conf");
    TopologyBuilder builder = new TopologyBuilder();  
    //配置spout线程数为24,此数要与kafka中topic的partition数一致,partition数越多,则spout读取数据的并行性越高,处理速度越快
    builder.setSpout("kafka-reader", new KafkaSpout(spoutConf),24);
    //配置bolt,此bolt开发处理逻辑,bolt可以串接多个
    builder.setBolt("etl", new splitBolt(), 24).shuffleGrouping("kafka-reader");  
    builder.setBolt("hdfs-bolt", hdfsBolt, 24).shuffleGrouping("etl");
    builder.setBolt("hbase-bolt", hBolt, 24).shuffleGrouping("etl");
    Config conf = new Config();
    //增加hbase配置,指定hbase在hdfs集群上的目录,zookeeper服务器集群
    Map<String, Object> hbConf = new HashMap<String, Object>();
    hbConf.put("hbase.rootdir", "hdfs://hdfsmaster:9000/hbase");
    hbConf.put("hbase.zookeeper.quorum","master,storm01,storm02");
    conf.put("hbase.conf", hbConf);
    String name = sighttphdfs.class.getSimpleName();   
    if (args != null && args.length > 0) {  
        conf.put(Config.NIMBUS_HOST, args[0]);
        conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, 0); 
        //设置拓扑占用worker数为4,根据实时处理数据量大小按需配置
        conf.setNumWorkers(4); 
        StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());  
    }
}

上面程序实现了Storm读Kafka写Hdfs和Hbase的例子,抽取类中可以根据不同的业务需求,通过Java代码实现不同的逻辑。编译后的jar包上传到集群,使用storm命令行提交Topology:

storm jar ./kafkastream.jar sighdfs.sighttphdfs stormmaster

总结

经过几个月的实际运行,我们的大数据实时处理架构能够始终保持稳定,话单处理速度高于话单生成速度,有效的支撑了运营商大数据的各种分析查询需求。开发和优化过程充满挑战,经过各种研究和尝试,问题逐渐解决,在此我们也积累了大量的开发和优化经验。

最后再分享2个我们实际遇到的问题:

  • Zookeeper配置造成Storm拓扑运行不稳定

因Storm集群需要Zookeeper集群作状态同步,因此所有是Storm服务器worker进程都会不停连接Zookeeper节点,Zookeeper节点的默认连接数是60,当Storm计算拓扑数量较多时,需要修改Zookeeper配置maxClientCnxns=1000,增加Zookeeper连接数。

  • Hdfs节点磁盘I/O高造成Storm拓扑运行不稳定

由于Storm是实时计算,每个环节的拥塞都将引起Storm拓扑的不稳定,在开发中我们遇到Hdfs某个节点磁盘I/O高,导致Storm写Hdfs超时,最终引发Supervisor杀掉worker,造成拓扑不稳定的问题。究其原因是在某个Hdfs节点上,Yarn任务正在进行Reduce操作,用iostat -x 1 10命令查看,Yarn的中间盘I/O长时间被100%占用,同时Yarn的中间盘也是Hdfs的数据盘,导致写入请求无法响应,最终导致Storm写Hdfs的worker超时,引发拓扑运行不稳定。此处建议配置Yarn的中间盘时,不要使用操作系统根盘,不要使用Hdfs的数据盘,可以有效避免Storm写Hdfs超时的问题。

SDCC 2017•上海站将于2017年3月17-19日登陆申城,三大技术峰会和24位嘉宾,汇聚国内知名的互联网公司CTO、架构师、技术总监,畅谈运维、数据库和架构的热门话题和技术热点,遇见精益运维发起人&优维科技CEO王津银、MongoDB大中华区首席架构师唐建法等大牛,五人团购报名立减1500元,详情点击注册参会


更多精彩,欢迎关注CSDN大数据公众号!

图片描述

评论