Not Aldıklarım ...

As you may recall that I have started to publish my Terraform notes for Azure while I am experiencing it as future another obsolete reference :) Please refer my first introductory article if you haven't read the basics of infrastructure as code - IAC

Purpose of this article is to teach you basic concepts like variables, modules, state files through very basic provisioning of a virtual server from the Azure environment.

Variables & Modules:

Terraform stores bits and pieces of infra in *.tf files where you might have only one long tf file or multiple *.tf files to manage a large number of assets and for flexibility and modularity. You can use representative names for above-mentioned tf files such as securityGroups.tf, servers.tf, resourceGroups.tf, resourcemanaegrs.tf etc. All of these files combined into one large tf file behind the scenes once terraform command such as plan, apply or destroy is called.

The proper naming convention will give you a clear overview of the setting and objects to be created. As you can see in the above image, various settings and resources are stored in different files. For example, variables are stored in variables.tf, where "azure_region" is a variable and its default value, is set to "North Europe". Now see the below snippet that uses a variable defined in another file (variables.tf):

resource "azurerm_public_ip" "public_ip_for_prototypeVM" {  name                         = "public_ip"  location                     = "${var.azure_region}"  resource_group_name          = "${azurerm_resource_group.azure_resource_group.name}"  allocation_method            = "Dynamic"  tags = {    environment = "Public Ip Azure Demo"  }

The code above (ipAddresses.tf) uses ${var.xxxxx} definition from variables.tf file and variables stored in this separate file can be reached from any tf file stored under the project folder.

location = "${var.azure_region}"

State Files:

State files such as terraform.tfstate keeps track of the id's of created resources to manage in later stages. Remember that state file might contain sensitive information such as plain passwords, secrets, connection strings, tenant id's and more so MUST NOT be committed to terraform the main repository.

Now is time to complete the mission: " spin a virtual server in the Azure Cloud" Here are the basic prerequisites:

Network/Sub Networks with private/public IP addresses.

Basic Security Group to control in/out traffic.

Network Interface - NIC to assign to our VM.

Let's start with the first item network and subnetwork as follows:

resource "azurerm_virtual_network" "azureVPC" {  address_space = ["10.0.0.0/16"]  location = "${var.azure_region}"  name = "azureVPC"  resource_group_name = "${azurerm_resource_group.azure_resource_group.name}"  dns_servers = ["10.0.0.4","10.0.0.5"] } resource "azurerm_subnet" "subnetOne_for_AzureVPC" {  address_prefix = "10.0.1.0/24"  name = "subnetOne"  resource_group_name = "${azurerm_resource_group.azure_resource_group.name}"  virtual_network_name = "${azurerm_virtual_network.azureVPC.name}" } resource "azurerm_subnet" "subnetTwo_for_AzureVPC" {  address_prefix = "10.0.2.0/24"  name = "subnetOne"  resource_group_name = "${azurerm_resource_group.azure_resource_group.name}"  virtual_network_name = "${azurerm_virtual_network.azureVPC.name}"

}

The above code is self-explanatory: it creates a new virtual network called "azureVPC' with address space of 10.0.0.0/16 under main resource group. Similar to the virtual network, subnetworks are created accordingly subnetOne and subnetTwo. Below is our basic security group:

resource "azurerm_network_security_group" "security_group_standartPorts" {  name                = "standartPorts-SSH-Web"  location            = "${var.azure_region}"  resource_group_name = "${azurerm_resource_group.azure_resource_group.name}"  security_rule {    name                       = "SSH"    priority                   = 1001    direction                  = "Inbound"    access                     = "Allow"    protocol                   = "Tcp"    source_port_range          = "*"    destination_port_range     = "22"    source_address_prefix      = "*"    destination_address_prefix = "*"  }

}

Our security group only contains one inbound rule for the SSH port 22 that accepts any port and IP as a source address. Now it is time to define our NIC but before we need an IP address to assign it:

resource "azurerm_public_ip" "public_ip_for_prototypeVM" {  name                         = "public_ip"  location                     = "${var.azure_region}"  resource_group_name          = "${azurerm_resource_group.azure_resource_group.name}"  allocation_method            = "Dynamic"  tags = {    environment = "Public Ip Azure Demo"  }

}

it is dynamically allocated IP address managed under same Azure resource group. Time to create a NIC and assign our fresh IP address to it:

resource "azurerm_network_interface" "interface1" {  name                = "interface1"  location            = "${var.azure_region}"  resource_group_name = "${azurerm_resource_group.azure_resource_group.name}"  network_security_group_id = "${azurerm_network_security_group.security_group_standartPorts.id}"  ip_configuration {    name                          = "myNicConfiguration"    subnet_id                     = "${azurerm_subnet.subnetOne_for_AzureVPC.id}"    private_ip_address_allocation = "Dynamic"    public_ip_address_id          = "${azurerm_public_ip.public_ip_for_prototypeVM.id}"  }

}

Our NIC called interface1 uses public IP address created in the previous step. Now showtime: Ask for our first VM:

resource "azurerm_virtual_machine" "demo_VM" {  name                  = "DemoVM"  location              = "${var.azure_region}"  resource_group_name   = "${azurerm_resource_group.azure_resource_group.name}"  network_interface_ids = ["${azurerm_network_interface.interface1.id}"]  vm_size               = "Standard_B1s"  storage_os_disk {    name              = "mainDisk"    caching           = "ReadWrite"    create_option     = "FromImage"    managed_disk_type = "Standard_LRS"  }  storage_image_reference {    publisher = "Canonical"    offer     = "UbuntuServer"    sku       = "16.04.0-LTS"    version   = "latest"  }  os_profile {    computer_name  = "DemoVM"    admin_username = "azureuser"  }  os_profile_linux_config {    disable_password_authentication = true    ssh_keys {      path     = "/home/azureuser/.ssh/authorized_keys"      key_data = file("~/.ssh/id_rsa.pub")    }  }

}

Our Linux VM is very tiny as Standart B1s type with Standard main disk. Don't forget you can separate image references, os related information into a centeralized file for further flexibilit.y you can find the list of VM types and their details at:

https://docs.microsoft.com/en-us/azure/virtual-machines/windows/sizes-general

Our tiny ubuntu uses ssh key (read from a local system as a file("~/.ssh/id_rsa.pub") ) to login rather password-based authentication. I am sure that you know to generate a key as (MAC) or use Putty for MS. Windows systems:

ssh-keygen -t rsa -b 4096

Time to see everything is ok and you already know which command to run!

terraform plan

See if there aren't any errors. Fix them and be ready to spin your first VM with:

terraform apply

I assume you reply "yes" after the above command. You can now see the created resources:

Apply complete! Resources: 8 added, 0 changed, 0 destroyed.

Find out your IP address and login to your new Linux using ssh as:

ssh [email protected]

Congratulations! Your first VM is ready now enjoy but delete all components for additional charges (terraform destroy)

Let me have some break and continue with another article such as deploy dockerized container or k8s using Terraform.

For the last decade, majority of companies adopted or built their systems (development, production, test and more) by practicing Infrastructure as code - IAC; managing and provisioning of the various resources such as networks, VMS, load balancers, etc. via definition/configuration files. Ansible, Puppet, Chef, Terraform, AWS CloudFormation are all used for the same purpose: From manual configurations/procedures to scripted error-proof automation for reducing configuration drifts, inconsistencies and human errors meanwhile reducing time to production.

In this article, I would like to share my notes for the newbies to learn the basics of the IAC using Terraform. I am sure that there are strengths and weaknesses of all the tools mentioned above but I personally found that Terraform is easy to learn, platform independent, fast, support for revisions and most importantly ability to generate immutable infrastructure (clean servers every time to avoid configuration drift), and prefer orchestration to repetitive configuration. But managing larger state files (tfstate) and merging them can be very challenging. I will be following only one cloud platform: Azure for resource provisioning and infrastructure design but remember that you have the same options for all cloud providers such as AWS and GCloud. Procedures are almost same but few differences due to terminology and the underlying architecture of various providers. Considering that I am an AWS certified cloud architect; this series also my notes for the Azure environment. (The reason is my current client uses Microsoft Azure)

Cut to the chase and roll the wheels.(Careful; bold texts are commands to be used in the terminal window or line to add your configuration files) We will start building a very basic network as a first step. Please see below:

Install Terraform: google it! Basically, you download Terraform zip, extract it and set an environmental variable for example for Mac edit profile as vim ~/.bash_profile and add:

export PATH=$PATH:/Users/yourname/TerraformBinaryFolder

Don't forget to source your new PATH variable using source ~/.bash_profile I suggest you keep the zip under the same folder to see the currently installed version. You can also see by entering:

terraform version

You need to install Azure CLI (similar in AWS, Gcloud) to set up your Azure credentials, secrets, client and tenant id. The easiest way for mac is to use brew as:

brew update && brew install azure-cli

and test it by login-in using az login in your terminal window. Now it is time to gets our hands dirty. To create a Terraform project ??? Yes, we call our first project as a Terraform project because it will contain configuration and connection and other files to be edited near soon. Thus it might be wise to use code repository like Github to store your project assets also remember that Github has a nice boilerplate future for Terraform files. So you already know it (or google it again) Goto Githup create a new repo but remember to select Terraform as below to add some of the configuration files to be ignored by Github (of course you can change manually if you forget to select)

Now clone your Terraform repo to your local directory to work with your favorite editor that is intellij IDEA for my case and open your local project folder via GUI or by idea . in your terminal window. Edit your .gitignofile to exclude some of the special files and folders

#  Local .terraform directories **/.terraform/* # .tfstate files *.tfstate *.tfstate.* # .tfvars files *.tfvars # Module directory .terraform/ .DS_Store .idea

Add a new file called connections.tf to configure Azure connection parameters, leave parameter values as it is, later we will be substituting with Azure credentials/. this is only required to fulfill the basic requirements.

provider "azurerm" { subscription_id="dummy" client_id="0" client_secret="passworddummy" tenant_id="1001" }

Now we can go terminal screen and run the terraform init command under the project directory where connection.tf is read and proper files are automatically downloaded based on the provider keyword "azurerm" Please check that you have now a hidden folder namely .terraform that contains required binaries and MUST NOT be committed to repo (added to gitignore above)

We are now ready to get our Azure subscription and related client, tenant ids and secret. Use your terminal window and enter

az account list

must output screen like below:

FIRATs-MacBook-Pro:Code firatdogan$ az account list [  {    "cloudName": "AzureCloud",    "id": "this is your subcription id  a1f532f0-529bf5dc-xxxxxx",    "isDefault": true,    "name": "Life time Free",    "state": "Enabled",    "tenantId": "this is your tenant id  bv1f532f0-529bf5dc-xxxxxx",    "user": {      "name": "[email protected]",      "type": "user"    }  } ]

Use the id above and type (terminal window again) :

az account set --subscription="this is your subcrip id32f0-529c-xxxxx" az ad sp create-for-rbac --role="Contributor" --scopes="subscriptions/you id here again"

must output your app, tenant subscription ids and your secret (password) Our aim is to use these values as our environmental variables and read them to our previously created connection.tf file. Thus edit your .bash_profile file and add the following lines:

export TF_VAR_subscription_id="this is your id " export TF_VAR_client_id="app id" export TF_VAR_secret="password" export TF_VAR_tenant_id="tenant id"

Don't forget above ids are removed due to security concerns replace with your own values. Now we can re-edit our connection.tf file and read environmental variables as follows:

variable "subscription_id" {} variable "client_id" {} variable "secret" {} variable "tenant_id" {} provider "azurerm" { subscription_id="${var.subscription_id}" client_id="${var.client_id}" client_secret="${var.secret}" tenant_id="${var.tenant_id}" }

All set ! let see everything works so far by entering:

terraform plan

Now time to configure some resources let start with the network by adding resources.tf (add new file to your project) file as follows:

resource "azurerm_resource_group" "test_network" { location = "North Europe"  name = "Devresourcegroup" } resource "azurerm_virtual_network" "azureVPC" {  address_space = ["10.0.0.0/16"]  location = "North Europe"  name = "azureVPC"  resource_group_name = "${azurerm_resource_group.test_network.name}"  dns_servers = ["10.0.0.4","10.0.0.5"]  subnet {    address_prefix = "10.0.1.0/24"    name = "subnetOne"  }  subnet {    address_prefix = "10.0.2.0/24"    name = "subnetTwo"  } }

Run terraform plan to see no typos or logical errors exists. You must now see the very first picture in this article that shows your future network to be created. It is now time to generate it:

terraform apply Apply complete! Resources: 2 added, 0 changed, 0 destroyed.

Answer yes and go! you congratulations you have now added your first resources via terraform orchestration.

I will be sharing with you how to add other resources to our network probably near soon.

Currently, I am playing with Azure Pipelines and it is good to share my notes for my self and others as a reference. (know that soon will be obsolete :)

First of all, there is no language/version control/Microsoft dependency for Azure DevOps/Pipelines means that you can use any programming language such as Java, Python, .net, or git, TFS, etc. This is also valid for the infrastructure and other cloud providers you don’t need to stick only Microsoft stack but have an option to use AWS, IBM or others. Majority of tools such as Nuget, npm, Maven, Gradle are also supported by the Azure Pipelines.

STAGE(s) are logical entities such as QA, Prod, Staging, Pre-Prod that are used to deploy a release and mostly independent from each other. However, App Db and Web tiers should be modeled as a single stage.

TASK(s) are building blocks for defining automation in a release: packed scripts or procedures. Tasks can be versioned for future reference and for the reuse. You can simply think that tasks are parts of a stage when combined together for example :

Task 1 can be “Get source from Github”

Task 2 can be a Terraform script that generates Infrastructure as Code on Azure

Task 3 deploys a web application to different environments

Task 1 + Task 2 + Task 3 = Stage

There are out of box Tasks available in Azure Pipelines’ so you don’t have to write scripts for simple/standard tasks; pick one from the list or pay for the one in the marketplace.

As you can see in the above example tasks can be TRIGGERed based on the rules such as commit a new piece of code to the repo or in different ways.

Below you can see very basic CI pipeline that is triggered by the new commit and executed unit test cases for a basic Java - Spring application (it is really a toy project / experimental )

It seems like I need more space for this entry for the CD part, so stop here and leave the remaining terminology like Deployment Triggers, Azure Resource Manager (ARM), ARM Templates and Build Agents for a next one if I can :)

