Over a million developers have joined DZone.

Write a Kafka Producer Using Twitter Stream

With the newly open sourced Twitter HBC, a Java HTTP library for consuming Twitter’s Streaming API, we can easily create a Kafka twitter stream producer.

· Integration Zone

Learn how API management supports better integration in Achieving Enterprise Agility with Microservices and API Management, brought to you in partnership with 3scale

Twitter open sourced its Hosebird client (hbc), a robust Java HTTP library for consuming Twitter’s Streaming API. In this post, I am going to present a demo of how we can use hbc to create a Kafka twitter stream producer, which tracks a few terms in Twitter statuses  and produces a Kafka stream out of it, which can be utilized later for counting the terms, or sending that data from Kafka to Storm (Kafka-Storm pipeline) or HDFS ( as we will see in next post about using Camus API).

You can download and run complete Sample here.

Requirements

  • Apache Kafka 0.8
  • Twitter Developer account ( for API Key, Secret etc.)
  • Apache Zookeeper ( required for Kafka)
  • Oracle JDK 1.7 (64 bit )

Build Environment

  • Eclipse
  • Apache Maven 2/3

How to Generate Twitter API Keys Using Developer Account

  1. Go to https://dev.twitter.com/apps/new and log in, if necessary.
  2. Enter your Application Name, Description and your website address. You can leave the callback URL empty.
  3. Accept the TOS.
  4. Submit the form by clicking the Create your Twitter Application.
  5. Copy the consumer key (API key) and consumer secret from the screen into your application.
  6. After creating your Twitter Application, you have to give the access to your Twitter Account to use this Application. To do this, click the Create my Access Token.
  7. Now you will have Consumer Key, Consumer Secret, Acess token, Access Token Secret to be used in streaming API calls.

Steps to Run the Sample

  • Start Zookeeper server in Kafka using following script in your Kafka installation folder  –
  • $bin/zookeeper-server-start.sh config/zookeeper.properties &

    and, verify if it is running on default port 2181 using –

    $netstat -anlp | grep 2181

    2. Start Kafka server using following script –

    $bin/kafka-server-start.sh config/server.properties  &

    and, verify if it is running on default port 9092

    3. Now, when we are all set with Kafka running and ready to accept messages on any dynamically created topic ( default setting), we will create a Kafka Producer, which makes use of hbc client API to get twitter stream for tracking terms and puts on topic named as “twitter-topic” .

  • First, we need to give maven dependencies for hbc-core for latest version and some other dependencies needed for Kafka –ssss
  • <dependency>
    <groupId>com.twitter</groupId>
    <artifactId>hbc-core</artifactId> <!-- or hbc-twitter4j -->
    <version>2.2.0</version> <!-- or whatever the latest version is -->
    </dependency>
    •  Then, we need to set properties to configure our Kafka Producer to publish messages to topic –
      private static final String topic = "twitter-topic";
    • Properties properties = new Properties();
              properties.put("metadata.broker.list", "localhost:9092");
              properties.put("serializer.class", "kafka.serializer.StringEncoder");
    • Set up a StatusFilterEndpoint , which will setup track terms to be tracked on recent status messages, as in the example, twitterapi and #AAPSweep ( change these to term you want to track) –
    • BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10000);
              StatusesFilterEndpoint endpoint = new StatusesFilterEndpoint();
              // add some track terms
              endpoint.trackTerms(Lists.newArrayList("twitterapi",
                      "#AAPSweep"));
    • Provide authentication parameters for OAuth ( we are getting them using commandline parameters for this program ) for using twitter that we generated earlier and create the client using endpoint and auth –
    Authentication auth = new OAuth1(consumerKey, consumerSecret, token,
                    secret);
            // Authentication auth = new BasicAuth(username, password);
            // Create a new BasicClient. By default gzip is enabled.
            Client client = new ClientBuilder().hosts(Constants.STREAM_HOST)
                    .endpoint(endpoint).authentication(auth)
                    .processor(new StringDelimitedProcessor(queue)).build();
  • Last step, connect to client, fetch messages from queue and send through Kafka Producer –
  • // Establish a connection
            client.connect();
            // Do whatever needs to be done with messages
            for (int msgRead = 0; msgRead < 1000; msgRead++) {
                KeyedMessage<String, String> message = null;
                try {
                    message = new KeyedMessage<String, String>(topic, queue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                producer.send(message);
            }
            producer.close();
            client.stop();

    To run the complete example run TwitterKafkaProducer.java class as a Java Application in  your favourite IDE.

    Also, to see how you can integrate Kafka with HDFS using camus from LinkedIn, you can visit the blog here.

    Unleash the power of your APIs with future-proof API management - Create your account and start your free trial today, brought to you in partnership with 3scale.

    Topics:
    kafka ,zookeeper ,hadoop ,big data ,twitter

    Published at DZone with permission of Saurabh Chhajed, DZone MVB. See the original article here.

    Opinions expressed by DZone contributors are their own.

    The best of DZone straight to your inbox.

    SEE AN EXAMPLE
    Please provide a valid email address.

    Thanks for subscribing!

    Awesome! Check your inbox to verify your email so you can start receiving the latest in tech news and resources.
    Subscribe

    {{ parent.title || parent.header.title}}

    {{ parent.tldr }}

    {{ parent.urlSource.name }}