Apache Hive getting started

In this post I'm going to explain installation of apache hive and some first steps.

Download

Of course first we have to download everything: hive and hadoop (hive will not work without it). For me, closest locations for download were:

http://ftp.man.poznan.pl/apache/hive/hive-3.1.2/apache-hive-3.1.2-bin.tar.gz
http://ftp.man.poznan.pl/apache/hadoop/common/hadoop-3.2.0/hadoop-3.2.0.tar.gz

If you don't have java installed - you should install it as well.

Unpack and env setup

Now we have to unpack downloaded archives somewere. I unpacked them into folders:
/home/dmitry/Develop/hadoop
/home/dmitry/Develop/hive

Next we have to create a folder for hive warehouse:
$ sudo mkdir -p /user/hive/warehouse
$ sudo chmod a+rwx /user/hive
$ sudo chmod a+rwx /user/hive/warehouse

To setup env settings we can create a file hadoop-env.sh: 

#!/bin/sh

export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64/

export HADOOP_HOME=/home/dmitry/Develop/hadoop

export HIVE_HOME=/home/dmitry/Develop/hive

export PATH=$PATH:$HADOOP_HOME/bin:$HIVE_HOME/bin

Please make sure JAVA_HOME points to proper location. 

To put all these variables into current session, run in terminal:
$source ./hadoop-env.sh

Another option is to add everything(content of file from above) into  $HOME/.bashrc

Checking if everything configured properly

Actually hadoop is not confugured and hdfs daemons are not started but we this is not neede for hive. We can check it this way:

Hadoop test: 

dmitry@dmitry-ThinkPad-X220:~/Develop$ hdfs dfs -ls /home

Found 3 items                                                                                                                         

-rw-r--r--   1 root   root           31 2018-04-09 19:28 /home/.directory                                                            

drwxr-xr-x   - dmitry dmitry       4096 2019-09-06 21:58 /home/dmitry                                                                 

drwx------   - root   root        16384 2019-08-29 21:11 /home/lost+found 

If it shows my files, so it works. 

Hive test:

First we have to init meta-data database:

dmitry@dmitry-ThinkPad-X220:~/Develop$ cd $HIVE_HOME

dmitry@dmitry-ThinkPad-X220:~/Develop/hive$ schematool -initSchema -dbType derby

At the end of execution it should print:

Initialization script completed

schemaTool completed

And now we can run hive CLI and try to create a table:

dmitry@dmitry-ThinkPad-X220:~/Develop$ hive

                   

Hive Session ID = a62b7d5e-1955-483d-973a-f7416626ebf8                                                                              

                                                                                                                                      

Logging initialized using configuration in jar:file:/home/dmitry/Develop/hive-3.1.2/lib/hive-common-3.1.2.jar!/hive-log4j2.properties Async: true

Hive-on-MR is deprecated in Hive 2 and may not be available in the future versions. Consider using a different execution engine (i.e. spark, tez) or using Hive 1.X releases.

hive> 

If everything is ok it should be possible to create a table:

hive> CREATE TABLE TEST(ID INT);

OK

Time taken: 0.233 seconds

hive>

Now we can try to insert new values into just created table:
insert into test values(1);

And checking:
hive> select * from test;
OK
1

Value was returned, but as you can see column header is disabled by default. Let's enable it:
hive>  set hive.cli.print.header=true;
hive> select * from test;
OK
test.id
1

If you always prefer seeing the headers, put the first line in your $HOME/.hiverc file.

Work with databases

By default, hive use "default" database:

hive> show databases;

OK

database_name

default

But can always create our own if its needed:

create database mydb comment 'my test db';

now we have : 

hive> show databases;

OK

database_name

default

mydb

Time taken: 0.027 seconds, Fetched: 2 row(s)

To get the details: 

hive> describe database mydb;

OK

db_name comment location        owner_name      owner_type      parameters

mydb    my test db      file:/user/hive/warehouse/mydb.db       dmitry  USER

Time taken: 0.036 seconds, Fetched: 1 row(s)

To switch database we can use:

hive> use mydb;

OK

Time taken: 0.031 seconds

To check tables:

hive> show tables;

OK

tab_name

Time taken: 0.039 seconds

Also  we can configure hive to print database name in prompt:

hive>  set hive.cli.print.current.db=true;

hive (mydb)> 

- and instead of "hive" we will have "hive (database name)"