As developers we know that story doesn’t end once we write and the build the project on our local environment. All of us familiar with very well-known quote; “it works on my machine. We require stand-alone servers for test/dev/prod, update and patch the OS, install pre-requisites for the application as JVM, Python .Net, configure servers, load balancers and auto scale features, monitor and maintain the servers continuously and eventually start over when there is a new release.

In Serverless you just need to deploy the application and monitor it! That’s all. So, no more server maintenance, no scaling and load balancing configuration and additional costs of high availability plus zero cost if there is no traffic to your app. Mobile - web backend, data processing, machine learning pipelines, streaming analysis are potential usage of Serverless approach. Again, there are servers behind the scenes right J But you don’t manage them and only use Serverless platform’s components provided and maintained by a cloud vendor.

One of the biggest Serverless platform provider is AWS with very well-known components as Lambda, S3, DynamoDB, API Gateway, SNS, SES, Kinesis and more. All of above listed services can be used to develop and host Serverless application. (mostly stateless)

Lambda is the key component of the AWS Serverless architecture. Any Lambda function can be triggered from other AWS or from AWS CLI. Functions runs in isolated environment, kept for a while for consecutive calls but runs in Ephemeral (non solid) containers: items stored in memory/disk might not be accessible for the next call. Here AWSs DynamoDB plays very important role for designing stateless applications. Since you can rely on DynamoDB to store and retrieve data from Lambda functions calls.

