Spark Architecture and Deployment Environment

A spark application consists of a driver which run either on the client or on application master node and many executors which run across slave nodes in the cluster.

An application can be used for a single batch job, an interactive session with multiple jobs spaced apart, or a long-lived server continually satisfying requests. Unlike MapReduce, an application will have processes, called Executors, running on the cluster on its behalf even when it’s not running any jobs. This approach enables data storage in memory for quick access, as well as lightning-fast task startup time.

Job of Spark Driver

It is responsible for creating spark context, creating DAG, breaking the job into stages and task and scheduling of the task. It defines the transformations and actions applied to the data set.

At its core, the driver has instantiated an object of the SparkContext class. This object allows the driver to acquire a connection to the cluster, request resources, split the application actions into tasks, and schedule and launch tasks in the executors.

The driver first asks the application master to allocate resources for the containers on the worker/slave nodes and create executors process.

Once the executors are created the driver directly coordinates with the worker nodes and assign the task to them.

Job of Executor

An executor is a JVM process which is responsible for executing the task. many tasks can run in parallel in the executor.

MapReduce runs each task in its own process. When a task completes, the process goes away. In Spark, many tasks can run concurrently in a single process, and this process sticks around for the lifetime of the Spark application, even when no jobs are running.

The advantage of this model, as mentioned above, is speed, Tasks can start up very quickly and process in-memory data. The disadvantage is coarser-grained resource management. As the number of executors for an app is fixed and each executor has a fixed allotment of resources, an app takes up the same amount of resources for the full duration that it’s running. (When YARN supports container resizing, we plan to take advantage of it in Spark to acquire and give back resources dynamically.)

Cluster Deployment

The SparkContext can connect to several types of cluster managers (either Spark’s own standalone cluster manager, Mesos or YARN), which allocate resources for containers on which spark executors runs.

Spark acquires executors on nodes in the cluster, which are processes that run computations and store data for your application. Next, it sends your application code (defined by JAR or Python files passed to SparkContext) to the executors. Finally, SparkContext sends tasks to the executors to run.

Submitting Application (spark-submit)

The spark-submit script in Spark’s bin directory is used to launch applications on a cluster. It can use all of Spark’s supported cluster managers through a uniform interface so you don’t have to configure your application especially for each one.