Work with tables:  Internal(Managed) Tables

Managed Tables
The tables we have created so far are called managed tables or sometimes called inter-
nal tables, because Hive controls the lifecycle of their data (more or less). As we’ve seen,
Hive stores the data for these tables in a subdirectory under the directory defined by
hive.metastore.warehouse.dir (e.g., /user/hive/warehouse), by default.
When we drop a managed table, Hive deletes
the data in the table.

Syntax of table creation is similar to regular databases:

hive (mydb)> CREATE TABLE IF NOT EXISTS users (

           >   user_name  STRING COMMENT 'Name',

           >   user_roles  ARRAY<STRING> COMMENT 'Roles',

           >   user_address STRUCT<city:STRING, street:STRING, zip:INT> COMMENT 'Address'

           > )

           > COMMENT 'My Table'

           > TBLPROPERTIES ('creator'='Dmitry');

Work with tables:  External Tables

The EXTERNAL keyword tells Hive this table is external and the LOCATION ... clause is

required to tell Hive where it’s located. Because it’s external, Hive does not assume it owns the data. Therefore, dropping the table does not delete the data, although the metadata for the table will be deleted.

Let's say we have a folder:
/user/hive/groups

We can create a csv file here with content(name of file doesn't matter):

1,Sysdba

2,Dev

3,Others

And now we can create an external table:

CREATE EXTERNAL TABLE IF NOT EXISTS groups (
group_id INT,
group_name STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION '/user/hive/groups';

Let's check it:

hive (mydb)> select * from groups;

OK

groups.group_id groups.group_name

1       Sysdba

2       Dev

3       Others

Time taken: 0.133 seconds, Fetched: 3 row(s)

To understand "who is who" (internal table or external) we can use command "describe extended":

hive (mydb)> describe extended groups;

OK

col_name        data_type       comment

group_id                int                                         

group_name              string                                      

                 

Detailed Table Information      Table(tableName:groups, dbName:mydb, owner:dmitry, createTime:1568231912, lastAccessTime:0, retention:0, sd:StorageDescriptor(cols:[FieldSchema(name:group_id, type:int, comment:null), FieldSchema(name:group_name, type:string, comment:null)], location:file:/user/hive/groups, inputFormat:org.apache.hadoop.mapred.TextInputFormat, outputFormat:org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{serialization.format=,, field.delim=,}), bucketCols:[], sortCols:[], parameters:{}, skewedInfo:SkewedInfo(skewedColNames:[], skewedColValues:[], skewedColValueLocationMaps:{}), storedAsSubDirectories:false), partitionKeys:[], parameters:{transient_lastDdlTime=1568231912, bucketing_version=2, totalSize=44, EXTERNAL=TRUE, numFiles=1}, viewOriginalText:null, viewExpandedText:null, tableType:EXTERNAL_TABLE, rewriteEnabled:false, catName:hive, ownerType:USER)

Time taken: 0.096 seconds, Fetched: 4 row(s)

From this output we can see: tableType:EXTERNAL_TABLE

Partitioning

To create partitioned table we just have to add to table definition PARTITIONED BY and list columns. Let's drop previously creater table users and re-create it partitioned by department:

CREATE TABLE IF NOT EXISTS users (

 user_name  STRING COMMENT 'Name',

 user_roles  ARRAY<STRING> COMMENT 'Roles',

user_address STRUCT<city:STRING, street:STRING, zip:INT> COMMENT 'Address'

)
PARTITIONED BY (user_department STRING);

Now let's add some data:
hive (mydb)> insert into users (user_name, user_department) values("joe","AAA");
hive (mydb)> insert into users (user_name, user_department) values("moe","BBB");


And let's now check what we have in file system:
dmitry@dmitry-ThinkPad-X220:/user/hive/warehouse/mydb.db/users$ pwd
/user/hive/warehouse/mydb.db/users
dmitry@dmitry-ThinkPad-X220:/user/hive/warehouse/mydb.db/users$ ll
total 16
drwxr-xr-x 4 dmitry dmitry 4096 wrz 12 21:07  ./
drwxr-xr-x 3 dmitry dmitry 4096 wrz 12 21:04  ../
drwxr-xr-x 2 dmitry dmitry 4096 wrz 12 21:07 'user_department=AAA'/
drwxr-xr-x 2 dmitry dmitry 4096 wrz 12 21:07 'user_department=BBB'/

Values with different department are located in different folders.

Exporting data

To export data from tables we can use next syntax:
hive> INSERT OVERWRITE LOCAL DIRECTORY '/home/dmitry/hive-export' SELECT * FROM GROUPS;

Now we can check output directly from hive:
hive> ! less /home/dmitry/hive-export/000000_0;
1Sysdba
2Dev
3Others

Loading data

To load data we can use next syntax:
LOAD DATA LOCAL INPATH '/home/dmitry/hive-export' OVERWRITE INTO TABLE GROUPS;

The end

And this is basically the end of this post:)

