1、启动kafkabin/kafka-server-start.sh  config/server.properties > /dev/null 2>&1 &2、列出所有的topicsbin/kafka-topics.sh --list --zookeeper zk-datanode-01:2181,zk-datanode-02:2181,zk-datanode-03:21813、创建topicsbin/kafka-topics.sh --create --topic apache-logs --zookeeper zk-datanode-01:2181,zk-datanode-02:2181,zk-datanode-03:2181 --replication-factor 1 --partitions 5bin/kafka-topics.sh --create --topic apache-error-logs --zookeeper zk-datanode-01:2181,zk-datanode-02:2181,zk-datanode-03:2181 --replication-factor 1 --partitions 5bin/kafka-topics.sh --create --topic apache-slow-logs --zookeeper zk-datanode-01:2181,zk-datanode-02:2181,zk-datanode-03:2181 --replication-factor 1 --partitions 5

$ git clone https://github.com/kiritbasu/Fake-Apache-Log-Generator.git
$ cd Fake-Apache-Log-Generator

 elapsed_us = random.randint(1 * 1000,1000 * 1000) # 1 ms to 1 secseconds=random.randint(30,300)increment = datetime.timedelta(seconds=seconds)otime += incrementip = faker.ipv4()dt = otime.strftime('%d/%b/%Y:%H:%M:%S')tz = datetime.datetime.now(pytz.timezone('US/Pacific')).strftime('%z')vrb = numpy.random.choice(verb,p=[0.6,0.1,0.1,0.2])uri = random.choice(resources)if uri.find("apps")>0:uri += `random.randint(1000,10000)`resp = numpy.random.choice(response,p=[0.9,0.04,0.02,0.04])byt = int(random.gauss(5000,50))referer = faker.uri()useragent = numpy.random.choice(ualist,p=[0.5,0.3,0.1,0.05,0.05] )()f.write('%s - - [%s %s] %s "%s %s HTTP/1.0" %s %s "%s" "%s"\n' % (ip,dt,tz,elapsed_us,vrb,uri,resp,byt,referer,useragent))log_lines = log_lines - 1flag = False if log_lines == 0 else True

import sys
import apache_log_parser
import jsonauto_incr_id = 1
parser_format = '%a - - %t %D "%r" %s %b "%{Referer}i" "%{User-Agent}i"'
line_parser = apache_log_parser.make_parser(parser_format)
while True:# we'll use pipeline = sys.stdin.readline()if not line:breakparsed_dict = line_parser(line)parsed_dict['id'] = auto_incr_idauto_incr_id += 1# works only python 2, but I don't care cause it's just a test module :)parsed_dict = {k.upper(): v for k, v in parsed_dict.iteritems() if not k.endswith('datetimeobj')}print json.dumps(parsed_dict)

python apache-fake-log-gen.py -n 0 | python parse-fake-log-gen-to-json-with-incrementing-id.py | ../kafka/bin/kafka-console-producer.sh --broker-list 192.168.46.160:9092 --topic apache-logs

bin/kafka-console-consumer.sh --zookeeper zk-datanode-01:2181,zk-datanode-02:2181,zk-datanode-03:2181 --topic apache-logs

${storm_home}/bin/storm sql apache_log_error_filtering.sql apache_log_error_filtering --artifacts "org.apache.storm:storm-sql-kafka:1.1.1,org.apache.storm:storm-kafka:1.1.1,org.apache.kafka:kafka_2.10:0.8.2.2^org.slf4j:slf4j-log4j12,org.apache.kafka:kafka-clients:0.8.2.2"${storm_home}/bin/storm sql apache_log_slow_filtering.sql apache_log_slow_filtering --artifacts "org.apache.storm:storm-sql-kafka:1.1.1,org.apache.storm:storm-kafka:1.1.1,org.apache.kafka:kafka_2.10:0.8.2.2^org.slf4j:slf4j-log4j12,org.apache.kafka:kafka-clients:0.8.2.2" --jars "UDFTest-0.0.1-SNAPSHOT.jar"