./bin/spark-submit \
  --class <main-class> \
  --master <master-url> \
  --deploy-mode <deploy-mode> \
  --conf <key>=<value> \
  ... # other options
  <application-jar> \
  [application-arguments

• --deploy-mode: Whether to deploy your driver on the worker nodes (cluster) or locally as an external client (client) (default: client)

if your application is submitted from a machine far from the worker machines (e.g. locally on your laptop), it is common to use cluster mode to minimize network latency between the drivers and the executors.

Cluster Manager

Spark Standalone cluster

Standalone Master is the resource manager for the Spark Standalone cluster

Standalone Worker (aka standalone slave) is the worker in the Spark Standalone cluster .

Standalone cluster mode is subject to the constraint that only one executor can be allocated on each worker per application.

A client first connects to standalone master and ask for the resources from the standalone master and start the executor process on the worker node and the driver either on the client node itself or in one of the worker node.

Here the client act as the application master which is responsible for requesting the resources from the resource manager/standalone master.

Hadoop Yarn

YARN (Yet Another Resource Negotiator) is the resource management layer for the Apache Hadoop ecosystem.

YARN is a software rewrite that decouples MapReduce’s resource management and scheduling capabilities from the data processing component, enabling Hadoop to support more varied processing approaches and a broader array of applications.

The Application Master oversees the full lifecycle of an application, all the way from requesting the needed containers from the Resource Manager to submitting container lease requests to the NodeManager.

Each application framework that’s written for Hadoop must have its own Application Master implementation. Spark also has the implementation of application master.

Yarn Architecture

Yarn Vs Spark Standalone cluster

• YARN allows you to dynamically share and centrally configure the same pool of cluster resources between all frameworks that run on YARN. You can throw your entire cluster at a MapReduce job, then use some of it on an Impala query and the rest on Spark application, without any changes in configuration.
• Spark standalone mode requires each application to run an executor on every node in the cluster, whereas with YARN, you choose the number of executors to use.

Spark SQL, Dataframe and Dataset

 

Spark: Spark SQL, Dataframe and Dataset

Spark SQL

Spark SQL is a library for structured data processing which provides SQL like API on top of spark stack it supports relational data processing and SQL literal syntax to perform operations on data.

Like an RDD, a DataFrame and DataSet is an immutable distributed collection of data. Unlike an RDD, data in DataSet is organized into named columns, like a table in a relational database.

DataFrame allows developers to impose a structure onto a distributed collection of data, allowing higher-level abstraction, it also provides a domain specific language API.

RDD vs DataFrame

RDD is the core abstraction over which DataFrame and Dataset are built upon.

1. RDD is low-level and type-safe API.
2. RDDs are mainly for semi-structured and non-structured data but is easier to write inefficient code over RDD.

RDD operation

1. DataFrame has Column and schema and structure data.
2. Dataframe provides High-Level abstraction and provides DSL and query language to manipulate and extract data.

Dataset Operation

RDD vs DataSet GroupBy

Dataframe and Dataset have inbuilt optimizer which optimizes the data flow and is much faster and more memory efficient because of tungsten encoders.

DataSets

Dataset provides typesafe API which provides you the relational and functional transformation while benefitting from other optimization which RDD doesn't provide.

In Dataset you can mix and match your functional code/lambda and relational expression.

listingDS.groupByKey(ls => ls.zipCode).agg(avg($ “price”).as[Double])

Creating a DataSet From Json

val ds = spark.read.json(“/databricks-public-datasets/data/iot/iot_devices.json”).as[DeviceIoTData]

From RDD and DataFrame

myDF.toDS /rdd.toDS

DataSet Transformation:(Typed and Untyped)

Typed Transformation:

Typed Transformation on DataSet

GroupByKey On DataSet

Calling GroupByKey on Dataset returns another type of Dataset “KeyValueGroupedDataset”.

which has a special set of aggeration operation available over them which return the DataSet again.

groupByKey on DataSet

KeyValueGroupedDataset Aggeration Operations

KeyValueGroupedDataset Aggeration Operation which Returns Dataset Back

What to pass in TypedColumn?

Various functions are available such as avg, min, max, sum which can be pass as the TypedColumn argument.

agg(avg($"columnname").as[Double]) ---> Returns Dataset 

ReduceByKey In DataSet

ReduceByKey using reduceGroups

Aggregators

Aggregator is the contract for user-defined typed Custom Aggregate function.

Aggregator Function Contract

Implementing of Aggregate for string concatenate

Spark Basics : RDD ,Stages, Tasks and DAG

 

Spark Basics : RDDs, Stages, Tasks and DAG

Resilient Distributed Datasets (RDD) is a fundamental data structure of Spark. It is an immutable distributed collection of objects.

RDDs

RDD(Resilient,Distributed,Dataset) is immutable distributed collection of objects. RDD is a logical reference of a dataset which is partitioned across many server machines in the cluster. RDDs are Immutable and are self recovered in case of failure. An RDD could come from any datasource, e.g. text files, a database via JDBC, etc.

Creating an RDD

val rdd = sc.textFile("/some_file",3)  
val lines = sc.parallelize(List("this is","an example"))

the argument ‘3’ in the method call sc.textFile() specifies the number of partitions

Partitions

RDD are a collection of various data if it cannot fit into a single node it should be partitioned across various nodes. So it means, the more the number of partitions, the more the parallelism. These partitions of an RDD is distributed across all the nodes in the network.

RDDs Operations(Transformations and Actions)

There are two types of operations that you can perform on an RDD- Transformations and Actions. 

Transformation applies some function on a RDD and creates a new RDD, it does not modify the RDD that you apply the function on.(Remember that RDDs are immutable). Also, the new RDD keeps a pointer to it’s parent RDD.

Transformations are lazy operations on a RDD that create one or many new RDDs, e.g. map,filter, reduceByKey, join, cogroup, randomSplit

At high level, there are two transformations that can be applied onto the RDDs, namely narrow transformation and wide transformation. Wide transformations basically result in stage boundaries.

Narrow transformation — doesn’t require the data to be shuffled across the partitions. for example, Map, filter etc..

wide transformation — requires the data to be shuffled for example, reduceByKey etc..

By applying transformations you incrementally build a RDD lineage with all the parent RDDs of the final RDD(s).Transformations are lazy, i.e. are not executed immediately. Only after calling an action are transformations executed.

val rdd = sc.textFile("spam.txt")
val filtered = rdd.filter(line => line.contains("money"))
filtered.count()

sc.textFile() and rdd.filter() do not get executed immediately, it will only get executed once you call an Action on the RDD — here filtered.count(). An Action is used to either save result to some location or to display it. You can also print the RDD lineage information by using the command filtered.toDebugString(filtered is the RDD here).

RDDs can also be thought of as a set of instructions that has to be executed, first instruction being the load instruction.

New RDD is created after every transformation.(DAG graph)

DAG(Directed Acyclic Graph),Stages and Tasks

DAGScheduler is the scheduling layer of Apache Spark that implements stage-oriented scheduling. It transforms a logical execution plan (i.e. RDD lineage of dependencies built using RDD transformations) to a physical execution plan (using stages).

DAGScheduler Transforming RDD Lineage(DAG) Into Stage DAG(Physical Execution Plan)

As mentioned above, the DAG scheduler splits the graph into multiple stages, the stages are created based on the transformations. The narrow transformations will be grouped (pipe-lined) together into a single stage.

val input = sc.textFile("log.txt")
val splitedLines = input.map(line => line.split(" "))
                        .map(words => (words(0), 1))
                        .reduceByKey{(a,b) => a + b}

above example will create 2 stages.

Shuffled RDD is created by reduceByKey wide transforamtion

Shuffle Operation or wide transformation define the boundary of 2 stages.Stages are separated by 2 shuffle operations.

The DAG scheduler will then submit the stages into the task scheduler. The number of tasks submitted depends on the number of partitions present in the textFile. Fox example consider we have 4 partitions in this example, then there will be 4 set of tasks created and submitted in parallel provided there are enough slaves/cores.

The stages that are not interdependent may be submitted to the cluster for execution in parallel: this maximizes the parallelization capability on the cluster. So if operations in our dataflow can happen simultaneously we will expect to see multiple stages launched.

val sfi  = sc.textFile("/data/blah/input").map{ x => val xi = x.toInt; (xi,xi*xi) }
val sp = sc.parallelize{ (0 until 1000).map{ x => (x,x * x+1) }}
val spj = sfi.join(sp)
val sm = spj.mapPartitions{ iter => iter.map{ case (k,(v1,v2)) => (k, v1+v2) }}
val sf = sm.filter{ case (k,v) => v % 10 == 0 }
sf.saveAsTextFile("/data/blah/out")

Stage 0 and Stage 1 executes in parallel to each other as they are not inter-dependent.

Stage 2 (join operation) depends on stage 0 and stage 1 so it will be executed after executing both the stages.

The follow-on operations working on the joined data may be performed in the same stage because they must happen sequentially. There is no benefit to launching additional stages because they can not start work until the prior operation were completed

Delete Files Older than X Days and Delete Empty Folders

(1) Create a new text document (Del_OldDocs.txt)
(2) copy the script

REM Remove files older than X Days
forfiles -p "D:\Dir1\OldFolders" -s -m *.* /D -7 /C "cmd /c del @path"

cd /D    D:\Dir1\OldFolders
for /f "delims=" %%d in ('dir /s /b /ad ^| sort /r') do rd "%%d">NUL

(3) save the text document as Del_OldDocs.bat
(4) Run the script
(5) You can also run the script as job in the Task Scheduler


Note:
Please do some testing before using it in Production

D:\Dir1\OldFolders   -- this is the path where you have your old folders to be deleted
X Days -- Replace X with days (7 is used for one week), replace 7 with the number of days you want

First part of the script removes all the files in the OldFolders directory and Empty Folders still remain
Second part of the script will remove all the Empty Folders within OldFolders directory

Active Directory - Get account names, office and other properties from accountnames Powershell script

Using this script, you should be able to check and create Active Directory groups based on the input from spreadsheet (CSV)

1. Copy the script in a notepad and save it as 'AD_DisplayNames_to_AccountNames.ps1'
2. Use the same format for csv and add your DisplayNames and save it in the same folder as your script.
3. Both script and csv are saved in the desktop
4. Open powershell as Administrator (or ) Windows Powershell ISE that comes along with debugging. Using Powershell ISE can skip 5 thru 8 and open your script and hit F5
5. In the powershell Navigate to your desktop (where script and csv are present)
6. Set-ExecutionPolicy RemoteSigned
7. Click Yes
8. Now run the script by typing   ./AD_DisplayNames_to_AccountNames.ps1
9. check for newly added txt and csv files for other properties in AD for the given displaynames

Powershell Script:

# Import active directory module for running AD cmdlets

Import-Module ActiveDirectory
$Sam = @()
$names = Import-csv "C:\Users\\Desktop\AD_input_DisplayNames.csv"


foreach ($name in $names)
{

$dispName = $name.displayname
$sam += get-aduser -filter  {displayname -like $dispName } -Property SamaccountName,DisplayName,office -Server ad3.xyz.com| Select DisplayName,SamaccountName,office


}
write-host $sam

#output to a text file format
$Sam | out-file 'C:\Users\\Desktop\AD_AccountNames.txt'
#output to a csv file format
$Sam | Export-Csv -Path 'C:\Users\Desktop\AD_AccountNames.csv'  -NoTypeInformation


Input:  (AD_input_DisplayNames.csv)

displayname
DisplayName1
DisplayName2


Output:  (AD_AccountNames.csv)

"DisplayName","SamaccountName","office"
"DisplayName1","accountname1","Office Name "
"DisplayName2","accountname2","Office Name "

Bulk create Active directory groups and add users from csv - Powershell

Using this script, you should be able to check and create Active Directory groups based on the input from spreadsheet (CSV)

1. Copy the script in a notepad and save it as 'Add_ADGroups_Members.ps1'
2. Use the same format for csv and add your own groups and update csv based on your domain and path and save it in the same folder as your script.
3. Both script and csv are saved in the desktop
4. Open powershell as Administrator (or ) Windows Powershell ISE that comes along with debugging. Using Powershell ISE can skip 5 thru 8 and open your script and hit F5
5. In the powershell Navigate to your desktop (where script and csv are present)
6. Set-ExecutionPolicy RemoteSigned
7. Click Yes
8. Now run the script by typing   ./Add_ADGroups_Members.ps1
9. check your AD groups getting created along with members added

Note: Users Domain is different to domain where groups are being created. You can have the same domain too, depending on your environment

# Import active directory module for running AD cmdlets

  Import-Module ActiveDirectory

    $Users = Import-Csv -Path "C:\Desktop\AD_input.csv"  

            

    foreach ($User in $Users)            

    {           

        $GroupName = $User.'GroupName'            

        $samAccountName= $User.'samAccountName'       

        write-host Current AD Group Name is $GroupName and accounts to add are $samAccountName

       

             

            Try

            {

              #Check if the Group already exists

              write-host checking to see if group $User.GroupName exists

              $exists = Get-ADGroup $User.GroupName

              Write-Host "Group $($User.GroupName) already exists! skipping group creation!"

         

            }

            Catch

            {

              #Create the group if it doesn't exist

                $create = New-ADGroup -Name $GroupName -GroupScope $User.GroupType -Path $User.GroupLocation -Description $User.GroupDescription

                Write-Host AD Group $GroupName created!

       

          

            }

            

         # Parse prospective members and add each to the new group. (multiple users to be added to same group)

         $Members = $User.samAccountName.Split(";")

      

         ForEach ($Member In $Members)

         {

             write-host Adding member  $Member

             $DomainGroupDN = Get-ADGroup -Identity $GroupName -Server ou.ad3.abc.com

             $SamDN = Get-ADUser $Member -Server ad3.abc.com

             Add-ADGroupMember $DomainGroupDN -Server ou.ad3.abc.com-Members $SamDN

             

             $Member = ""

         }

   

     Write-Host Members $Members added to Group $GroupName

}

Spreadsheet:   (save it as *.csv)

GroupName,GroupDescription,GroupType,GroupLocation,samAccountName

Test1,TestDescription1,DomainLocal,"OU=TESTOU,DC=ou,DC=ad3,DC=abc,DC=com","ramii;mreck"

Test2,TestDescription2,DomainLocal,"OU=TESTOU,DC=ou,DC=ad3,DC=abc,DC=com","larzi;mkhill"

Install ADUC on Windows 7 or Windows 10

Remote Server Administration Tool (RSAT) can be downloaded here
https://www.microsoft.com/en-us/download/confirmation.aspx?id=7887

(1) Download 64 bit or 32 bit depending on OS version installed on your computer

(2) After installing - to enable AD Management Tools

1. From the Control Panel, click on Programs.
2. Under Programs and Features, select Turn Windows features on or off.
3. Under Remote Server Administration Tools > Role Administration Tools, select AD DS and AD LDS Tools

Click Ok. This will take a minute for windows to make changes to features

SharePoint - Scroll To Top for Large Pages

Add this html to your SharePoint Page

<p id="page-top"><a href="#nav">Back to Top</a> </p>

Add below Style and Scripts to CEWP on your page.
 <script src="....//jquery-1.11.0.min.js"></script>

 <style>

#page-top {
                 position: fixed;
                bottom: 20px;
                right: 50px;
       }

#page-top a
{
background: #666;
text-decoration: none;
color: #fff;
width: 100px;
padding: 8px 0;
text-align: center;
display: block;
border-radius: 3px;
 
}

#page-top a:hover
{
text-decoration: none;
background: #999;
}

</style>


 <script>          
$(document).ready(function() {

//var scrollPosition = document.getElementById("s4-workspace").scrollTop;
//alert("Scroll Position: " + scrollPosition);
//$('#s4-workspace').animate({scrollTop:0}, 'fast');


var topBtn = $('#page-top');
                                 topBtn.hide();
$('#s4-workspace').scroll(function () {

   if ($(this).scrollTop() > 400) {
  topBtn.fadeIn();
} else {
  topBtn.fadeOut();
                    }
                 });


               topBtn.click(function () {
                $('#s4-workspace').animate({
                  scrollTop: '0px'
                 }, 500, "swing");
                return false;
           });





});
</script>

//(Tested in IE10, Firefox, Chrome)