Apache Spark - getting started: batch and stream data processing using scala

0. Intro 

In this post I'm going to explain basics of Apache Spark: RDD, SparkSQL, DataFrames, SparkStreaming. Actually Spark -  it's just a library for data processing. So it can be executed without any BigData-related stuff. You can just run code from this post without any Hadoop/HDFS.

1. What is Apache Spark

from wiki:
Apache Spark is an open-source cluster-computing framework. Originally developed at the University of California, Berkeley's AMPLab, the Spark codebase was later donated to the Apache Software Foundation, which has maintained it since. Spark provides an interface for programming entire clusters with implicit data parallelism and fault tolerance.

From official documentaion:
Apache Spark is a fast and general-purpose cluster computing system. It provides high-level APIs in Java, Scala, Python and R, and an optimized engine that supports general execution graphs. It also supports a rich set of higher-level tools including Spark SQL for SQL and structured data processing, MLlib for machine learning, GraphX for graph processing, and Spark Streaming.

As you can see, there is no mentioning of Hadoop/HDFS at all. It CAN work with Hadoop/HDFS, it CAN work with cluster resource manager like YARN. But for getting familiar,  also it can be used  for processing local files in standalone mode.

2. What is Lambda architecture

From wiki:
Lambda architecture is a data-processing architecture designed to handle massive quantities of data by taking advantage of both batch- and stream-processingmethods. This approach to architecture attempts to balance latency, throughput, and fault-tolerance by using batch processing to provide comprehensive and accurate views of batch data, while simultaneously using real-time stream processing to provide views of online data. The two view outputs may be joined before presentation. The rise of lambda architecture is correlated with the growth of big data, real-time analytics, and the drive to mitigate the latencies of map-reduce.^[1]

In shorts, lambda architecture is a combination of 2 processing types:
- slow but precese - in this post we will use Spark RDD/DataFrame for it
- fast but not precise -  in this post we will use Spark Streaming for it

3. Project setup: pom.xml

Here I'm using maven with scala. In pom.xml I'm adding dependencies for spark-core, spark-sql, spark-streaming.

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.demien</groupId>
    <artifactId>sparktest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>


    </dependencies>


</project>

4. Project structure

As you can see, we will be using 2 processing types of batch processing: RDD and DataFrame. And one type of stream processing: DStream. Apart from that we will be using SparkUtils class for creation of common sprak entities: SparkContext, SparkSession. And a generator of text files(for streaming) TextFileCreator.

5. SparkUtils

As I mentioned before, we will be using SparkUtils class for creation of common sprak entities: SparkContext, SparkSession:

package com.demien.sparktest

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

object SparkUtils {

  val DEF_APP_NAME = "MySparkApp"  val DEF_MASTER = "local[2]"
  //The SparkContext object - connection to a Spark execution environment and created RDDs  def getSparkContext(appName: String, master: String): SparkContext = new SparkContext(new SparkConf().setAppName(appName).setMaster(master))

  def getSparkContext(): SparkContext = getSparkContext(DEF_APP_NAME, DEF_MASTER)

  //The SparkSession - connection to dataframes and SQLs  def getSparkSession(appName: String, master: String): SparkSession = SparkSession
    .builder()
    .appName(appName)
    .master(master)
    .getOrCreate()

  def getSparkSession(): SparkSession = getSparkSession(DEF_APP_NAME, DEF_MASTER)


}

6. Batch processing: RDD

From official documentation:

The main abstraction Spark provides is a resilient distributed dataset (RDD), which is a collection of elements partitioned across the nodes of the cluster that can be operated on in parallel. RDDs are created by starting with a file in the Hadoop file system (or any other Hadoop-supported file system), or an existing Scala collection in the driver program, and transforming it. Users may also ask Spark to persist an RDD in memory, allowing it to be reused efficiently across parallel operations. Finally, RDDs automatically recover from node failures.

