Streaming实战

矫情吗；* 2023-06-08 03:30 24阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70][]

![20191023151459748.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 3][]

### **用Python日志产生器开发实之产生访问Url和Ip信息，时间，http请求头，查询内容：** ###

根据主站的url和ip地址可以看到页面的信息，然后根据实际情况来定时产生ip和url

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 4][]

通过定时调度工具每一分钟产生一批数据

Linux crontab

网站 ：[http://tool.lu/crontab][http_tool.lu_crontab]

每一分钟执行一次的crontab表达式 ： \*/1 \* \* \* \*

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 5][]

在linux中只写写好的 python代码：python2代码

def sample_ip():
    
            slice = random.sample(ip_slices , 4)
    
            return ".".join([str(item) for item in slice])
    
    
    
    def sample_referer():
    
            if random.uniform(0, 1) > 0.2:
    
                    return "-"
    
    
    
            refer_str = random.sample(http_referers, 1)
    
            query_str = random.sample(search_keyword, 1)
    
            return refer_str[0].format(query=query_str[0])
    
    
    
    def sample_status_code():
    
            return random.sample(status_codes, 1)[0]
    
    
    
    def generate_log(count = 10):
    
            time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    
    
    
            f = open("/home/hadoop/data/project/logs/access.log","w+")
    
    
    
            while count >= 1:
    
                    query_log = "{ip}\t{local_time}\t\"GET /{url} HTTP/1.1\"\t{status_code}\t{referer}".format(url=sample_url(), ip=sample_ip(), referer=sample_referer()
    
    , status_code=sample_status_code(),local_time=time_str)
    
                    f.write(query_log + "\n")
    
    
    
                    count = count - 1
    
    
    
    if __name__ == '__main__':
    
            generate_log(100)

创建log\_generator.sh脚本添加执行内容：

![20191023151710770.png][]

修改脚本权限：

![20191023151727263.png][]

利用 crontab  -e写进执行脚本：

![20191023151742398.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 6][]

当日志产生后：

![20191023151814710.png][]

对接python日志产生输出的日志到flume
    Streaming_project.conf
    选型：access.log ==> 控制台输出
    
    
    
    Streaming_project.conf:
    
    exec-memory-logger.sources = exec-source
    exec-memory-logger.sinks = logger-sink
    exec-memory-logger.channels = memory-channel
    
    exec-memory-logger.sources.exec-source.type = exec
    exec-memory-logger.sources.exec-source.command = tail -F /home/hadoop/data/project/logs/access.log
    exec-memory-logger.sources.exec-source.shell = /bin/sh -c
    
    exec-memory-logger.channels.memory-channel.type = memory
    
    exec-memory-logger.sinks.logger-sink.type = logger
    
    exec-memory-logger.sources.exec-source.channels = memory-channel
    exec-memory-logger.sinks.logger-sink.channel = memory-channel
    
    
    启动flume:(注意：这里需要等号后面不应该有空格，编程习惯会习惯性带空格)
    flume-ng agent --name exec-memory-logger --conf $FLUME_HOME/conf --conf-file /home/hadoop/data/project/streaming_project.conf --Dflume.root.logger=INFO,console

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 7][]

日志==》flume==>kafka

*  启动zk:./zkServer.sh start
 *  启动kafka Server :
 *  修改flume配置文件使得flume sink数据到kafka

Kafka的server.properties:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 8][]

kafka时先启动 zookeeper

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 9][]

启动zookeeper

![20191023152033547.png][]

然后启动 kafka：

./kafka-server-start.sh -daemon /home/hadoop/app/kafka\_2.11-0.9.0.0/config/server.properties

![20191023152047531.png][]

Streaming\_project2.conf:

