rezaadie / sparkstreamingtohbase

Example Program : Kafka Producer, Kafka Consumer, and insert streaming data to HBase using Java and Spark.

SparkStreamingHBase

This example program, is example how to create Kafka Producer, Kafka Consumer, and insert streaming data to HBase using Java and Spark.

• Producer: reads hourly time series of humidity data (humidity.json) and sends to Kafka.
• Consumer: consumes data from Kafka, transforms, and saves to HBase.

Details

Kafka

• Kafka topic: test-v1
• Kafka topic partitions: 3

HBase

• HBase table name: test-table-1
• HBase row id: city name (example: portland, san_francisco)
• HBase column family name: humidity
• HBase column qualifier name: datetime (example: 2017-11-29_23:00:00, 2017-11-30_00:00:00)

	humidity
	2017-11-29_23:00:00	2017-11-30_00:00:00	...	yyyy-MM-dd_HH:mm:dd
portland	50.0	60.5	...	45.0
san_francisco	55.0	50.0	...	40.0

Program Arguments

Producer

args[0]

{
  "kafka": {
    "broker": "host:port",
    "kafka_topic": "topic_name"
  }
}

args[1]: path to JSON file

Consumer

args[0]

{
  "hbase": {
    "output_table": "table_name",
    "master": "host",
    "quorum": "host",
    "hbaseZkPort":"port",
  },
  "kafka": {
    "broker": "host:port",
    "group_name": "group_name",
    "auto_offset_reset": "none/earliest/latest",
    "auto_commit": true/false,
    "kafka_topics": {
      "topic_name1": number_partitions,
      "topic_name2": number_partitions
    }
  }
}

How to compile

mvn install

Running the program

Producer

spark-submit --class com.reza.dev.learn.main.Main_Producer --master local[2] target/SparkStreamingHBase-1.0-SNAPSHOT-jar-with-dependencies.jar '{"kafka":{"broker":"datanode01:6667,datanode02:6667,datanode03:6667","kafka_topic":"test"}}' '/doc/data/humidity.json'

Consumer

spark-submit --class com.reza.dev.learn.main.Main_Consumer --master local[2] --conf spark.streaming.backpressure.enabled=true --conf spark.streaming.kafka.maxRatePerPartition=100 target/SparkStreamingHBase-1.0-SNAPSHOT-jar-with-dependencies.jar '{"hbase":{"output_table":"test-table-1","master":"namenode01,namenode02","quorum":"master,namenode01,namenode02","hbaseZkPort":"2181"},"kafka":{"broker":"datanode01:6667,datanode02:6667,datanode03:6667","group_name":"group-test","auto_offset_reset":"earliest","auto_commit":true,"kafka_topics":{"test-v1":3}}}'