RDD Example: 

package com.demien.sparktest.batch

import com.demien.sparktest.SparkUtils
import org.apache.spark.rdd.RDDwith

// https://spark.apache.org/docs/2.3.0/rdd-programming-guide.htmlobject RddExample extends App {

  val sc = SparkUtils.getSparkContext()
  val file = sc.textFile("src/main/resources/sample.txt")
  val words: RDD[String] = file.flatMap(l => l.split(" ")).filter(w => w.length > 1)
  val pairs: RDD[(String, Int)] = words.map(s => (s, 1)) // [the, of, the] => (the, 1) (of, 1) (the, 1)  val counts: RDD[(String, Int)] = pairs.reduceByKey((a, b) => a + b) // (the, 1) (of, 1) (the, 1) => (the, 2) (of, 1)  val countByWord: RDD[(Int, String)] = counts.map(p => p.swap) // (the, 2) (of, 1) => (2, the) (1, of)  val countByWordSorted: RDD[(Int, String)] = countByWord.sortByKey(false)
  val top5 = countByWordSorted.take(5)

  top5.foreach(p => println(p))
}

- I added comments and  datatypes for RDD variables(which of cource are not needed here) to make it more clear. This RDD example is processing sample text file - it's just a text from ApacheSpark wiki. We are splitting text into words, creating for every work "paired object" with word itself and number 1. After that, we are groupping these pairs using word as a key and counting provided numbers.

Results:
(55,the)
(46,of)
(43,Spark)
(39,and)
(24,in)

- as you can see, most popular word(excluding "the", "of", "and", "in") in Spark wiki is "Spark" :)

7. Batch processing: DataFrame/Spark SQL

From official documentation:

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs. The DataFrame API is available in Scala, Java, Python, and R. In Scala and Java, a DataFrame is represented by a Dataset of Rows. In the Scala API, DataFrame is simply a type alias of Dataset[Row]. While, in Java API, users need to use Dataset<Row> to represent a DataFrame.

DataFrames - are the structured datasets, so as a sample file we will be using not TEXT file but  JSON like this:

{
  "name": "Keeley Bosco",
  "email": "katlyn@jenkinsmaggio.net",
  "city": "Lake Gladysberg",
  "mac": "08:fd:0b:cd:77:f7",
  "timestamp": "2015-04-25 13:57:36 +0700",
  "creditcard": "1228-1221-1221-1431"}
{
  "name": "Rubye Jerde",
  "email": "juvenal@johnston.name",
  "city": null,
  "mac": "90:4d:fa:42:63:a2",
  "timestamp": "2015-04-25 09:02:04 +0700",
  "creditcard": "1228-1221-1221-1431"}

DataFrame example:

package com.demien.sparktest.batch

// https://spark.apache.org/docs/latest/sql-programming-guide.html
import org.apache.spark.sql.SparkSession

object DataFrameExample extends App {

  val spark = SparkSession
    .builder()
    .appName("Spark SQL basic example")
    .config("spark.master", "local")
    .getOrCreate()

  // For implicit conversions like converting RDDs to DataFrames  import spark.implicits._


  val df = spark.read.json("src/main/resources/people.json")
  df.printSchema()
  df.createOrReplaceTempView("people")
  val sqlDF = spark.sql("SELECT * FROM people where email like '%net%' ")
  sqlDF.show()

  case class Person(name: String, email: String, city: String, mac: String, timestamp: String, creditcard: String)

  val peopleDS = spark.read.json("src/main/resources/people.json").as[Person]
  val filteredDS = peopleDS.filter(p => p.email != null && p.email.contains("net"))
  filteredDS.show()


}

We are using SparkSQL to query our structured dataset(DataFrame) for people which have "%net%" in their emails. Also, at the end we are doing the same thing again, but using using DataSet api.
Or cource, in both cases results are the same:

+---------------+-------------------+--------------------+-----------------+----------------+--------------------+
|           city|         creditcard|               email|              mac|            name|           timestamp|
+---------------+-------------------+--------------------+-----------------+----------------+--------------------+
|Lake Gladysberg|1228-1221-1221-1431|katlyn@jenkinsmag...|08:fd:0b:cd:77:f7|    Keeley Bosco|2015-04-25 13:57:...|
|           null|1228-1221-1221-1431|emery_kunze@rogah...|3a:af:c9:0b:5c:08|Celine Ankunding|2015-04-25 14:22:...|
+---------------+-------------------+--------------------+-----------------+----------------+--------------------+


