Multi-Dimensional Data Base scaling

What is Data Base scaling ?

every day the importance, complexity and quantity of data is growing. Tools and Frameworks to deal with this use-cases also growing. To meet different needs to analyse the data,  industry is understanding that monolithic server architecture may not be sufficient.

Systems requires flexibility in scaling. when we talk about scale, potentially we are talking about scale-Up and scale-out.

Scale-Up:

In this we will replace smaller capacity machines ( less CPU and less storage )  with higher capacity Machines.

Pros:

we can experience increase in performance immediately since all cpu, memory and disk ar attached together.

easy to maintain the system.

Cons:

we will experience longer downtime, because we need to setup the new server and migrate the data.

we will end up splitting the jobs to perform them on individual machines when the data grows. 

Scale-Out:

In this we will be adding more approximately optimal capacity machines to address the data growth

Pros:

less or no down time in adding a node to the existing cluster of nodes.

we can have cheaper machines to accommodate more data in the database.

good for GET and SET operations since we would not require to handle data indexing

Cons:

since we can not have proper data indexing retrieval will be very slow for complex data joins.

we need orchestration to keep tab of data availability and data locality.

we will not have flexibility to optimize the query performance.

To address above mentioned issues in system scale another idea is Multi-Dimensional Scala

Multi-Dimensional scale address 3 key areas in Data Base

1) Query performance

2) Indexing

3) Storage/Data

In Multi-Dimensional scale we would be able to scale-Up and Scale-out any of the above mentioned components.

as described in the below diagram, each of the above components would be installed on different zone/cluster of machines as a service.

Internals of the How Multi-Dimensional scale is facilitated in apache couchdb and SnowFlake would be discussed later.

Spark - Kafka Direct Streaming

The Spark Streaming integration for Kafka 0.10 Direct Stream approach. It provides simple parallelism, 1:1 correspondence between Kafka partitions and Spark partitions, and access to offsets and metadata. However, because the newer integration uses the new Kafka consumer API instead of the simple API, there are notable differences in usage

Each item in the stream is a ConsumerRecord

For possible kafkaParams, see Kafka consumer config docs. If your Spark batch duration is larger than the default Kafka heartbeat session timeout (30 seconds), increase heartbeat.interval.ms and session.timeout.ms appropriately. For batches larger than 5 minutes, this will require changing group.max.session.timeout.ms on the broker. Note that the example sets enable.auto.commit to false, for discussion see Storing Offsets below.

Environment we were using

Spark: 2.3

sparkstreaming-kafka: 0.10

scala: 2.11

export SPARK_MAJOR_VERSION=2

1) Get the latest offset information from the dataStore or the HDFS checkpointing file.

val lastcommitedoffsetdf = spark.read.option("delimiter", ",").schema(SubscriberMediaSchema.buildoffsetschema).option("header", "true").csv(appconf.getString("output.offsetcheckpoint"))
var fromOffsets = scala.collection.mutable.Map[TopicPartition,Long]()
val rows = lastcommitedoffsetdf.collect()
//"topic,partition,fromOffset,untilOffset,count"
rows.foreach (row => fromOffsets.put(new TopicPartition(row.getString(0), row.getInt(1)), row.getLong(3)))

2) Create the TopicPartition --> offset Map

val rows = lastcommitedoffsetdf.collect()

3) Construct the DirectStream with the offsets been created.

var messagesDStream : InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc, ConsumerStrategies.Subscribe[String, String](topicsset, kafkaparams, fromOffsets)

Execute the below steps if the cluster is kerberized environment

kinit -kt user.keytab user@DNS   [ user@BDA.NETSCOUT.COM] 

export KAFKA_OPTS="-Djava.security.auth.login.config=/tmp/kafka_jaas.conf" 

 

Default producer setup 

Start the producer

./bin/kafka-console-producer.sh --broker-list "hostxxxx.test.com:6667,hostxxxx.test.com:6667,tnblfxxxx.test.com:6667" --topic topicname --producer.config=/usr/hdp/2.6.3.13-5/kafka/config/producer.properties --security-protocol SASL_PLAINTEXT 

  

