Spark execution plans

 Spark execution plan: Spark's Catalyst optimizer creates plans

• Logical Plan: abstract syntax tree (AST) and this doesn’t have how data divided into partitions, what algorithm would use
• Physical Plan: how the data will be partitioned, which specific algorithms will be used, and how the results will be returned
• Execution Plan: physical operation that involves shuffling data, reading data, filtering data, or performing computations

How Spark executer handles threads

 How spark executer handles threads:

Spark executor handles threads by determining the number of tasks that can be run in parallel based on the number of cores, scheduling tasks to available threads, ensuring thread safety, handling task failures, and managing resources

1. Number of cores

2. Scheduling

3. Thread Safety

4. Fault Tolerance

5. Resource Management

How to fix Lazy evaluation overhead

Technic to handle Lazy evaluation overhead 

1. Caching and Persisting RDDs

2. Using Broadcast Variables 

3. Avoiding Operations that Cause Shuffling : Operations like groupByKey and reduceByKey 

4. Using the right data structures**: DataFrames and Datasets 

5. Tuning Spark configurations: auto scale confgs

6. Checkpointing

Spark - Shuffle Optimisation:

Spark -  Shuffle Optimisation:

• spark.sql.shuffle.partitions parameter  - 200mb
• spark.serializer to org.apache.spark.serializer.KryoSerializer faster than java serial and deserial
• Reduce Disk I/O  - spark.shuffle.compress — whether the engine would compress shuffle outputs or not. Default value is “true”.
• spark.shuffle.spill.compress — whether to compress intermediate shuffle spill files or not. Default value is “true”.
• spark.io.compression.codec codec for compressing the data, which is snappy by default.

For smaller datasets , generally snappy works well with most of the datasets but there is emerging compression codec “zstld” which surpasses snappy performance. Details are here.This has been introduced by Facebook and attached is the link for details.

Optimize Spark’s In-memory computation: Spark uses memory to store intermediate results during shuffling. Adjust the memory usage for shuffling by tuning the spark.memory.fraction parameter. By default, this parameter is set to 0.6, which means that 60% of the executor memory is used for storage/caching and 40% is used for execution

BigData - Distributed System Design Patterns

Below are thlist of interesting topic/technics which are used/considered while building Hadoop or any Distributed systems.

Bloom Filters  - used to check key available in big/huge dataset

High-water mark index -  index refers, index all followers are written

Lease - lock release on resource

Heartbeat - Worker periodically send signal to master to indicate there availability. 

Fence -  Put a 'Fence' around the previous leader to prevent it from doing any damage or causing corruption. Fencing is the idea of putting a fence around a previously active leader so that it cannot access cluster resources and hence stop serving any read/write request. The following two techniques are used:

• Resource fencing
• Node fencing

Examples: HDFS uses fencing to stop the previously active NameNode from accessing cluster resources, thereby stopping it from servicing requests.

High-water mark index

Distributed systems keep multiple copies of data for fault tolerance and higher availability. To achieve strong consistency, one of the options is to use a leader-follower setup, where the leader is responsible for entertaining all the writes, and the followers replicate data from the leader.

Quorum - used for HA. 

Write-ahead log (WAL) -  log file where mater node write/append data 

Circuit Breaker Micro Service design Pattern

 

Circuit Breaker Pattern - it acts as a safeguard against service failures by monitoring x, setting thresholds, and temporarily halting/Stopping traffic to failing services. It helps prevent cascading failures and maintains system stability, ensuring reliable performance in distributed architectures.

Closed State: Initially, the circuit breaker is in a Closed state, allowing requests through.

Open State: If a certain number of requests fail (like timeouts or errors), the breaker "trips" to an Open state. This stops calls to the failing service, giving it time to recover.

Half-Open State: After a cooldown period, the breaker enters a Half-Open state, allowing a limited number of test requests through. If these succeed, it goes back to Closed; if not, it returns to Open.

Top Micro services Design Patterns in 5 mins

 

Critical RCE 0day (ZERO day) in Apache Log4j library was reported (CVE-2021-44228) - Fix

Critical RCE 0day in Apache Log4j library was reported (CVE-2021-44228) and Apache suggested to upgrade your log4j to version 2.15.x:  https://threatpost.com/zero-day-in-ubiquitous-apache-log4j-tool-under-active-attack/176937/

https://logging.apache.org/log4j/2.x/

The Log4j team has been made aware of a security vulnerability, CVE-2021-44228, that has been addressed in Log4j 2.15.0.

For those who cannot upgrade to 2.15.0, in releases >=2.10, this vulnerability can be mitigated by setting either the system property log4j2.formatMsgNoLookups or the environment variable LOG4J_FORMAT_MSG_NO_LOOKUPS to true. For releases from 2.0-beta9 to 2.10.0, the mitigation is to remove the JndiLookup class from the classpath: zip -q -d log4j-core-*.jar org/apache/logging/log4j/core/lookup/JndiLookup.class.