Unfortunatelly, spark is not showing full values, but these email values  are:

"katlyn@jenkinsmaggio.net, "emery_kunze@rogahn.net"

- emails which contain "net".

8. Stream processing: DStream

From official documentation:

Spark Streaming is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams. Data can be ingested from many sources like Kafka, Flume, Kinesis, or TCP sockets, and can be processed using complex algorithms expressed with high-level functions like map, reduce, join and window. Finally, processed data can be pushed out to filesystems, databases, and live dashboards. In fact, you can apply Spark’s machine learning and graph processing algorithms on data streams.

8.1 Stream processing: DStream: TextFileCreator

To simulate stream of data, we will create the simple application which is creating text file every 10 seconds. As a source for this file I will be using again text from ApachSpark wiki. 

package com.demien.sparktest

import java.io.FileWriter
import java.util.Date

import scala.io.Source
import scala.util.Random

object TextFileCreator extends App {

  val listOfLines = Source.fromFile("src/main/resources/sample.txt").getLines.toList
  val rnd = new Random()

  while (true) {

    val fileName = new Date().getTime
    val fullFileName = "data/" + fileName + ".txt"    val fw = new FileWriter(fullFileName, true)
    println("writing to " + fullFileName)

    val linesCount = rnd.nextInt(20) + 5    for (i <- 1 to linesCount) fw.write(listOfLines(rnd.nextInt(100)) + "\n")

    fw.close()
    Thread.sleep(10000)

  }

}

8.2 Stream processing: DStream: Streaming example itself

Our application will be monitoring "data" folder for new files. When new file is received - it will be processed. To simulate some statefull activity we will be using function:  specFunc - the point is to constantly calculate count of words. 

package com.demien.sparktest.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}

object DStreamExample extends App {


  val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  val ssc = new StreamingContext(conf, Seconds(10))
  ssc.checkpoint("spark-checkpoint")

  val lines = ssc.textFileStream("data")
  val words = lines.flatMap(_.split(" "))
  val pairs = words.map(word => (word, 1))
  val wordCounts = pairs.reduceByKey(_ + _)


  def specFunc = (key: String, value: Option[Int], state: State[Int]) => {
    var newState = state.getOption().getOrElse(0)
    var newValue = value.getOrElse(1)
    newState = newState + newValue
    state.update(newState)
    (key, newValue)
  }

  val spec = StateSpec.function(specFunc).timeout(Seconds(30))

  val wordsMapped = wordCounts.mapWithState(spec)

  // top 10  wordsMapped.stateSnapshots().foreachRDD(rdd => {
    rdd.map(e => (e._2, e._1)).sortByKey(false).take(10).foreach(e => print(e._1, e._2))

  })

  ssc.start() // Start the computation  ssc.awaitTermination() // Wait for the computation to terminate
}

8.3. Stream processing: DStream: Execution

Of course, we have to run both: TextFileCreator and DStreamExample.

TextFileCreator is creating files:
writing to data/1524921644177.txt
writing to data/1524921654262.txt
writing to data/1524921664264.txt
writing to data/1524921674266.txt
writing to data/1524921684267.txt
writing to data/1524921694268.txt
writing to data/1524921704270.txt


And DStreamExample is processing them and counting words:

.....
(17,a)(16,Spark)(15,)(14,the)(10,Apache)(10,of)(7,//)(5,is)(5,can)(5,in)
(17,)(17,a)(17,Spark)(14,the)(10,Apache)(10,of)(7,//)(5,is)(5,can)(5,in)
(25,)(24,the)(21,Spark)(21,a)(19,of)(12,Apache)(10,can)(10,as)(10,and)(9,is)
(35,)(27,the)(25,Spark)(25,a)(20,of)(17,and)(14,Apache)(12,as)(12,//)(10,can)
(45,)(27,the)(26,Spark)(25,a)(20,of)(17,and)(14,Apache)(12,as)(12,//)(10,can)

As you can see, results are similar to what we had in RddExample: most popular words are Spark and Apache.

9. The end. 

As you can see, to try Apache Spark you don't need Hadoop/Yarn - it's possible to run it in a standalone mode without all these compicated things. Source code can be downloaded from here.

Apache Kafka - getting started. Simple java project.

0. Intro

Kafka® is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies.

From wiki: 

The project aims to provide a unified, high-throughput, low-latency platform for handling real-time data feeds. Its storage layer is essentially a "massively scalable pub/sub message queue architected as a distributed transaction log,"^[3] making it highly valuable for enterprise infrastructures to process streaming data. Additionally, Kafka connects to external systems (for data import/export) via Kafka Connect and provides Kafka Streams, a Java stream processing library.

The design is heavily influenced by transaction logs.^[4]

Last time Apache Kafka is getting more and more popular. With growing popularity of event-sourcing concept, more and more developers are switching to Kafka as primary storage of events. Kafka has everything for this: it's very hast, compact, scalable, "user-friendly"....
In this post I'll show basic simple operations like "send"(by producer) and "receive"(by consumer) messages.

1. Downloading and running kafka

This page is explaining very well how to download and run kafka. If you're using Windows, you can use next commands from "bin/windows" folder:

Run these commands from your Kafka root folder:

cd bin/windows

Then run Zookeper server:

zookeeper-server-start.bat ../../config/zookeeper.properties

Then run Kafka server:

kafka-server-start.bat ../../config/server.properties

Now when kafka is running you can check it by creating a topic and getting topic list:

Create a topic:

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test0

List topics:

kafka-topics.bat --list --zookeeper localhost:2181

Response should be something like:
D:\Projects\kafka_2.11-1.0.0\bin\windows>kafka-topics.bat --list --zookeeper localhost:2181
test0

2. Project structure

Our project structure is very simple: we need just 2 files MessageProducer and MessageConsumer. 

build.gradle file:

group 'com.demien'version '1.0-SNAPSHOT'
apply plugin: 'java'
sourceCompatibility = 1.8
repositories {
    mavenCentral()
}

dependencies {
    compile 'org.apache.kafka:kafka-clients:0.9.0.0'    compile 'org.slf4j:slf4j-api:1.7.12'    compile 'org.slf4j:slf4j-log4j12:1.7.12'    compile 'log4j:log4j:1.2.17'

    testCompile group: 'junit', name: 'junit', version: '4.11'}

3. Producer

It designed as generic by KEY,VALUE types. Also I added to constructor optional messageSentCallback parameter - this callBack will be called when message was sent.

package com.demien.kafka;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;

public class MessageProducer<K, V> {

    private final Producer kafkaProducer;
    private final String topicName;
    private final Consumer<RecordMetadata> messageSentCallback;

    public MessageProducer(String topicName) {
        this(topicName, null);
    }

    public MessageProducer(String topicName, Consumer<RecordMetadata> messageSentCallback) {
        Properties configProperties = new Properties();
        configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        this.kafkaProducer = new KafkaProducer(configProperties);
        this.topicName = topicName;
        this.messageSentCallback = messageSentCallback;
    }


    public void sendMessage(K key, V value) {
        ProducerRecord<K, V> rec = new ProducerRecord<K, V>(topicName, key, value);
        Future<RecordMetadata> future = kafkaProducer.send(rec);
        if (messageSentCallback != null) {
            CompletableFuture.supplyAsync(() -> {
                try {
                    RecordMetadata recordMetadata = future.get();
                    messageSentCallback.accept(recordMetadata);
                } catch (Exception e) {
                }
                return null;
            });
        }
    }

    public void close() {
        kafkaProducer.close();
    }


    public static void main(String[] args) throws InterruptedException {
        MessageProducer<String, String> testProducer = new MessageProducer<String, String>("test0", (recordMetadata) -> {
            System.out.println("Message was sent: offset:" + recordMetadata.offset() + " partition:" + recordMetadata.partition() + " topic:" + recordMetadata.topic());
        });
        testProducer.sendMessage(null, "Test 1 " + new Date().toString());
        testProducer.sendMessage(null, "Test 2 " + new Date().toString());
        testProducer.sendMessage(null, "Test 3 " + new Date().toString());
        testProducer.close();
    }


}

4. Consumer

This class is more complicated, because it's designed to deal with the offsets for reading the data.
Consumer can start reading form the beginning, from the end, or from provided offset. That is why constructor is so complicated. Method for receiving messages is pretty simple. Supplier for cuncumed messages is provided in constructor.

package com.demien.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.Properties;

import java.util.function.BiConsumer;

public class MessageConsumer<K, V> {
    private final String topic;
    private final String groupId;
    private final long startingOffset;
    private final KafkaConsumer<K, V> kafkaConsumer;

    public MessageConsumer(String topic, String groupId) {
        this(topic, groupId, -1);
    }

    /**     * @param topic - id of topic     * @param groupId - id of consumer group     * @param startingOffset - offset to read messages. 0 - from the beginning.      *                       -1 - from the end. other values - start reading from this value                            */    public MessageConsumer(String topic, String groupId, long startingOffset) {
        this.topic = topic;
        this.groupId = groupId;
        this.startingOffset = startingOffset;

        Properties configProperties = new Properties();
        configProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        configProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        configProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        configProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, "testClient");
        configProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        configProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        kafkaConsumer = new KafkaConsumer<>(configProperties);

        kafkaConsumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.printf("%s topic-partitions are revoked from this consumer\n", Arrays.toString(partitions.toArray()));
            }

            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.printf("%s topic-partitions are assigned to this consumer\n", Arrays.toString(partitions.toArray()));
                Iterator<TopicPartition> topicPartitionIterator = partitions.iterator();
                while (topicPartitionIterator.hasNext()) {
                    TopicPartition topicPartition = topicPartitionIterator.next();
                    System.out.println("Current offset is " + kafkaConsumer.position(topicPartition) + " committed offset is ->" + kafkaConsumer.committed(topicPartition));
                    if (MessageConsumer.this.startingOffset == 0) {
                        System.out.println("Setting offset to begining");

                        kafkaConsumer.seekToBeginning(topicPartition);
                    } else if (MessageConsumer.this.startingOffset == -1) {
                        System.out.println("Setting it to the end ");

                        kafkaConsumer.seekToEnd(topicPartition);
                    } else {
                        System.out.println("Resetting offset to " + MessageConsumer.this.startingOffset);
                        kafkaConsumer.seek(topicPartition, MessageConsumer.this.startingOffset);
                    }
                }
            }
        });

    }

    public void startReceiving(BiConsumer<K, V> biConsumer) {
        try {
            while (true) {
                ConsumerRecords<K, V> records = kafkaConsumer.poll(100);
                records.forEach(record->  biConsumer.accept(record.key(), record.value()));
                if (startingOffset == -2) kafkaConsumer.commitSync();
            }
        } finally {
            kafkaConsumer.close();
        }
    }

    public static void main(String[] args) {
        final MessageConsumer<String, String> testConsumer = new MessageConsumer<>("test0", "testGroup");
        testConsumer.startReceiving( (k,v) -> System.out.println("received:"+v) );

    }
}