Default Consumer setup 

Start the consumer 

./bin/kafka-console-consumer.sh --bootstrap-server tnblfxxxx.test.com:6667 --topic topicname --from-beginning --security-protocol SASL_PLAINTEXT 

  

http://shzhangji.com/blog/2015/05/31/spark-streaming-logging-configuration/ 

 

Spark-submit: 

 

Local mode 

spark-submit \ 
--master local[2] \ 
--driver-memory 2g \ 
--jars "$ROOT/bin/myappp-assembly-0.1-deps.jar" \ 
--conf spark.driver.extraJavaOptions="-Dconfig.file=/data/kafkaapp/resources/application.conf -Djava.security.auth.login.config=/data/kafkaapp/kafka_local_jaas.conf" \ 
--class "NimsConsumerStream"   $ROOT/bin/myappp-assembly-0.1.jar "arguments" 

 

Yarn client mode 

spark-submit \ 
--master yarn  \ 
--deploy-mode client \ 
--driver-memory 2g \ 
--executor-memory 2g \ 
--executor-cores 2 \ 
--jars "/data/kafkaapp/bin/myapp-assembly-0.1-deps.jar" \ 
--files log4j.properties,application.conf,kafka_jaas.conf,app.truststore.jks,remote.keytab,krb5.conf \ 
--conf "spark.driver.extraJavaOptions=-Dlog4j.configuration=file://log4j.properties -Dlog4j.debug=true -Dconfig.file=application.conf -Djava.security.auth.login.config=kafka_jaas.conf" \ 
--conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=file://log4j.properties -Dlog4j.debug=true -Dconfig.file=application.conf -Djava.security.auth.login.config=kafka_jaas.conf" \ 
--class "NimsConsumerStream" \ 
$ROOT/bin/myappp-assembly-0.1.jar "arguments" 

 

Yarn Cluster mode 

spark-submit \
        --name "MediaApp" \
        --master yarn  \
        --deploy-mode cluster \
        --queue queue-name-exe \
        --principal user@DNS.DOMAIN.COM \
        --keytab /etc/user.keytab \
        --num-executors 10 \
        --driver-memory 10g \
        --executor-memory 5g \
        --executor-cores 5 \
        --jars "$ROOT/bin/myappp-assembly-0.1-deps.jar" \
        --files ${ROOT}/resources/log4j.properties,${ROOT}/resources/application.conf,${ROOT}/keys/kafka_jaas.conf,${ROOT}/keys/app.truststore.jks,${ROOT}/keys/kafka_cluster.keytab,${ROOT}/keys/krb5.conf \
        --conf "spark.driver.extraJavaOptions=-Dlog4j.configuration=./log4j.properties -Dlog4j.debug=true -Dconfig.file=./application.conf -Djava.security.auth.login.config=./kafka_jaas.conf -Djava.security.krb5.conf=./krb5.conf" \
        --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=./log4j.properties -Dlog4j.debug=true -Dconfig.file=./application.conf -Djava.security.auth.login.config=./kafka_jaas.conf -Djava.security.krb5.conf=./krb5.conf" \
        --conf "spark.hadoop.fs.hdfs.impl.disable.cache=true" \
        --conf "spark.streaming.backpressure.enabled=false" \
        --conf "spark.streaming.backpressure.initialRate=30000" \
        --conf "spark.streaming.receiver.maxRate=80000" \
        --conf "spark.driver.memoryOverhead=5000" \
        --conf "spark.executor.memoryOverhead=5000" \
        --class "com.kafka.mypackage.myConsumer" \
        $ROOT/bin/myapp-assembly-0.1.jar >myappsubmit_stdout_`date +%Y%m%d`.log 2>&1 &

New and better approach is the spark structured streaming using kafka connector which is a unbounded in-memory table using dataframes. we will discuss this in another post.