Google Cloud Dataflow Pipeline using JdbctoBigQuery template "timezone region not found" issue - Fix

Sometimes Dataflow pipeline with JdbctoBigQuery template may not work due to : "error occurred at recursive SQL level 1 ORA-01882: timezone region not found"

Error details:

Error
2021-07-21T06:22:13.292218504ZError message from worker: java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.sql.SQLException: Cannot create PoolableConnectionFactory (ORA-00604: error occurred at recursive SQL level 1 ORA-01882: timezone region not found ) org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$1.typedApply(IntrinsicMapTaskExecutorFactory.java:197) org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$1.typedApply(IntrinsicMapTaskExecutorFactory.java:168)

Steps to Fix the issue:

1. extract oracle/jdbc/defaultConnectionProperties.properties file from ojdbcXXXXX.jar file.

    jar xf o ojdbcXXXXX.jar oracle/jdbc/defaultConnectionProperties.properties

2. add oracle.jdbc.timezoneAsRegion=false propery

3. then update jdbc jar file with updated defaultConnectionProperties.properties

    jar uf ojdbcXXXXX.jar oracle/jdbc/defaultConnectionProperties.properties

Hibernate Annotation configuration With Spring

Basically, there are two ways you can define an entity POJO:

* At the getter-methods level
* At the object's properties level

The following example shows you how you can create an entity bean named MyObjectVO:

Entity bean with annotations declared at method level

package org.annotationmvc.vo;

import java.io.*;
import javax.persistence.*;

@Entity(access = AccessType.PROPERTY)
@Table (name="myobject")
public class MyObjectVO implements Serializable {

private int id,
private String name;
private String address;
private String email;
private String phone;

@Id (generate = GeneratorType.AUTO)
public int getId() {
return id;
}

@Column (length=100)
public String getName() {
return name;
}

@Column (length=100)
public String getAddress() {
return address;
}

@Column (length=30)
public String getEmail() {
return email;
}

@Column (length=15)
public String getPhone() {
return phone;
}

public void setXXX() {
....
}

}

Entity bean with annotations declared at variable/properties level

package org.annotationmvc.vo;

import java.io.*;
import javax.persistence.*;