5. Execution 

Let's start the Consumer now. It should output something like:

[test0-0] topic-partitions are assigned to this consumer
Current offset is 0 committed offset is ->null


Now let's start the Producer. Is should send 3 test messages and print information about them:

Message was sent: offset:0 partition:0 topic:test0
Message was sent: offset:1 partition:0 topic:test0
Message was sent: offset:2 partition:0 topic:test0

Consumer also should print information about received messages:

received:Test 1 Tue Jan 23 15:36:29 CET 2018
received:Test 2 Tue Jan 23 15:36:29 CET 2018
received:Test 3 Tue Jan 23 15:36:29 CET 2018

Let's restart our consumer now. By default value in our constructor, if will be reading data from the end, so previous messages will not be shown:

[test0-0] topic-partitions are assigned to this consumer
Current offset is 3 committed offset is ->OffsetAndMetadata{offset=3, metadata=''}
Setting it to the end 


Now we can try to read previous messages by changing the constructor parameter:

public static void main(String[] args) {
    final MessageConsumer<String, String> testConsumer = new MessageConsumer<>("test0", "testGroup", 2);
    testConsumer.startReceiving( (k,v) -> System.out.println("received:"+v) );
}

- it's now 2 so we will be reading from offset 2. Let's restart it again:

[test0-0] topic-partitions are assigned to this consumer
Current offset is 3 committed offset is ->OffsetAndMetadata{offset=3, metadata=''}
Resetting offset to 2
received:Test 3 Tue Jan 23 15:36:29 CET 2018

- now last previous message with the offset 2 was read.

6. The end

Source code can be downloaded from here.