Page History

...

Now that we have the sensor set up and generating logs we need to figure out how to pipe these logs to a Kafka topic.  To do so the first thing we need to do is setup a new Kafka topic for Squid.

cd /usr/hdp/current/kafka-broker/bin/./kafka-topics.sh --zookeeper localhost:2181 --create --topic squid --partitions 1 --replication-factor 1

./usr/hdp/current/kafka-broker/bin//kafka-topics.sh --zookeeper localhost:2181 --list

...

name: "squid"

config:

    topology.workers: 1

 

 

components:

    -   id: "parser"

        className: "org.apache.metron.parsing.parsers.GrokParser"

        constructorArgs:

            - "/apps/metron/patterns/squid"

            - "SQUID_DELIMITED"

        configMethods:

            -   name: "withTimestampField"

                args:

                    - "timestamp"

    -   id: "writer"

        className: "org.apache.metron.parsers.writer.KafkaWriter"

        constructorArgs:

            - "${kafka.broker}"

    -   id: "zkHosts"

        className: "storm.kafka.ZkHosts"

        constructorArgs:

            - "${kafka.zk}"

    -   id: "kafkaConfig"

        className: "storm.kafka.SpoutConfig"

        constructorArgs:

            # zookeeper hosts

            - ref: "zkHosts"

            # topic name

            - "squid"

            # zk root

            - ""

            # id

            - "squid"

        properties:

            -   name: "ignoreZkOffsets"

                value: falsetrue

            -   name: "startOffsetTime"

                value: -1

            -   name: "socketTimeoutMs"

                value: 1000000

 

spouts:

    -   id: "kafkaSpout"

        className: "storm.kafka.KafkaSpout"

        constructorArgs:

            - ref: "kafkaConfig"

 

bolts:

    -   id: "parserBolt"

        className: "org.apache.metron.parsers.bolt.ParserBolt"

        constructorArgs:

            - "${kafka.zk}"

            - "squid"

            - ref: "parser"

            - ref: "writer"

 

streams:

    -   name: "spout -> bolt"

        from: "kafkaSpout"

        to: "parserBolt"

        grouping:

            type: SHUFFLE

...