Kafka系列四 flume-kafka-storm整合

flume-kafka-storm

flume读取日志数据，然后发送至kafka。

1、flume配置文件

agent.sources = kafkaSource
agent.channels = kafkaChannel
agent.sinks = kafkaSink

agent.sources.kafkaSource.type = exec
agent.sources.kafkaSource.command = tail -F /home/hadoop/kafkaData/kafka.log
agent.sources.kafkaSource.channels = kafkaChannel

agent.sinks.kafkaSink.channel = kafkaChannel
agent.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkaSink.topic = stormTopic
agent.sinks.kafkaSink.brokerList = 192.168.25.151:9092,192.168.25.152:9092,192.168.25.153:9092
agent.sinks.kafkaSink.kafka.flumeBatchSize = 20
agent.sinks.kafkaSink.kafka.producer.acks = 1
agent.sinks.kafkaSink.kafka.producer.linger.ms = 1
agent.sinks.kafkaSink.kafka.producer.compression.type = snappy
agent.sinks.kafkaSink.serializer.class=kafka.serializer.StringEncoder 

agent.channels.kafkaChannel.type=memory                                                                                                       
agent.channels.kafkaChannel.capacity=10000                                                                                                    
agent.channels.kafkaChannel.transactionCapacity=100

2、启动flume

bin/flume-ng agent --conf-file  conf/flume-kafka.conf -c conf/ --name agent -Dflume.root.logger=DEBUG,console

3、需要在flume机器上修改hosts文件，添加上kafka的主机名和ip的映射。

4、在kafka上创建主题

bin/kafka-topics.sh --create --zookeeper hadoop2:2181 --replication-factor 1 --partitions 3 --topic stormTopic

5、模拟生成日志脚本

for((i=0;i<=10000;i++));
do echo "kafka_test-"$i>>/home/hadoop/kafkaData/kafka.log;
done

6、在kafka上开启消费者

bin/kafka-console-consumer.sh --zookeeper hadoop2:2181 --from-beginning --topic stormTopic

至此，flum->kafka的数据流走通。

7、整合Storm，将kafka作为stom的spout，将使用KafkaSpout。

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.itcast</groupId>
    <artifactId>kafkaStorm</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>
    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.storm/storm-core -->
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.2.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-kafka</artifactId>
            <version>1.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.12</artifactId>
            <version>1.0.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>1.0.0</version>
        </dependency>


    </dependencies>
    <build>
        <plugins>
            <!-- 资源文件拷贝插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-resources-plugin</artifactId>
                <version>2.7</version>
                <configuration>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <!-- java编译插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.2</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>2.4</version>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.4</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            <mainClass>cn.itcast.kafka.Kafka2Storm</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

package cn.itcast.kafka;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.BrokerHosts;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.ZkHosts;
import org.apache.storm.topology.TopologyBuilder;

public class Kafka2Storm {
    public static void main(String[] args)
            throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        BrokerHosts hosts = new ZkHosts("192.168.25.142:2181,192.168.25.143:2181,192.168.25.144:2181");
        /**
         * hosts:用以获取Kafka broker和partition的信息,在zk上获取，此处填写zk的地址
         * topic:从哪个topic读取消息 zkRoot:进度信息记录于zookeeper的哪个路径下
         * id:进度记录的id，想要一个新的Spout读取之前的记录，应把它的id设为跟之前的一样
         */
        SpoutConfig spoutConfig = new SpoutConfig(hosts, "stormTopic", "/mykafka", "kafkaSpout");
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
        topologyBuilder.setSpout("kafkaSpout", kafkaSpout);
        // 将一行行的文本切分成单词
        topologyBuilder.setBolt("valueBolt", new ValueBolt(), 1).shuffleGrouping("kafkaSpout");
        // 启动topology的配置信息
        Config config = new Config();
        // 定义集群分配多少个工作进程来执行这个topology
        config.setNumWorkers(3);
        
         LocalCluster localCluster = new LocalCluster();
         localCluster.submitTopology("kafkaStomrTopology", config,
         topologyBuilder.createTopology());
        // 集群模式提交topology
//        StormSubmitter.submitTopologyWithProgressBar("kafkaStomrTopology", config, topologyBuilder.createTopology());
    }
}

package cn.itcast.kafka;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ValueBolt extends BaseRichBolt {
    Logger logger = LoggerFactory.getLogger(ValueBolt.class);

    /**
     * 
     */
    private static final long serialVersionUID = 1L;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        // TODO Auto-generated method stub

    }

    @Override
    public void execute(Tuple input) {
        logger.info(new String((byte[]) input.getValue(0)));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // TODO Auto-generated method stub

    }

}