大数据实时处理实战

a1.sources.r1.ignorePattern = ^(.)*\\.tmp$

a1.sources.r1.decodeErrorPolicy = IGNORE

export JAVA_OPTS="-Xms1024m -Xmx2048m -Dcom.sun.management.jmxremote"

/hadoop/apache-flume-1.6.0-bin/bin/flume-ng agent -c /hadoop/apache-flume-1.6.0-bin/conf/

a1.channels.c1.transactionCapacity = 2000a1.sinks.k1.batchSize = 2000

producer.sources.s.deserializer.maxLineLength=65535

a1.sources.r1.inputCharset = ISO8859-1

producer.sources.s.decodeErrorPolicy=IGNORE

a1.sources.r1.deletePolicy = immediate

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

# 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"

/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 &

#本机在kafka集群中的idbroker.id=48#服务端口port=9092#主机名host.name=storm01# The number of threads handling network requestsnum.network.threads=3# The number of threads doing disk I/Onum.io.threads=8# The send buffer (SO_SNDBUF) used by the socket serversocket.send.buffer.bytes=102400# The receive buffer (SO_RCVBUF) used by the socket serversocket.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=10000log.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 policieslog.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 zookeeperzookeeper.connection.timeout.ms=6000#是否能够删除topic的配置，默认false不能删除topicdelete.topic.enable=true

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

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

[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)

#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

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"));
    }
}

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();
    }
}

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 jar ./kafkastream.jar sighdfs.sighttphdfs stormmaster