Deploying Kafka into ubuntu

Apache Kafka is a distributed message broker designed to handle large volumes of real-time data efficiently. Unlike traditional brokers like ActiveMQ and RabbitMQ, Kafka runs as a cluster of one or more servers which makes it highly scalable and due to this distributed nature it has inbuilt fault tolerance while delivering higher throughput when compared to its counterparts

Implementation of Single Node Kafka

Installing Java

sudo apt-get update
sudo apt-get install default-jre

Installing Zookeeper

sudo apt-get install zookeeperd

Create a service User for Kafka

sudo adduser --system --no-create-home --disabled-password --disabled-login kafka

Download Kafka

cd ~
curl http://kafka.apache.org/KEYS | gpg --import
wget https://dist.apache.org/repos/dist/release/kafka/1.0.1/kafka_2.12-1.0.1.tgz.asc
gpg --verify kafka_2.12-1.0.1.tgz.asc kafka_2.12-1.0.1.tgz

Create a directory for extracting Kafka

sudo mkdir /opt/kafka
sudo tar -xvzf kafka_2.12-1.0.1.tgz --directory /opt/kafka --strip-components 1

Delete Kafka tarball and .asc file

rm -rf kafka_2.12-1.0.1.tgz kafka_2.12-1.0.1.tgz.asc

Configuring Kafka Server

Setup Kafka to start automatically on bootup

Copy the following init script to /etc/init.d/kafka:

======***

DAEMON_PATH=/opt/kafka/bin
DAEMON_NAME=kafka
# Check that networking is up.
#[ ${NETWORKING} = "no" ] && exit 0

PATH=$PATH:$DAEMON_PATH

# See how we were called.
case "$1" in
 start)
       # Start daemon.
       echo "Starting $DAEMON_NAME";
       nohup $DAEMON_PATH/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
       ;;
 stop)
       # Stop daemons.
       echo "Shutting down $DAEMON_NAME";
       pid=`ps ax | grep -i 'kafka.Kafka' | grep -v grep | awk '{print $1}'`
       if [ -n "$pid" ]
         then
         kill -9 $pid
       else
         echo "Kafka was not Running"
       fi
       ;;
 restart)
       $0 stop
       sleep 2
       $0 start
       ;;
 status)
       pid=`ps ax | grep -i 'kafka.Kafka' | grep -v grep | awk '{print $1}'`
       if [ -n "$pid" ]
         then
         echo "Kafka is Running as PID: $pid"
       else
         echo "Kafka is not Running"
       fi
       ;;
 *)
       echo "Usage: $0 {start|stop|restart|status}"
       exit 1
esac

exit 0

======***

Make the Kafka service

sudo chmod 755 /etc/init.d/kafka
sudo update-rc.d kafka defaults

Start Stop the Kafka Services

sudo service kafka start
sudo service kafka status
sudo service kafka stop

Testing Kafka topics

sudo service kafka start
sudo service kafka status

Topic creation

/opt/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

Publish Msg to test topic

/opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

This will prompt for Msgs,  we can enter a test Msg

Consume Msg from the topic

/opt/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

Making Kafka Scalable

Requirement

Clustering the Zookeeper in all the Servers

Clustering the Kafka in All the servers

Install Zookeeper on all the servers and configure the servers in

/etc/zookeeper/conf/zoo.cfg
to mention all the nodes of the zookeeper

server.0=10.0.0.1:2888:3888
server.1=10.0.0.2:2888:3888
server.2=10.0.0.3:2888:3888

Once Kafka is installed in all the servers

/opt/kafka/config/server.properties
We will change the following settings.

broker.id should be unique for each node in the cluster.

for node-2 broker.id=1
for node-3 broker.id=2
change zookeeper. connect value to have such that it lists all zookeeper hosts with port

zookeeper.connect=10.0.0.1:2181,10.0.0.2:2181,10.0.0.3:2181

Enabling hive Authorization in Qubole

Once the Hive authorization is enabled in qubole we need to manage the users and permission by Hive authentication,  following are the some of the commands which will be used for the same.

1. Listing the Current Roles

Set role admin;

show roles

2. Create the roles

CREATE ROLE <role_name>;
Creates a new role. Only the admin role has privilege for this.

Eg:

Set role admin;
Create role sysadmin;

3. Grant Role to users

GRANT ROLE <role_name> TO USER <user_name>
 

Eg:

Set role admin;

Grant Role sysadmin to user rahul ;

4. Revoke a role from user

REVOKE ROLE <role_name> FROM USER <user_name>;

Eg:

Set role admin;
REVOKE Role sysadmin from user rahul;

5. List  Roles attached to a user

SHOW ROLE GRANT USER <user_name>;

Eg.

Set role admin;

show role grant user `rahul`;

6. List Users under a role

SHOW PRINCIPALS <Role_name>

Eg

Set role admin;

SHOW PRINCIPALS sysadmin

7. Assign Role access to tables

Sample Permission

SELECT privilege: It provides read access to an object (table).
INSERT privilege: It provides ability for adding data to an object (table).
UPDATE privilege: It provides ability for running UPDATE queries on an object (table).
DELETE privilege: It provides ability for deleting data in an object (table).
ALL privilege: It provides all privileges. In other words, this privilege gets translated into all the above privileges.

GRANT <Permission> ON <table_name> TO ROLE <role_name>;

Eg:

Grant all on default.testtable to role sysadmin

8. View Role/user Permissions on tables

Check all users who have been granted with a specific role

SHOW GRANT USER <user_name> ON <table_name|All>;
SHOW GRANT ROLE <user_name> ON <table_name|All>;

Eg:

SHOW GRANT user analytics on all

Parsing Value from a Json Field in Qubole.

Description of how to extract a value from a JSON field in Hive using the get_json_object function. When the data in one of the fields in the Hive environment is in JSON format, and we need to extract a value out of the JSON, we can use the get_json_object function. For example, if we have a column named jdata containing the following JSON:

get_json_object(column_name, '$.keyvalue')

The column name is : jdata and json the Column is as followes.

{
    "Foo": "ABC",
    "Bar": "20090101100000",
    "Quux": {
        "QuuxId": 1234,
        "QuuxName": "Sam"
    }
}

if we have to extract ABC : get_json_object(jdata, '$.Foo')