Tuesday, July 16, 2024

Spark submit on YARN - Execution flow


When a Spark job is submitted to YARN, the following steps occur:

1. The client submits the job to the YARN ResourceManager (RM) via the YARN client.

2. The RM allocates a container for the ApplicationMaster (AM) on one of the cluster nodes. The AM is responsible for managing the lifecycle of the Spark application on YARN.

3. The AM starts on the allocated container and registers itself with the RM.

4. The AM negotiates resources with the RM to allocate containers for Spark executors.

5. The RM assigns containers to the AM, which in turn launches the Spark executors on these containers.

6. The executors register themselves with the AM.

7. The AM submits tasks to the executors and monitors their progress.

8. The executors execute the tasks and report the results back to the AM.

9. The AM aggregates the results and updates the job status.

10. Once the job is completed or terminated, the AM releases the resources and notifies the RM.


The ApplicationMaster is responsible for coordinating the execution of the Spark application on YARN. It negotiates with the RM to allocate resources (containers) for the Spark executors and manages their lifecycle. It also monitors the progress of the executors and handles failures by requesting the RM to re-allocate containers if needed.


The ApplicationManager is a component within the RM that receives requests from clients to submit applications. It is responsible for accepting application submissions, allocating resources to the applications, and tracking their status. The ApplicationManager works closely with the ApplicationMaster to manage the execution of Spark applications on YARN.


The relationship between the ApplicationMaster and the executors is that the AM is responsible for launching and managing the lifecycle of the executors. It negotiates with the RM to allocate resources (containers) for the executors and monitors their progress. The executors register themselves with the AM and execute the tasks assigned by the AM. The AM communicates with the executors to collect the results and update the job status.

 

HDFS - Edit logs and FsImage

 HDFS - Edit logs and FsImage

What is DistCp? - It is a tool that is used for copying a very large amount of data to and from Hadoop file systems in parallel

Default replication factor? By default, the replication factor is 3. There are no two copies that will be on the same data node. Usually, the first two copies will be on the same rack, and the third copy will be off the shelf. It is advised to set the replication factor to at least three so that one copy is always safe, even if something happens to the rack.

The two types of metadata that NameNode server stores are in Disk and RAM.

Edit logs: all transaction logs like added block, deleted blocks, replication details after last fsimage file created. The updates are periodically merged to FSImage.

FsImage: file stored in OS system and contains the complete directory stucture of hdfs. Name node reads when starting to memory


Spark execution optimisers - Catalyst and Tungsten


Spark uses two engines to optimize and run the queries - Catalyst and Tungsten, in that order. Catalyst basically generates an optimized physical query plan from the logical query plan by applying a series of transformations like predicate pushdown, column pruning, and constant folding on the logical plan. 

This optimized query plan is then used by Tungsten to generate optimized code, that resembles hand written code, by making use of Whole-stage Codegen functionality introduced in Spark 2.0. This functionality has improved Spark's efficiency by a huge margin from Spark 1.6, which used the traditional Volcano Iterator Model 

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

 




Monday, December 13, 2021

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.


Wednesday, July 21, 2021

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




Monday, July 2, 2018

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}