Hello world Spark example

In this article, there is 3 hello world level demos.  The difference is where does the spark get its input data.

• The first one, spark get its input data from external socket server. This method is often used in testing or developing phase.
• The second demo, spark get its input data from Flume agent.
• The third one, spark get its input data from Kafka. The last one may be used more common in real project.

1. Brief

Flume,Spark, Kafka are all apache projects.

Flume is mainly designed to move data, large amount of data, from one place to another, especially from Non-Big-Data-World to Big-Data-World.  e.g. from log file to HDFS. Usually the role of Flume agent in the cluster is data producer.

Spark is designed to process large scale data in a near-real-time way.  Input data will be process in consecutive batch of operations, usually these operation are map-reduce. If the batch interval set to 10 seconds, a map-reduce operation can process the data coming at that time windows every 10 seconds.  An working scenario is using spark 's input is infinite stream of lines of web access log.  Usually the role of Spark in the cluseter is data consumer.

Kafka, like a JMS in traditional non big data application, can be used to connect Flume, namely producer, to Spark, namely consumer.

2. Spark demos

All three spark demo just get input from stream and print out to screen. Every demo is very simple, only has one main class. You can replace the print out with some real task in your own use. In this article all demos run on CDH release 5.4.9. (flume 1.5.0, spark 1.3.0, kafka 2.0 )

Pay attention to how they get input stream as well as how to process the stream. The generic types of the stream are different, which are String, FlumeEvent and Tuple2<String,String> for demo1, 2, 3 respectively.

2.1 Data from external socket server.

In demo1, Spark gets its input from a external socket server.  Add dependency to  spark-streaming_xxx to you pom.

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.10</artifactId>
      <version>1.3.0</version>
    </dependency>

The main class is defined like below.

package com.shengwang.demo;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;


public class NakedSparkMain {
  private static final int INTERVAL = 5000;
  
  private static final String DATA_HOST = "localhost";
  private static final int DATA_PORT = 11111;
  
  public static void main(String[] args) {
    NakedSparkMain main = new NakedSparkMain();
    main.createJavaStreamingDStream();
  }
  
  private void createJavaStreamingDStream() {
    // =========================================
    // init Streaming context
    // =========================================
    SparkConf conf = new SparkConf().setAppName("naked-spark application");
    JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(INTERVAL));

    // =========================================
    // Data input - get data from socket server
    // =========================================
    JavaReceiverInputDStream<String> stringStream = ssc.socketTextStream(DATA_HOST, DATA_PORT); 
    
    // =========================================
    // Data process part - print every line 
    // =========================================
    stringStream.foreach(x -> {
      x.foreach(y -> {
        System.out.println("----- message -----");
        System.out.println("input = " + y);
      });
      return null;
    });
    
    ssc.start();
    ssc.awaitTermination();
  }
}

2. 2 Data from Flume sink

In demo2, Spark gets its input from a flume avro agent.  Add dependency to  spark-streaming-flume_xxx to you pom.

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-flume_2.10</artifactId>
      <version>1.3.0</version>
    </dependency>

The main class is defined like below.

package com.sanss.qos.spark.stream.flume;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.flume.FlumeUtils;
import org.apache.spark.streaming.flume.SparkFlumeEvent;

public class FlumeMain {
  private static final int INTERVAL = 5000;
  
  private static final String FLUME_HOST = "localhost";
  private static final int FLUME_PORT = 11111;
  
  public static void main(String[] args) {
    FlumeMain main = new FlumeMain();
    main.createJavaStreamingDStream();
  }
  
  private void createJavaStreamingDStream() {
    // =========================================
    // init Streaming context
    // =========================================
    SparkConf conf = new SparkConf().setAppName("flume-to-spark application");
    JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(INTERVAL));
    
    // =========================================
    // Data input - get data from flume agent
    // =========================================
    JavaReceiverInputDStream<SparkFlumeEvent> flumeStream = FlumeUtils.createStream(ssc, FLUME_HOST, FLUME_PORT); 
    
    // =========================================
    // Data process part - print every line 
    // =========================================
    flumeStream.foreach(x -> {
      x.foreach(y -> {
        System.out.println("----- message -----");
        System.out.println("input = " + getMessageFromFlumeEvent(y));
      });
      return null;
    });
    
    ssc.start();
    ssc.awaitTermination();
  }

  private String getMessageFromFlumeEvent(SparkFlumeEvent x) {
    return new String (x.event().getBody().array());
  }
}

This demo hardcode the flume sink ip and port to localhost and 11111, make sure the flume configuration match the code.

2.3 Data from Kafka

In demo3, Spark gets its input from kafka topic.  Add dependency to  spark-streaming-kafka_xxx to you pom.

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-kafka_2.10</artifactId>
      <version>1.3.0</version>
    </dependency>

The main class is defined like below.

package com.sanss.qos.spark.stream.flume;

