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.
- Kafka topic: test-v1
- Kafka topic partitions: 3
- 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 |
args[0]
{ "kafka": { "broker": "host:port", "kafka_topic": "topic_name" } }
args[1]: path to JSON file
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 } } }
mvn install
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'
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}}}'