@Entity(access = AccessType.FIELD)
@Table (name="myobject")
public class MyObjectVO implements Serializable {

@Id (generate = GeneratorType.AUTO)
private int id;

@Column (length=100)
private String name;

@Column (length=100)
private String address;

@Column (length=30)
private String email;

@Column (length=15)
private String phone;


public String getXXX()
{
....
}
public void setXXX() {
....
}

For this example, I have chosen to use the method-level annotations declaration.
As you can see from the previous example, every bound persistent POJO class is an entity bean and is declared by using the @Entity annotation. @Entity declares the class as an entity bean (in other words, a persistent POJO class). @Table declares the database table to which the class corresponds. This is optional. If you do not include this attribute, the default class name is used as the table name. @Id declares the identifier property of this entity bean. The other mapping declarations are implicit. The @Entity annotation also allows you to definewhether an entity bean should be accessed through its getters/setters methods or whether the entity manager should access the fields of the object directly.
Some Rules of Thumb for Defining an EntityIn short, here are the few actions you must not forget when defining an entity POJO:

* Import the javax.persistence.* packages to enable the JSDK5 annotation feature.
* At the class level, declare the @Entity and @Table to map an entity object with a table name.
* Define the access type for the POJO properties. access = AccessType.PROPERTY means that method level annotations are used, which will only be applied with the getter methods. If access = AccessType.FIELD were used, then the annotations would be associated with the fields. If AccessType is not mentioned, PROPERTY is used as the default type (in other words, for getter methods).
* For a primary key field, use the annotation @Id. There are five types of GeneratorType variables: AUTO, TABLE, IDENTITY, SEQUENCE, and NONE. Because this is a numeric type variable, you can also use SEQUENCE in this case, although the AUTO generator is the recommended type for portable applications.
* For the other properties in the POJO (in other words, not the primary field), the @Column annotation is used. Here is the syntax:

@Column(name = "address", nullable =
false, length=100, unique=false)

o name: (optional). This property refers to the corresponding table field name. If not mentioned, the default getter property is used.
o nullable: (optional). Whether null values are allowed or not for this field. Default is true.
o length: This determines the length of the field. Although optional, it is recommended to mention the field size.
o unique: (optional). This determines whether the property is unique or not. Default is false.

Create a Hibernate configuration file

The next step after defining the entity is to create a Hibernate configuration file. Start by creating a file named hibernate.cfg.xml and place it under the WEB-INF/ directory. The fully qualified class name of the POJO should be included within the tag (highlighted portion below). If more POJOs are created, they will simply be included with additional tags under the tag. (Click here to get the hibernate.cfg.xml.file.) The file should look like this

Hibernate configuration file:

hibernat.cfg.xml

"-//Hibernate/Hibernate Configuration DTD 3.0//EN"
"http://hibernate.sourceforge.net/hibernate-configuration-3.0.dtd">






The DAO interface
package org.annotationmvc.dao;

public interface MyObjectDao {

MyObjectVO findMyObjectById(int id);
void insertMyObjectVO(MyObjectVO myObjectVO);

}

The DAO implementation

package org.annotationmvc.dao;

import org.springframework.orm.hibernate3.support.HibernateDaoSupport;

public class MyObjectDaoImpl extends HibernateDaoSupport
implements MyObjectDao {

public MyObjectVO findMyObjectById(int id) {
List list=getHibernateTemplate().find("from MyObjectVO
where id=?",id);
return (MyObjectVO) list.get(0);
}

}

Spring bean configuration file :







oracle.jdbc.driver.OracleDriver


jdbc:oracle:thin:@localhost:1521:global

spring


spring



class="org.springframework.orm.hibernate3.
LocalSessionFactoryBean">


WEB-INF/hibernate.cfg.xml

org.hibernate.cfg.AnnotationConfiguration


org.hibernate.dialect.
Oracle9Dialect
create










another way to create hibernate session factory

class="org.springframework.jdbc.datasource.DriverManagerDataSource">

${connection.Driver}


${connection.url}


${connection.username}


${connection.password}




class="org.springframework.orm.hibernate3.annotation.AnnotationSessionFactoryBean">



com.optrasystems.model.BaseObject
com.optrasystems.model.Image

com.optrasystems.model.User
com.optrasystems.model.Category
com.optrasystems.model.Parameter
com.optrasystems.model.Object
com.optrasystems.model.Annotation
com.optrasystems.model.ShareImage
com.optrasystems.model.UserRole
com.optrasystems.model.Experiment
com.optrasystems.model.ImageFTPLocation
com.optrasystems.model.ImageNotes
com.optrasystems.model.MetaData
com.optrasystems.model.ThorImageExperimentXML




class="org.hibernate.cfg.DefaultComponentSafeNamingStrategy" />




${hibernate.dialect}


${hibernate.cache.use_second_level_cache}


${hibernate.cache.provider_class}


${hibernate.hbm2ddl.auto}


${hibernate.show_sql}

How HashMap put and get operations works in JAVA

Put operation

The put operation performs the following steps :
1. calculate hashcode for key
2. rehash it. lets call the rehashed results as h
3. calculate bucket index as h & (capacity -1)
4. now iterate over the bucket and compare key with all existing keys using equals()
5. if the key already exists, change the value of that Entry object
6. else create a new Entry object and add to the head of the linked list
7. increment mod count
8. resize if necessary

get operation

The get operation performs the following steps :
1. calculate hashcode for key
2. rehash it. lets call the rehashed results as h
3. calculate bucket index as h & (capacity -1)
4. now iterate over the bucket and compare key with all existing keys using equals()
5. if the key already exists return the corresponding value in the Entry object

Difference Between executeQuery() Vs executeUpdate() Vs execute() In JDBC

ResultSet executeQuery(String sql) throws SQLException :

This method is used for SQL statements which retrieve some data from the database. For example is SELECTstatement. This method is meant to be used for select queries which fetch some data from the database. This method returns one java.sql.ResultSet object which contains the data returned by the query.

int executeUpdate(String sql) throws SQLException :

This method is used for SQL statements which update the database in some way. For example INSERT, UPDATE and DELETE statements. All these statements are DML(Data Manipulation Language) statements. This method can also be used for DDL(Data Definition Language) statements which return nothing. For example CREATE and ALTER statements. This method returns an int value which represents the number of rows affected by the query. This value will be 0 for the statements which return nothing.

boolean execute(String sql) throws SQLException :

This method can be used for all types of SQL statements. If you don’t know which method to use for you SQL statements, then this method can be the best option. This method returns a boolean value. TRUE indicates that statement has returned a ResultSet object and FALSE indicates that statement has returned an int value or returned nothing.

Enable HDFS Short Circuit Reads

short-circuit reads bypass the DataNode, allowing a client to read the file directly, as long as the client is co-located with the data.

Add Below properties in hdfs-site.xml.


    dfs.client.read.shortcircuit  - true

    dfs.client.read.shortcircuit.streams.cache.size -  1000

    dfs.client.read.shortcircuit.streams.cache.expiry.ms -  10000

    dfs.domain.socket.path -   /var/run/hadoop-hdfs/dn._PORT