exec-memory-kafka.sources = exec-source
    exec-memory-kafka.sinks = kafka-sink
    exec-memory-kafka.channels = memory-channel
    
    exec-memory-kafka.sources.exec-source.type = exec
    exec-memory-kafka.sources.exec-source.command = tail -F /home/hadoop/data/project/logs/access.log
    exec-memory-kafka.sources.exec-source.shell = /bin/sh -c
    
    exec-memory-kafka.channels.memory-channel.type = memory
    
    exec-memory-kafka.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
    exec-memory-kafka.sinks.kafka-sink.brokerList = hadoop000:9092
    exec-memory-kafka.sinks.kafka-sink.topic = streamingtopic
    exec-memory-kafka.sinks.kafka-sink.batchSize = 5
    exec-memory-kafka.sinks.kafka-sink.requiredAcks = 1
    
    exec-memory-kafka.sources.exec-source.channels = memory-channel
    exec-memory-kafka.sinks.kafka-sink.channel = memory-channel

**启动flume:**

flume-ng agent --name exec-memory-kafka --conf $FLUME_HOME/conf --conf-file /home/hadoop/data/project/streaming_project2.conf -Dflume.root.logger=INFO,console

**streaming\_project2.conf：**

exec-memory-kafka.sources = exec-source
    exec-memory-kafka.sinks = kafka-sink
    exec-memory-kafka.channels = memory-channel
    
    exec-memory-kafka.sources.exec-source.type = exec
    exec-memory-kafka.sources.exec-source.command = tail -F /home/hadoop/data/project/logs/access.log
    exec-memory-kafka.sources.exec-source.shell = /bin/sh -c
    
    exec-memory-kafka.channels.memory-channel.type = memory
    
    exec-memory-kafka.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
    exec-memory-kafka.sinks.kafka-sink.brokerList = hadoop000:9092
    exec-memory-kafka.sinks.kafka-sink.topic = streamingtopic1
    exec-memory-kafka.sinks.kafka-sink.batchSize = 5
    exec-memory-kafka.sinks.kafka-sink.requiredAcks = 1
    
    exec-memory-kafka.sources.exec-source.channels = memory-channel
    exec-memory-kafka.sinks.kafka-sink.channel = memory-channel

启动flume后需要有一个客户端能够消费我们的kafka：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 10][]

启动kafka消费：

kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic streamingtopic1

启动后flume接受到的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 11][]

kafaka消费者消费数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 12][]

使用spark Streaming对接kafka的数据：

配置参数：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 13][]

数据处理过程未完待续！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70]: /images/20230531/765aa64d95a6485fa55e824c325e6694.png
[20191023151459748.png]: /images/20230531/3552c629fe814696b567e458e68bb19e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 1]: /images/20230531/f1d08dec2add470b8df7db14ef68f93e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 2]: /images/20230531/88139e802c734f21aa37d0a125728b0c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 3]: /images/20230531/e08d37e460aa42c6a2f902f925f01b21.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 4]: /images/20230531/f88b4c877f8f4d639bb6e55430696c17.png
[http_tool.lu_crontab]: http://tool.lu/crontab
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 5]: /images/20230531/950af0f0c2a04a9798cfc24fabb7c295.png
[20191023151710770.png]: /images/20230531/0b009f047ef445cda36361700adc1379.png
[20191023151727263.png]: /images/20230531/6efd3089fadb4704b61f089ea94bec8e.png
[20191023151742398.png]: /images/20230531/db6d4bfe604443d6a65106b36b1b8bd5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 6]: /images/20230531/d7712b4c13cc476cb05f07597b90084c.png
[20191023151814710.png]: /images/20230531/a74a3849ad7646089e5a9fc98b8ee6a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 7]: /images/20230531/34888e83900f40f58e82f776a787b288.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 8]: /images/20230531/9cd53adcea584d238b49e94d0bf6c9b1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 9]: /images/20230531/3e55e422adcd46a5a40e36e853a93951.png
[20191023152033547.png]: /images/20230531/6af35ba80c8646029765395de4a61605.png
[20191023152047531.png]: /images/20230531/4a4f4d6798894657b454bd944e1a46a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 10]: /images/20230531/f1468d00768542e381e8d7378690dade.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 11]: /images/20230531/f44f62531b6148d8acd8213c9a2a7564.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 12]: /images/20230531/3ce2a036d9f8458fa167091a8acd82fb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNDc5NDY0_size_16_color_FFFFFF_t_70 13]: /images/20230531/68048fea7b5e457491f47907ea5a5567.png