PMML -> predictive modeling markup language is the solution if you need to decouple modeling (producing/building) and prediction (consuming/scoring) parts. Imagine that you have worked in one environment let say used scikit learn for the model generation but would prefer to score your model in Spark Mlib or vice versa. For such cases, you need to define/export your model in a common format that can be used in prediction phase. Almost all of the machine learning libraries allow us to export our models in PMML (XML) and use this model in a different or same library. This is not only an interchange format, but it will also give us the flexibility of real-time model updates even if used in same environments.

Here is the pseudo like the simple line:

yourmodelObject.toPMML("/pathtofile")

Aws’nin dünya çapında dağıtım networkünün yani CDN (Content Delivery Network) ‘ in adı. Örneğin USA’ den bir kullanıcı herhangi bir şirket domainine ulaşmak istediğinde, DNS (Route 53) bu kullanıcıyı dosyaların saklandığı sunucudan – Orijin sunucu (örneğin Avrupa) daha farklı bir sunucuya yönlendirir. (örneğin USA’ de bir sunucu) Böylece kullanıcı içeriğe daha az gecikme ile sahip olabilir. Bu ek sunucu noktalarına Edge lokasyonu adı verilir. Bu o bölgeden yapılan ilk ziyaret ise, orijin sunucudaki dosyalar optimize edilmiş hızlı network üzerinden önce Edge noktasına ardından kullanıcıya aktarılır. Bu ilk yedek depolama adımından sonra aynı bölgeden gelen isteklere Edge lokasyonundan çok daha hızlı biçimde cevap verilebilir. Edge sunucularının sadece Aws networkünde bulunan içerik sağlayıcılar olması zorunluluğu yoktur. AWS dağıtım ağında bulunmayan başka sunucular da bu CDN ağına dahil edilerek içerik sağlayıcı olarak kullanılabilir. CloudFront http veya https üzerinden hizmet sunar, statik ve dinamik sayfaların iletilmesin destekler. Özellikle resim, video gibi geniş bant ihtiyaçları için http veya RTMP stream gerçekleştirebilir.

CloudFront bir dağıtım - “distribution” oluşturmak ile başlar. Aws bu dağıtım için size genellikle rastgele karakterlerden oluşan bir link üretir. Bu linki Route 53 veya farklı bir DNS yönetimi üzerinden daha kısa ve anlamlı isimlere dönüştürmek mümkündür.   Ardından merkez – “Origin” tanımı ile dosyaların Edge lokasyonlara taşınacağı kaynak belirlenir. Bu kaynak AWS de yayınlanan bir S3 ovası – bucket, EC2 instance’ı veya http hizmet veren bir sunucu olabilir. Son adımda ise (Cache Control) kaynaktan edge taşınan içeriğin burada ne kadar saklanacağını belirlemek gerekir. Varsayılan yedekleme zamanı 24 saattir. Cache kontrol Header’ ları ile bu süreyi değiştirmek veya API üzerinden “Invalidation”  çağrısı ile yedek kopyanın güncelliğini yitirmesi sağlanabilir. (Bu durumda biraz daha fazla para ödemeniz gerekir) Invalidation yerine uygulamanın abc.com/v1, abc.com/v2 seklinde versiyonları oluşturulduğunda Edge noktalarının güncellemesi otomatik olarak gerçekleştirilir.

Eğer “tüm internet kullanıcılarının” ulaşmasını istemediğiniz özel içeriklere sahipseniz; bu içerikleri:

·         Zaman aralığı veya IP sınırlaması olan işaretli URL –“Signed URL”

·         Public & Private anahtar kullanan “Signed Cookie”

·         OAI (Origin Access Identification) ile belirli Aws hesaplarının

Bu yazımızda Key/Value çiftleri ile çalışmayı göstermek için basit bir örnek üzerinden ilerleyelim.   Elimizde No, isim, yaş ve harcama tutarları şeklinde virgüller ile ayrılmış aşağıdaki biçimde bir dosyamız olsun:

1, Fırat, 38, 3000 2,Murat,40,2000 3,Çağla,39,2000 4,Melis,38,7000

Bu veri seti için her yaş için ortalama harcama tutarını hesaplayan bir Spark çözümü için, önce dosyayı okuyup split etmemiz gerekecektir.