import java.util.HashMap;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

public class KafkaSparkMain {
  private static final int INTERVAL = 5000;
  
  private String zookeeper = "cdh01:2181";   // hardcode zookeeper
  private String groupId = "qosTopic";
  private String topicName = "mytopic1";
  private Map<String, Integer> topics = new HashMap<>();
  
  public static void main(String[] args) {
    KafkaSparkMain main = new KafkaSparkMain();
    main.createJavaStreamingDStream();
  }
  
  private void createJavaStreamingDStream() {
    // =========================================
    // init Streaming context
    // =========================================
    SparkConf conf = new SparkConf().setAppName("kafka-to-spark application");
    JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(INTERVAL));

    // =========================================
    // Data input - get data from kafka topic
    // =========================================
    topics.put(topicName, 1);
    JavaPairReceiverInputDStream<String, String> kafkaStream = KafkaUtils.createStream(ssc, zookeeper, groupId, topics);  
  
    // =========================================
    // Data process part - print every line 
    // =========================================
    kafkaStream.foreach(x -> {
      x.foreach(y -> {
        System.out.println("----- message -----");
        System.out.println("input = " + y._1 + "=>" + y._2);
      });
      return null;
    });
    
    ssc.start();
    ssc.awaitTermination();
  }
}

This demo hardcode read data from a kafka topic 'mytopic1', make sure you have data published to that topic.

3. Run the demo

For simplicity, only run the first one, the other 2 result same but need more external environment. The first one only need a external socket server, like linux  netcat tool can do this for you. That's also why the first one can be very convenient during developing phase.

First let's start a socket server listen on port 1111 with netcat.

Now start our spark application on a server with spark installed.

spark-submit --master local[*]  --class com.shengwang.demo.NakedSparkMain  /usr/sanss/qos-spark-1.0.jar

After spark application runs, you will find it dump a lot of log every 5 seconds, because we set the batch duration is 5 seconds in programe. Now type something in netcat window

You will see the message pring out in spark outout like below.

How to send message to Flume (avro or thrift) in Java

Apache Flume is mostly used to transfer data from "Out-of-Hadoop" world into Hadoop family.  In Flume, every message transfered is called 'Event', input is call "Source", output is called "Sink".

1. Flume Source & Sink type

Flume has many input and output types.

Here are some  wildly used source(input) types:

• "syslog" is comming with Unix system. 
• "avro" and "thrift", these 2 are very similar. Both are apache projects.  Flume will listen on a given port for RPC call and data is encoded by Avro or Thrift way repectively.
• "netcat"  listens on a given port and turns each line of text into an event. Just like a normal socket server. Every line is an Event.
• "exec", get source from a external command, like 'tail -f you_log_file'
• "http", A source type which accepts Flume Events by HTTP POST and GET.
• "jms", allows convert message from JMS. e.g.  ActiveMQ, into flume event
• "kafka", reads message from Apache Kafka topic.

Here are some  wildly used sink(output) types:

• "logger", log event at INFO level. Typically useful for testing/debugging purpose.
• "hdfs", writes events into the Hadoop Distributed File System (HDFS)
• "hive", write events into a Hive table or partition
• "hbase" and "asynchbase", writes data to HBase in sync or async way.
• "avro" and "thrift", events sent to this sink are turned into Avro or Thrift events and sent to the configured hostname / port pair.
• "null" just discards all events like /dev/null
• "kafka", public data to kafka topic.

2. Flume client for Avro/Thrift source.

If you using avro or thrift source as message input,  following is a java client to send message into flume. This is useful for developing or testing. For me a client like this helps me debugging my spark code.

package com.shengwang.demo;

import java.nio.charset.Charset;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

public class DemoMain {

  public static void main(String[] args) throws EventDeliveryException {
    String ip = "192.168.203.156";
    int port = 34343;
    
//  RpcClient client = RpcClientFactory.getDefaultInstance(ip, port); // Avro
    RpcClient client = RpcClientFactory.getThriftInstance(ip, port);  // Thrift
    Event event = EventBuilder.withBody("hello flume",Charset.forName("UTF8"));
    client.append(event);
    client.close();
  }
}

You need to add following dependency to your pom.xml file

    <dependency>
      <groupId>org.apache.flume</groupId>
      <artifactId>flume-ng-sdk</artifactId>
      <version>1.5.0</version>
    </dependency>

Since I'm using CDH 5.4.9, the flume version is 1.5.0 in my environment. Choose the version matches yours. The hierarchy looks like below.

3. Run it

For my lab's configuration, flume has a thrift source and output event to kafka. So I can use kafka console comsumer to print out the message. The message send out from my Java client to Flume, then transfers to Kafka topic by flume and finally gets consumed and prints out on command line. 

So first start the kafka-console-consumer on server running kafka, which is a tool come with kafka package, it will block and wait for message. Then run our DemoMain client. The message "hello flume" will print out on screen.