Lines=sc.textFile(”file://Data/harcamalar.csv”) Rdd=lines.map(lambda x: (x.split(‘,’)[2] , x.split(‘,’)[3])) #2. Eleman yaş 3. Eleman ise ortalama harcama tutarıdır. Split sonucu ilk değerin indeksi 0 dır. 1 nolu indeks=isimdir örneğin Fırat.

Böylece yaşın Key value’ nun ise harcama miktarı olduğu bir Rdd üretmiş olduk. Rdd’ miz:

38,3000 40,2000 39,2000 38,7000

Şimdi her harcama tutarından kaç tane olduğunu sayabilmek için mapValues kullanarak her bir value için (3000,2000,…) sabit “1” (bir) değerini oluşturacağız. Dikkat edin keylere hiç dokunmadık. mapValues ile sadece harcama tutarlarının (yani values) yanlarına “1” koyarak bir çift (tuple) oluşturduk. Ardından elimizdeki değerleri Key’ lere göre azaltmak için reduceByKey kullanarak, aynı Key’ e sahip değerler üzerinde toplama işlemi gerçekleştireceğiz. Böylece aynı Key’ e sahip olan 2 satırın ilk (harcama tutarı) ve ikinci elemanlarını (hepsi 1 olan)  toplayıp aynı key altında indirgeyebildik:

ToplamYasaGoreHarcamalar=Rdd.mapValues(lambda x: ( x,1)).reduceByKey(lambda x,y : (x[0]+y[0] , x[1]+y[1]))

Sırayla çağırılan transformasyonların çıktılarına detaylı bakalım.                                 

Son adımda yaş’a göre ortalamayı mapValues ile hesaplayabiliriz.

ortalamaHarcamaTutar= ToplamYasaGoreHarcamalar.mapValues(lambda x : (x[0] / x[1] ))

mapValues kullandığımız için Key’ lere dokunulmadı (Böylece daha hızlı hesaplama yapılabildi) mapValues adında da anlaşılacağı üzere sadece valuelar üzerinde çalışır ve onları yeniden map eder. ReducebyKey ile elde edilen 38, (10000,2) için mapValues x olarak (10000,2) tuple’ı üzerinde çalışarak (x[0]=10000 ve x[1]=2)  10000/2 işlemini gerçekleştirir. Bu durumda aşağıdaki sonuç RDD’ sini elde ederiz:

(38,5000) -  (40,2000) - (39,2000)  (Fazla yer tutumasın diye yan yana yazdık)

Yine önemli ve bildiğiniz bir noktanın altını çizelim. Yukarıdaki kod bloklarında sadece dönüşümler yaptık, bir aksiyon olmadan Spark RDD’ leri işlemeye başlamaz. Tek düğüme sığabilecek verilerde collect() veya take(5) gibi bir aksiyon ile dönüşümlerin ve hesaplamaların gerçekleşmesini sağlayabiliriz. Aşağıda sonuçları nasıl ekrana yazdırdığımı görebilirsiniz:

sonuclar= ortalamaHarcamaTutar.collect() for sonuc in sonuclar            Print sonuç  

Bir Sonraki yazımızda Key/Value çiftleri ile ilgili bir örnek daha yapacağım. Ben yazmadan önce denemek ister misiniz ? O halde örneğimizi şimdiden yazalım: Aşağıda verdiğim veri dosyası için 5 den fazla kitap okuyanların yaşları ve ortalama okumalarını raporlayabilir misiniz? (no, isim, yaş, okunan kitap sayısı)

1,Fırat, 38, 3 2,Çağla,39,6 3,Leyla,7,14 …..

Spark’ da çoğu operasyon için verilerin temizlenerek (Spark ETL) key/value (anahtar/değer) şeklinde veri yapısına dönüştürülmesi gerekmektedir. Böylece aynı anahtara ait verileri gruplayabilir, örneğin bir ürüne ait yorum sayısını hesaplamak gibi bir işlemi gerçekleştirebiliriz. Bölümleme (partitioning) sayesinde düğümler arasındaki iletişim ihtiyacını azaltabilir ve daha hızlı hesaplamalar yapılmasını sağlayabiliriz. Örneğin reduceByKey() aynı anahtara sahip verileri işler, join() iki farklı RDD’ yi aynı anahtar üzerinden gruplayabilir. Map fonksiyonu ile key/value çiftleri oluşturmak mümkündür.

Örneğin (python):

ciftler=satirlar.map(lambda x: (x.split(“ “)[0],x))  

 ile her satırdaki ilk kelimeyi key, satırın tamamını ise value olarak aldık.

 Aynı örnek Scala’ da

çiftler=satirlar.map(x=>(x.split(“ “)(0),x))

 Java’ da çift (tüple) türü standart kütüphanelere dahil değildir. Bu nedenle Scala.Tuple2 sınıfının kullanılması gerekir. Tuple2(eleman1,eleman2) şeklinde kullanılır ve ._1() , ._2() metodları ile elemanlara erişilir. Java’ da geliştirme yapılırken Map yerine mapToPair() kullanılmalıdır:

PairFunction<String, String, String> keyData= new PairFunction< String, String, String>() {

           public Tuple2< String, String> call(String x) {

                       return new Tuple2(x.split(“ “)[0],x );

}

};

Bu fonksiyonu çağırmak için :

JavaPairRDD< String, String>  çiftler=satirlar.mapToPair(keyData);

Benzer şekilde In-Memory olarak pair oluşturmak için Python ve Scala’ da

SparkContext.parallelize(); Java’ da ise  SparkContext.parallelizePairs() fonksiyonlarını çağırmak gereklidir. Standart RDD’ ler üzerinde gerçekleştirilebilen tüm dönüşümler key/value türünde RDD’lerede uygulanır. Bu durumda tek fark “sadece bir değer” üzerinde çalışan fonksiyonlar yerine tuple’lar üzerinde çalışabilecek fonksiyonlara ihtiyaç duyulmasıdır.

ÖNEMLİ: Eğer key’ leri değiştirmeyecek isek (modify) key/value oluştururken map yerine mapValues() veya flatMapValues() kullanmak çok ciddi performans artışları sağlayacaktır. Bu durumda Spark partition’ ları karıştırmadan (Shuffle) aynı düğüm üzerinde tutmaya çalışır.

Bir Sonraki yazımızda Key/Value çiftleri ile çalışabileceğimiz basit bir örnek ile konunun pekişmesini sağlayacağız. Hatta isteklerinize cevap verecek biçimde daha fazla uygulama ile yolumuza devam edeceğiz.

RDD’ ler üzerinde gerçekleştirilecek aksiyonların sürekli tekrarlanması performans sorunlarına neden olabilir. Bu nedenle persist() kullanarak RDD’ nin düğümler (nodes) üzerinde saklanmasını sağlayabiliriz. Spark farklı seviyelerde cache (persistance) mekanizmaları sunar. Bunlar

MEMORY_ONLY, MEMORY_ONLY_SER, MEMORY_AND_DISK, MEMORY_AND_DIS_SER, DISK_ONLY flag’ leri (belirteç/opsiyon/bayrak) ile ayarlanabilir.

Örneğin Scala da:

val result=input.map(x=>x*x) result.persist(StorageLevel.DISK_ONLY) println(result.count()) println(result.collect().mkString(“,”))

mkString(“,”) ile kolleksiyonu aralarına virgüller koyarak stringe dönüştürdük..

Yukarıdaki koda dikkat edecek olursak daha ilk aksiyon olan count() kullanmadan RDD üzerinde persist() çağrısı yaptık. RDD’ nin kendi üzerinde uygulanan persist() çağrısını gerçekleştirdik. RDD’ lerin kendi üzerinde yapılan bu persist çağrısı ile hesaplama gerçekleştirilmez.

Eğer çok fazla veriyi hafızada saklamak isterseniz, Spark en az kullanılan (LRU-last recently used) verileri kullanmaz ve yeni veriler için hafızada yer açmaya çalışır. Bu şekilde büyük verilerin programı durdurması engellenir. ANCAK; bu işlem için diskin kullanılması hesaplamaların yavaşlamasına neden olabilir.

unpersist() ile cache’ e alınmış bilgiyi düğüm hafızalarından silebiliriz. (Persist-> Sevgili Spark Diske Yazabilirsin, Cache->Lütfen Sadece Hafızayı (RAM) kullan)

 Bir sonraki yazımda en önemli konumuz olan KEY/VALUE çiftleri ile çalışmayı anlatmayı planlıyorum.

Aksiyonları RDD’ ler üzerinde hesaplamalar yapmak için kullanılır. En sıklıkla kullanılan aksiyon reduce() iki eleman üzerinde çalışarak aynı türde yeni bir değer üretir. Örneğin + gibi bir fonksiyon ile tüm değerleri toplayabilir, toplam kayıt sayısını alabilir farklı türde hesaplama fonksiyonları kullanabiliriz.

Örneğin Python’ da: toplam=RDD.reduce(lambda x,y: x+y) Benzer şekilde Java’da: Integer toplam=RDD.reduce(new Function2<Integer,Integer>() { public Integer call(Integer x, Integer y) { return x+y; });

Fold() aksiyonu reduce()’ e benzer biçimde kullanılır. Fakat reduce’ den farklı olarak bir başlangıç “sıfır” değeri ile çalışır. Bu “sıfır” değer toplama için sıfır (0) çarpma için ise bir (1) gibi etkisiz eleman olmalıdır. Bu başlangıç değeri fold()’ un boş liste üzerinde’ de hatasız çalışmasını sağlar. Eğer hiç elemanı olmayan bir RDD üzerinde reduce() çalıştırırsak “null” değerler üzerinde işlemler programın kırılmasına neden olur.

Reduce() ve fold() sonuç olarak RDD ile aynı türde/tipte değer değer geriye döndürür. Bu sum gibi bir operasyonda sorun teşkil etmez ancak bazı durumlarda hesaplanan sonucun farklı türde olması gerekebilir. Örneğin tüm tamsayı değerin ortalamasının alınması operasyonunda olduğu gibi.  Ortalama hesaplamaya çalıştığımız durumlarda, hem toplamı hesaplamalı hem de kaç verimiz (count) olduğunu saklayabilmeliyiz. Bu durumda iki farklı yöntem uygulamak mümkündür. İlk yöntem map() ile her elemanı ayırıp yanına sayı olarak “1” oluşturmak ve (4,1), (8,1), (-3,1) gibi çiftler üretmektir. Ardından reduce() ile bu çiftler üzerinde çalışıp hem count hem de toplamı bulmak mümkündür.

4+8+(-3) = 9 1+1+1=3 9/3= 3 ortalama bulunur.

Diğer yöntem ise aggregate() fonksiyonunu kullanarak kaynak RDD’ den farklı türde hesaplama yapmaktır. Aggregate() aksiyonunda da tıpkı fold’ da olduğu gibi bir başlangıç “sıfır” değeri (0,1) verilir. Fakat bu başlangıç sıfır değeri kaynak RDD (int) ‘ den farklı (double) olabilir. Ayrıca RDD’ deki elemanları birleştirmek için bir fonksiyon adı parametre olarak verilir. Son adımda ise ikinci bir fonksiyon ile farklı düğümlerde yapılan hesaplamaların birleştirilmesi sağlanır.

Sık Kullanılan Aksiyonlar:

RDD={1,2,3,3} .collect() tüm RDD nin ana düğüme getirilmesini sağlar .count() RDD’nin eleman sayısını getirir (4). .countByValue() RDD’ nin her elemanın sayısını ayrı ayrı getirir. { (1,1),(2,1),(3,2)} .take(sayı) örneğin take(2) RDD’ den verilen sayı kadar değer döndürür. take(2)-> {1,2} .takeOrdered(sayı(order) RDD.takeOrdered(2)(order)-> {3,3} .reduce(function) ile tüm elemanlar paralel işlenir RDD.reduce(/x,y)->x+y)=9 RDD.fold((0)(x,y)->x+y)=9 RDD.aggregate(zeroVal, operation, combination)- RDD.aggregate((0,0),(x,y)->(x._1+y,x._2+1),(x,y)->(x._1+y._1, x._2+y._2))

Bir sonraki yazımda kısaca Spark Cache  (Persistance) kavramlarına kısa bir giriş yaptıktan sonra, belki de en önemli konumuz olan KEY/VALUE çiftleri ile çalışmayı anlatmayı planlıyorum.

Spark birleşim, kesişim, fark gibi kümeler üzerinde gerçekleştirilebilecek işlemleri destekler. Tüm küme operasyonlarının aynı veri türü üzerinde gerçekleştirilmesi gereklidir. Aşağıdaki gibi verilen iki RDD için küme işlemlerini örnekler ile açıklayalım:

RDD1={“Merhaba”,”dünyalı”,”biz”,”dostuz”} RDD2={“Merhaba”,”dünyalı”,”marsta”,”hayat”,”varmış”} RDD1.union(RDD2) -> Tüm elemanları birleştirir, tekrar vardır. RDD1.intersection(RDD2)->Kesişim Kümesi -> {“Merhaba”,”dünyalı”} RDD1.subtract(RDD2)->Fark Kümesi ->={”biz”,”dostuz” ,”marsta”,”hayat”,”varmış”}

Herhangi bir RDD’ ye uygulanabilen distinct() operasyonu ise o kümedeki farklı elemanlardan yenibir RDD oluşturur. Örneğin:

RDD={”to”, “be”, “or”, “not”, “to”, “be”} RDD.distinct()->{”to”, “be”, “or”, “not”}

Distinct operasyonunu nerede kullanabilirsiniz?  Aynı filmlerin farklı kullanıcılar tarafından oylandığı bir veri yapısı düşünün. Her filmin bu kümede yalnızca bir kez bulunmasını istediğinizde Distinct kullanabilirsiniz. Ancak Distinct tüm verinin network üzerinden düğümlere dağıtılması ve verinin karıştırılması (suffle) işlemlerini barındırdığı için yavaş çalışan bir operasyondur.

Farklı veri kaynaklarından gelen verileri bir araya toplamak için union kullanılabilir. Benzer şekilde union sonrası Distinct kullanarak veri tekrarını önleyebilirsiniz. Intersection her kümedeki ortak/aynı değeri geri döndürür. Ayrıca bu işlemi tekrar eden verileri yaparak gerçekleştirir. Intersection aynı RDD (kaynak) üzerine uygulanabilir. Böylece sadece tekil elemanlar elde edilir. Benzer şekilde Intersection ortak değerleri bulmak için tüm network hesaplama düğümlerine erişmek zorundadır. Bu nedenle yine yavaş çalışabilir.

Seçilen bazı verileri veri kümesinden çıkarmak için subtract kullanmak avantajlıdır. İki RDD’ nin kombinasyonlarını oluşturmak için ise cartesian kullanılır. Kartezyen dönüşüm olası tüm çiftleri oluşturur. (Dikkatli kullanılmalıdır. Veri setiniz çok büyüyebilir)

RDD1={“Fırat”,”Çağla”} (n=2) RDD2={“Leyla”,”Barış”,”Nusret”} (m=3)

RDD1.cartesian(RDD2) sonrasında nxm=2x3=6 çift oluşacaktır. Bunlar {(Fırat,Leyla),(Fırat,Barış),(Fırat,Nusret),(Çağla,Leyla),(Çağla,Barış),(Çağla,Nusret)}

Tahmin edebileceğiniz üzere büyük boyutlu RDD’ lerden üretilen Kartezyen RDD çok zaman alan bir operasyona dönüşebilir. Sample ve takeSample dönüşümleri bir RDD kümesinden örneklerin alınması için kullanılır. TakeSample veri setinin kaç elemandan oluştuğunu örnekleme için kullanmak zorundadır ve count() fonksiyonun çağırır. Bu nedenle sample ile kıyaslandığında takeSample daha yavaş çalışır.

Zaten şimdiye kadar yazdıklarımızı okuma fırsatı buldu iseniz Map ve Filter’ ın en sık kullanılan Spark fonksiyonlarının başında geldiğini sizler de fark etmişsinizdir. Map ve Filter veri kümesinden bağımsız olarak tüm RDD’ ler üzerinde çalışabilir. Map RDD içerisinde ki her elemanına tek tek uygulanır ve yeni bir RDD üretmek için kullanılır. Map sonucunda oluşan yeni RDD’ deki değerlerin aynı türde olma zorunluluğu yoktur. Örneğin tam sayılardan oluşan bir RDD’ deki her değerin karekökünü alarak küsuratlı (double) türde elemanları içeren yeni bir RDD oluşturmak mümkündür.

Örneğin aşağıda tüm elemanların 2 katını hesaplayan bir Map (Python) kodu verilmiştir.

sayilar=sc.parallelize([1,2,3,4]) ikikati=sayilar.map(lambda sayi:sayi*2).collect() for sayi in ikikati: print “%s “ % (sayi)

Java’ da benzer şekilde :

JavaRDD<Integer> sayilar=sc.parallelize(ArraysçasList(1,2,3,4)); JavaRDD<Integer> ikiKati=sayilar.map(new Function<Integer,Integer>() { Public Integer call(Integer s) { return s*2; } }); System.out.println(StringUtils.join(ikiKati.collect(),”-“));

Bazı durumlarda her bir RDD elemanı için birden fazla çıktı üretmek gerekebilir. Flatmap() bu amaçla kullanılan bir dönüşümdür.

Örneğin: [“Merhaba Spark”,”marsda”,”hayat varmış”] şekliden 3 elemanlı bir String array map ile boşlıukları ayıraç olacak biçimde split edildiğinde, sonuç array yine 3 elemanlıdır. (mapRdd.count() =3)  Herbir eleman : [[‘Merhaba’,’Spark’],[‘marsda’],[‘hayat’,’varmış’]] şeklindedir. mapRDD.take(1) uyguladığımızda [[‘Merhaba’,’Spark’]] çıktısını elde ederiz. Fakat flatmap() kullandıldığında yeni RDD 5 elamanlı olur elemanlar ise: [“Merhaba’,’Spark’,’marsda’,’hayat’,’varmış’] olur flatmapRDD.take(1) çıktısı ise [‘Merhaba’] olur. Flatmap() herbir eleman üzerinde tek tek çalışır fakat bir veya birden fazla elaman üretebilir.

Merhaba Spark-> Merhaba, Spark marsda -> marsda hayat varmış -> hayat, varmış.

Daha önce notlarımızda belirttiğimiz gibi RDD’ ler üzerinde gerçekleştirilen dönüşümler anlık olarak gerçekleştirilmez. Spark motoru bir dönüşüm isteği geldiğini kayıt altına alır. RDD’leri veri saklayan bileşenler gibi düşünmek yerine, RDD’ lerin verinin nasıl işleneceğini belirleyen komut zincirleri olduğunu düşünmek daha doğrudur. Bu verinin hdfs,s3, hive gibi herhangi bir kaynaktan yüklenmesi esnasında da geçerli bir kuraldır. Bir hesaplama isteği (action) olmadan veri gerçekte kaynaktan yüklenmez. Hadoop üzerinde geliştirme yapan programcılar çoğu zaman kullanılacak tüm operasyonları gruplamaya çalışarak MapReduce çevrimlerini azaltmayı hedeflerler. Spark’ da kullanılan Lazy Evaluation (Tembel Değerlendirme :) ) art arda sıralanan dönüşümlerin gruplanarak tek bir operasyon gibi çalıştırılmasını sağlar. Böylece Spark’ da çevrimlerini azaltmak için karmaşık map fonksiyonlarına gerek kalmaz. Spark dönüşüm işlemlerinin (transformations) neredeyse tamamı ilgili veri seti üzerinden yeni sonuçlar üretir. Bu dönüşümlerde yeni veri setini oluşturacak fonksiyonlar kullanılır. Aksiyonların bir bölümünde de benzer fonksiyon yaklaşımı kullanılır. Programlama diline bağlı olarak Spark dönüşümlerinde parametre olarak kullanılacak fonksiyonların yazım şekli farklılık gösterir. Örneğin Python dilinde fonksiyonları Lambda yazımı ile oluşturmak mümkündür.

Örnek: Hatalar=tumVeri.filter(lambda x:”error” in x)

Benzer şekilde ayrı bir fonksiyon oarak

def hataicerensatirlar(satir) return “error” in satir

Kod içerisinde ise

Hatalar=tumVeri.filter(hataicerensatirlar)

Olarak kullanılabilir. Python’ da fonksiyon adı verilerek bir dönüşüm gerçekleştirildiğinde dikkat edilmesi gereken önemli nokta: objeye ait bir metod veya field fonksiyonda parametre olarak kullanılırsa, Spark objenin tamamını hesaplama düğümlerine yolluyor olmasıdır. Bu çok daha fazla verinin hesaplama düğümlerine yollanmasına neden olur. Hatta bu yaklaşım Python’ un bazı sınıfların içerebileceği alt objeleri nasıl işleyeceğini çözememesi nedeni ile kodun fail etmesine neden olabilir. Bu gibi durumlara neden olmamak için objeye ait instance variableların (alanların) yerel değişkenlere atanması ve bu değişkenlerin dönüşüm fonksiyonlarında kullanılması daha doğrudur. Java’ da dönüşümlere parametre olarak verilecek fonksiyonlar spark.aspi.java.function paketindeki fonksiyonları uygulayan (implement) nesneler ile kullanılır. Fonksiyonların geri dönüş tipine göre farklı arabirimler (interface) mevcuttur.

Örneğin: Function<T,R> R call(T) tek bir parametre alarak bir çıktı üretir, genellikle map ve filter gibi dönüşümlerde kullanılır. Function2<T1,T2,R> R call(T1,T2) 2 parametre (girdi) alarak yine tek bir sonuç üretir genelikle aggregate ve fold gibi dönüşümlerde kullanılır. Dikkat edileceği üzere farklı veri tiplerinde çalışılabilmesi için bu sınıflar generics yapısı ile template türünde oluşturulmuştur.

Örneğin: class hataliSatirlar implements Function<String,Boolean>() { public Boolean call (String satir) { return satir.contains(“error”); } } Bu sınıfı kullanmak için ise : RDD<String> hatalar=tumVeri.filter(new hataliSatirlar());

Yukarıda tanımlanan sınıfı filter dönüşümünde anaonim bir sınıf olarak oluşturmak da mümkündür. Java 8 sonrası Lambda yazımı ile çok daha kısa şekilde :

RDD<String> hatalar=tumVeri.filter(satir->satir.contains(“error”));

Java sınıflarına yapıcıları (constructor) kullanarak parametre yollamak da mümkündür.

class hataliSatirlar implements Function<String,Boolean>() { private String param; public hataliSatirlar(String kelime) { this.param=kelime; } public Boolean call (String satir) { return satir.contains(param); } }

Parametrik kullanım için ise:

RDD<String> hatalar=tumVeri.filter(new hataliSatirlar (“error”));

Bir sonraki notumuzda sık kullanılan aksiyon ve dönüşümleri kapsayacak.

Selenium tarihçesi ve bileşenlerinden daha önceki notlarımızda yer almıştı. Henüz okuma fırsatı bulamadıysanız bu adreste.  Bu küçük hatırlatma notunda ise Selenium scriptlerinizi kendi makinanız yerine farklı test sistemlerinde eş zamanlı olarak çalıştırmayı özetleyeceğiz. Test etmeniz gereken çok fazla senaryonuz var ayrıca web uygulamanızı birden fazla browserda test etmek gibi bir ihtiyaç ortaya çıktı ise, test döngülerinizi hızlandırmak ve “biraz” zaman kazanmanız gerekebilir. Size Selenium Standalone Server’ ı GRID konfigürasyonu ile kurmanızı öneriyorum.  Bu notu kayıt altına aldığım esnada stand alone server’ ın son versiyonu olan 2.48.2 ihtiyacınız olan tüm bileşenleri yani hub, webdriver ve eski RC versiyonunu içermekte ve aşağıdaki adresten indirilebilmekteydi.

http://selenium-release.storage.googleapis.com/2.48/selenium-server-standalone-2.48.2.jar

eğer bu link çaçışmıyor ise selenium ana sayfasından stand alone server’ ı indrimeyi deneyebilirsiniz (http://www.seleniumhq.org/download/)

Bileşenleri biraz açıklayalım HUB aslında tüm istekleri alan ve yönlendiren selenium merkezi bileşeni. Test frameworkünü oluşturduğumuz ve yönettiğimize makinenin bu anlamda HUB görevi ile başlaması için:

java -jar selenium-server-standalone-2.48.2.jar -role hub

komutunu yazmamız yeterli. Sisteminizde Java’ nın güncel versiyonu kurulu ise, HUB çalıştığında son iki satırda şu mesajları görmelisiniz:

Nodes should register to http://192.168.1.21:4444/grid/register/ Selenium Grid hub is up and running

Selenium Grid, işleri diğer düğümlere dağıtmak için hazır durumda. HUB varsayılan port olarak 4444’ de çalışmaya başlar. Hub’ ın durumunu görmek için tarayıcı adresinize aşağıdaki linki yazabilirsiniz.

http://localhost:4444/grid/console

Fakat anlayacağınız üzere testleri çalıştıracağınız makineleri/düğümleri (nodes) yukarıda (benim bilgisayarıma ait IP) verilen IP adresi (ya da hostname) ve portu kullanarak HUB’ a tanıtmanız (register) gerekiyor.  Bunun için uzak düğümlerede aynı versiyon standa alone dosyasını kopyalamalı (selenium-server-standalone-2.48.2.jar) ve düğümleri register etmek için aşağıdaki komutu çalıştırmalısınız (IP kısmını değiştirmeyi unutmayın) :

java -jar selenium-server-standalone-2.14.0.jar -role node  -hub http:// 192.168.1.21:4444/grid/register

Bu durumda yukarıdaki komutu çalıştırdığınız makineyi bu defa NODE olarak, yani HUB’ ın emirlerine hazır, selenium scriptlerini çalıştıracak  bir asker olarak ayağa kaldırmış oldunuz. Yine onay mesajı orak HUB üzerinde  bu bilgiyi aşağıdaki biçimde görebilirsiniz.:

INFO - Registered a node http://192.168.1.23:5555

Hatta yine http://localhost:4444/grid/console  yazdığınızda eklediğiniz tüm uzak düğümleri ve yüklü tarayıcı versiyonlarını görmeniz mümkündür.

Harika aslında gereken tüm önemli adımları tamamladınız. Son WebDriver kullanan bir C# projesinde Selenium scriptini aynı makine yerine uzak düğümlere yollamak için yapmanız gereken:

IWebDriver driver = new FirefoxDriver();

Satırınızı :

DesiredCapabilities capa = DesiredCapabilities.Firefox(); IWebDriver driver = new RemoteWebDriver(new Uri("http://localhost:4444/wd/hub"), capa);

İle değiştirmek. Bu durumda istekleri Hub’a yönlendirmiş oldunuz. HUB zaten NODE olarak eklenmiş olan düğümleri biliyor ve istekleri onlara yönlendiriyor olacak. Birden fazla düğüm eklediğinizde DesiredCapabilities ile biraz oynayıp isteği doğru düğümlere yönlendirmek te mümkün. Bunu sizlere bırakıyorum. Unutmadan sorun yaşarsanız firewall ayarlarınızı kontrol etmeyi, ayrıca 4444 ve 5555 portlarının (HUB-NODE varsayılan portları) engellenmediğinden emin olmanızı öneririm.

Sizlerinde görebileceği üzere, bu aralar Parallel and Distributed Computing başlıkları altında sıklıkla kullandığım MongoDB, Hive, NoSQL, Hadoop, Spark gibi araçlara kafayı fazlaca takmış ve özellikle Spark’a daha çok yer ayırmış durumdayım. Söz Selenium yazılarımı artıracağım

RDD’ler üzerinde gerçekleştirilen bir operasyon sonunda yeni bir RDD elde ettiğimiz operasyonlara transfer/dönüşüm (transformation) operasyonları adı verilir. Tamamı için geçerli olmasa da, çoğu dönüşüm RDD üzerindeki her bir veri üzerinde tek tek çalışır. Dönüşümler tembel (lazy) olarak hesaplanır. Yani dönüşüm işlemi bir aksiyon (action) uygulandığında gerçekleştirilir. Bir RDD üzerinde gerçekleştirilen dönüşüm, kaynak RDD ‘ yi değiştirmez. Farklı yeni bir RDD nesnesi oluşturulur.  (Ancak hesaplanmaz!) Kaynak RDD başka operasyonlar için kullanılabilir. Örneğin: inputRDD=sc.textFile(“log.tct”) errorsRDD=inputRDD.filter(lambda x:”error” in x) warningsRDD=inputRDD.filter(lambda x:”error” in x) errAndwarnRDD=errorsRDD.union(warningsRDD)

Yukarıdaki örnekte inputRDD kullanılarak iki farklı RDD oluşturulmuştur. Union ile bu iki farklı RDD birleştirilmiştir. Aslında Union yerine aynı anda he error hem de warningleri filtreleyen tek bir lambda yazmak mümkündür. Fakat Union “lineage” kavramının anlaşılması açısından önemlidir. Spark RDD’ ler arasındaki tüm ilişki ve bağımlılıkları saklar. Bu ilişkileri gösteren graf’a lineage (bağımlılık) graf adı verilir.

Dönüşüm sonrasında oluşan RDD’ler üzerinde çalıştırılabilecek iki önemli operasyon take() ve collect() aksiyonlarıdır. Take ile büyük bir veri setinden oluşan RDD içerisinden küçük/az sayıda örnek almak mümkündür. Python: inputRDD=sc.textFile(“log.tct”) errorsRDD=inputRDD.filter(lambda x:”error” in x) warningsRDD=inputRDD.filter(lambda x:”error” in x) errAndwarnRDD=errorsRDD.union(warningsRDD) print “Filtrelenmiş RDD “ + errAndwarnRDD.count() + “ satirdan oluşmaktadır.” print “Hatalı satırların 10 tanesi:” for satir in errAndwarnRDD.take(10) print satir

Aynı örnek Java ile geliştirildiğinde (önceki yazılarımızdan  RDD nin javada nasıl oluşturulduğuna bakabilirsiniz) System.out.Println(“ Hatali satirlarin 10 tanesi…”); for(String satir: errAndwarnRDD.take(10)) System.out.Println(satir);

RDD’ler oluşturulduktan sonra durumu güncellenemeyen (immutable) dağıtık nesneler koleksiyonudur. Her RDD parçalara ayrılarak farklı hesaplama düğümleri tarafından işlenecek şekilde otomatik olarak dağıtılır. RDD’ler Python, Java, Scala veya kullanıcı tanımlı objeleri içerebilir ve bunlar üzerinde çalışabilir.  RDD’ ler herhangi bir kaynaktan veri yükleyerek veya obje koleksiyonları (liste, set, küme) kullanılarak oluşturulabilir.

Örneğin:

lines=sc.textFile(“C:\Spark\README.md”)

Kullanımı ile text dosyası olunarak lines adlı bir RDD oluşturulur. RDD’ ler oluşturulduktan sonra dönüşümler ve aksiyonlar ile (Transformation, action) işlenirler. Dönüşümler ile kaynak RDD’ den yeni bir RDD oluşturulur. Daha önce kullandığımız filter operasyonu yeni bir RDD üreten bir dönüşümdür:

Sparklines=lines.filter(lambda satir: “Spark” in satir)

 Aksiyonlar ise RDD’ yi kullanarak bir sonuç üretirler. Bu sonuç ana programa geri döndürülebilir veya HDFS gibi bir dış saklama sistemine kayıt edilebilir.

sparklines.first() ise bir aksiyondur. Hesaplamalar dönüşümler ve aksiyonlar için farklı gerçekleşir. RDD oluşturuyor gibi görünsek de, RDD’ nin hesaplanması bir aksiyon çağırıldığında gerçekleştirilir. (Lazy evaluation) Böylece ardı ardına oluşturulan RDD zincirinden sadece gerekli olan, doğru zamanda hesaplandığı için boş yere hafıza kaybı yaşanmaz. Spark core engine tüm dönüşümü bir bütün olarak görür ve sadece son aksiyon için ihtiyaç duyulan hesap gerçekleştirilir.

Örneğin:

lines=sc.textFile(“C:\Spark\README.md”)

sparklines=lines.filter(lambda line: “Spark” in line)

sparklines.first()

kod bloğunda tüm dosya okunmaz. İlk “Spark” bulunana satırda işlem sonlandırılır.  Spark RDD’ leri aksiyonlar kullanıldığında yeniden hesaplar. Eğer RDD’ nin birden fazla aksiyonda kullanılmasını planlıyorsanız

Rdd.persist() veya Rdd.cache() diyerek kalıcı olarak saklanmasını sağlayabilirsiniz

.RDD’ leri hafıza yerine (ram) Diskte saklamak da mümkündür. Aslında Spark’ ın RDD’ leri yeniden hesaplaması “Resilient” kavramı ile ilgilidir. RDD veya herhangi bir düğümdeki bir RDD yi saklayan makine de bir sorun yaşandığında, Spark kullanıcının haberi olmadan eksik bölümü yeniden hesaplar. RDD oluşturmanın diğer bir yöntemi ise spark.paralelize() metodunu kullanmaktır. Bu metoda parametre olarak kendi veri yapılarımızı (array, list vb.) vererek RDD oluşturabiliriz.

sc.paralelize([“To be”,”or”,”not to be”])  Python

sc.paralelize(Arrays.asList(“To be”,”or”,”not to be”))  Java

Bir sonraki yazımızda dönüşüm ve aksiyonları daha detaylı anlatmayı ve RDD bağımlılıklarını saklayan lineage graf’ ından bahsetmeyi son not olarak ekleyelim.

Spark context (sc) objesi hesaplama kümesine (computing cluster) bağlantıyı sağlayan temel objedir. Herhangi bir Shell başlatıldığında sparkcontext otomatik olarak oluşturulur. Spark context RDD’ leri oluşturmak için kullanılır. Örneğin sc.textfile(…) ile dosyadan sc.hadoopFile(…) ile HDFS veya sc.parallelize(…) ile list, array vb. gibi herhangi bir veri setinden RDD oluşturmak mümkündür. RDD oluşturulduktan sonra .count(), .first() gibi operasyonlar ile RDD işlenir.

Eğer cluster üzerinde çalışyor isek .count() gibi bir operasyon ana program (driver code) tarafından hesaplama düğümlerine (worker node, executors) dağıtılır ve her düğüm noktası kendi count işlemini paralel olarak gerçekleştirir. Operasyonlarda fonksiyonları kullanmak da mümkündür. Böylece RDD nin çıktısı tanımlı fonksiyona girdi olarak sunulur ve yeni RDD’ ler üretilebilir. Örneğin bir dosyadaki “Spark” geçen ik satırı bulmak istersek:

lines=sc.textFile(“:\Spark\README.md”)

sparklines=lines.filter(lambda satir:”Spark” in satir)

sparklines.first()

Yukarıda kullanılan lambda yazımı aslında fonksiyon oluşturmanın Python’ daki kısa halidir. Aynı kod fonksiyon olarak yazılırsa:

def  sparkgecenSatirlar(satir) :

           return “Spark” in satir

Lambda yazımı yukarıda tanımlı fonksiyonu kullanacak biçimde düzenlenirse:

Sparklines=lines.filter(sparkgecenSatirlar) elde edilir.

Java 8 versiyonuna kadar lambda yazımı desteklenmemekteydi. Bu nedenle, benzer işlem için Java’ da “Function” arabirimini (interface) uygulayan (implement) aşağıdaki kodu yazmalıyız:

JavaRDD<String> sparklines=lines.filter(

new Function<String,Boolean>() {

           Boolean call(String satir)

           {

                       return satir.contains(“Spark”);

           }

}

);

Java 8 de lambda yazımı ile aynı örnek :

JavaRDD<String> sparklines=lines.filter(satir->satir.contains(“Spark”));

 Filter(…) gibi bir fonksiyon kullanarak çalışan operasyonların tüm hesaplama düğümlerinde paralel çalıştırılması ile işlemlerin hızlı sonuçlandırılması hedeflenir. Filter operasyonuna parametre olarak verilen fonksiyon (satir->satir.contains(“Spark”)) tüm hesaplama düğümlerine yollanır. Böylece tek makinede çalışacak biçimde yazılan program (driver) tüm düğümlerde paralel olarak da işletilebilir.

Spark Context Shell dışında Java, Python, Scala gibi programlama dillerinde geliştirilecek ayrı uygulamalar da da üretilebilir.

Örneğin Python da:

conf=SparkConf().setMaster(“local”).setAppName(“Spark Test Uygulaması”)

sc=SperkContext(Conf=conf)

 Ya da Java’da:

SparkConf conf=new sparkConf().setMaster(“local”).setAppName(“Spark Test Uygulaması”)

JavaSparkContext sc=new JavaSparkContext(conf)

 Yukarıda ki her iki örnek içinde sparkcontext ve conf kütüphanelerinin import edilmesi gereklidir.