Uzak Sunucudaki (Oracle Cloud) Jupyter Notebook Ortamına Lokalden Erişim

Herkese Selam,

Çalıştığımız projelerin büyüklüğü ve uğraştığımız veri setlerinin hacmi arttıkça lokal makinelerimizde çalışmakta zorlanmaya başlamaktayız. Çalıştığımız bilgisayarın donanımsal kısıtları nedeni ile projelerimizi sağlıklı bir şekilde geliştirememekteyiz. Bu gibi durumlarda imdadımıza kendi lokal bilgisayarlarımızdan daha güçlü olan remote sunucular koşmakta. Remote sunucular, çalıştığımız şirket içerisinde bulunan bilgisayarlar olacağı gibi satın aldığımız veya kullanma şansımızın bulunduğu cloud ortamlarda olabilir. Bu yazıda, cloud ortamda bulunan bir makineye Anaconda ve jupyter kurup, lokal makinemden bu ortamda ayağa kaldırmış olduğum jupyter notebook’a erişimi anlatacağım. Umarım farkındalık anlamında faydalı bir yazı olur.

İlk iş olarak kullanacağım uzak sunucuyu OCI (Oracle Cloud Infrustructure) üzerinde kuracağım. Oracle Cloud üzerinde bir demo hesap açıp sizde böyle bir makine edinebilirsiniz (https://cloud.oracle.com/en_US/tryit). Şimdi bu sunucuyu çok hızlı bir şekilde nasıl ayağa kaldıracağımıza bakalım.

İlk olarak bir Compute instance’ı oluşturacağız. Sol üst menüden resimdeki path’i izleyerek instances sayfasına ulaşabiliriz.

Yeni gelen pencerede Create Instance butonuna basıp ayağa kaldıracağımız makinenin spesifikasyonlarını belirleyelim.

Ben bir ubuntu’lu bir makine başlatmayı tercih ediyorum. Burada menünden farklı imajlarda tercih edilebilir.

Aynı sayfa içerisinde makinenin donanımsal özelliklerinide seçebiliyoruz. Ben 2 CPU ve 30 GB’lık bir memory opsiyonu tercih ediyorum. Bu spesifikasyonlar isteğe göre arttırılabilir veya azaltılabilir.

Yine aynı ekran üzerinde bu makineye güvenli bir şekilde erişebilmemiz için bir ssh key’e ihtiyacımız var. Bu key dosyasını oluşturmak için lokal makinemizde aşağıdaki linux komutunu çalıştırıp id_rsa ve id_rsa.pub dosyalarını oluşturabiliriz. Bu dosyalar oluştuktan sonra aşağıdaki ekranda key istenen yere id_rsa.pub dosyasını yükleyerek bu makineye hangi şifre ile bağlanacağımızı söyleyebiliriz.

ssh-keygen -t rsa 

Bu işlemide tamamladıktan sonra sayfa sonundaki Create butonuna basarak çok kısa bir sürede seçtiğimiz spesifikasyonlara uygun bir makinenin ayağa kalktığını ekran üzerinden göreceğiz. Makine ayağa kalktıktan sonra public IP’sini ve bir çok özelliğini yine ekran üzerinden gözlemleyebileceğiz.

Evet makine görüldüğü üzere ayağa kalktı. Burada kullanacağımız en önemli bilgi lokalden ssh yapabilmemiz ve external bağlantılar kurabilmememizi sağlayacak public IP bilgisi. Şimdi bu bilgiyi kullanarak makineye bağlanıp, bu makine içerisine Anaconda ve Jupyter Notebook kurulumu yapalım.

Makineye bağlanırken ürettiğimiz key dosyalarından id_rsa dosyasını kullanacağız. Ssh bağlantısı kurmak için lokal makinemizden aşağıdaki linux komutunu çalıştırıyoruz.

ssh -i id_rsa ubuntu@PUBLIC_IP_ADDRESS

Bağlantıyı tamamlamak için id_rsa dosyasını oluştururken verdiğimiz şifreyi giriyoruz.
Doğrulama işlemi tamamlandıktan sonra Oracle Cloud üzerindeki sunucuya başarılı bir şekilde ulaşıyoruz.

Makineye bağlandığımıza göre şimdi Anaconda ve Jupyter notebook indirip kuralım.

wget https://repo.anaconda.com/archive/Anaconda3-2019.07-Linux-x86_64.sh
sh Anaconda3-2019.07-Linux-x86_64.sh
sudo snap install jupyter

Anaconda ve jupyter kurduktan sonra şimdi makinenin 30303 portundan bir jupyter notebook ayağa kaldıralım.
Ancak jupyter’i başlatmadan önce Cloud’daki makinenin 30303 numaralı portunu Oracle Cloud arayüzünden açmamız gerekiyor. Üst menüden Networking –> Virtual Cloud Network –> Security Lists path’i izlenerek 30303 portu dışardan erişime açılır.

Evet makinenin ilgili portunu erişme açtıktan sonra şimdi Cloud üzerinde başlattığımız makinenin 30303 portundan bir jupyter notebook başlatalım.

jupyter notebook --no-browser --port 30303

Notebook başarılı cloud’daki makinede bir şekilde başladı. Şimdi lokal makinemizden, cloud’daki makinenin 30303 portundan servis edilen jupyter notebook’a bağlanabilmek için lokal makinemizden ilgili yönlendirme işlemini yapalım. Bu komut ile uzak sunucudaki uygulamayı lokalimizdeki bir porta yönlendirerek, lokalimizdeki port üzerinden bu uygulamaya erişmiş olacağız.

Şimdi lokal makinemizden ilgili yönlendirmeyi yapıp, lokal makinemizdeki bir browser üzerinden jupyter notebook’a bağalanalım.

ssh -N -f -L localhost:10001:localhost:30303 ubuntu@PUBLIC_IP_ADDRESS

Lokal makinemde uygulamayı 10001 portuna yönlendirdim. Şimdi browser’a http://localhost:10001 yazıp çalıştıralım.

Evet giriş sayfasına başarılı bir şekilde ulaştık. Bize uygulamanın içine girebilmek için bir token sormakta. Bu tokenı sistem bize jupyter notebook’u başlattığımızda dönmekte.

jupyter notebook --no-browser --port 30303

Yukarıdaki komutu çalıştırdığımız ekran görüntüsünde bu token’ı görebilirsiniz. Bu tokenı girdikten sonra artık istediğimiz şekilde notebook başlatıp ana makinemiz üzerinden uzaktaki sunucuya bağlanıp kod geliştirebiliriz.

Kaynaklar:

https://jupyter.readthedocs.io/en/latest/

https://docs.anaconda.com/

https://docs.oracle.com/en/cloud/

Advertisements

Spark ve Hive ile Basit bir ETL Uygulaması

Herkese Selam,

Bu yazıda HDFS (Hadoop File System) üzerinde Spark ile örnek bir veri setini okuyup  basit bir analitik operasyon yapıp daha sonra Hive’da yaratacağım bir tabloya yazacağım.

Günümüzde üretilen veri miktarının çok yüksek ölçekte olması ve toplanan verinin analiz edilme ihtiyacının her geçen gün artması sonucunda büyük veri teknolojileri hayatımıza hızla girdi. Artık neredeyse bir çok firmanın büyük veri ortamları bulunmakta ve bu ortamlarda büyük veri teknolojilerini kullanarak analizler yapmaktalar.

Endüstride çok bilinen ve kullanımı en yaygın olaran büyük veri ortamları Hadoop’tur. Hadoop ekosisteminde, verinin saklanmasından verinin işlenmesine, verinin  görselleştirilmesinden sorgulanmasına kadar bir çok çözüm bulunmaktadır.  Hadoop ekosisteminde bulunan her bir çözüm ve kütüphane ayrı uzmanlıklar gerektiren derinlikte teknolojilerdir. Bu ekosistemin barındırdığı farklı teknolojileri ve kullanım alanlarını aşağıdaki resimden inceleyebilirsiniz.

Şimdi örnek uygulamanın yapımına geçelim. Yapacağımız örnekte veriyi önce linux file sistemimizden, Hadoop ekosisteminin veri saklama ünitesi olan HDFS’e göndereceğiz (Bu adıma Extraction diyebiliriz.). Daha sonra buraya yazdığımız veriyi Spark ile önce okuyacağız ve daha sonra basit bir transformasyon (Transformation) uygulayıp Hive’a yazacağız (Load). Hive, hadoop ekosisteminde var olan ve bu ortamda saklanan verileri sorgulayabilmemizi sağlayan bir alt yapı. Biz bu alt yapı ile büyük veri ortamımızdaki verileri SQL dili kullanarak kolayca sorgulayabiliyoruz.

Öncelikle örnek çalışacağımız veri setini tanıyalım. Çalışacağımız veri seti Kaggle üzerindenki “Sample Sales Data” veri seti. Aşağıdaki tabloda kolon adlarını ve tiplerini görebilirsiniz.

Column Name	Data Type
ORDERNUMBER	Number
QUANTITYORDERED	Number
PRICEEACH	Number
ORDERLINENUMBER	Number
SALES	Number
ORDERDATE	Date
STATUS	String
QTR_ID	Number
MONTH_ID	Number
YEAR_ID	Number
PRODUCTLINE	String
MSRP	Number
PRODUCTCODE	String
CUSTOMERNAME	String
PHONE	Number
ADDRESSLINE1	String
ADDRESSLINE2	String
CITY	String
STATE	String
POSTALCODE	Number
COUNTRY	String
TERRITORY	String
CONTACTLASTNAME	String
CONTACTFIRSTNAME	String
DEALSIZE	String

Bu veriseti, örnek satış işlemleri transactionlarını içermektedir. Biz bu örnek satış datasından bazı transformationlar yapıp örnek bir rapor datası oluşturup bunu bir Hive tablosuna insert edeceğiz.

İlk olarak Kaggle’dan download ettiğimiz örnek veriyi HDFS’e kopyalayalım. Bunun için hdfs’te öncelikle bir dizin oluşturalım.

hadoop fs -mkdir samplesales

Evet, hdfs’te dizini yarattık şimdi ise lokal dizinimiz de duran örnek veriyi hdfs’e kopyalayalım.

hadoop fs -copyFromLocal sales_sample_data.csv samplesales

hadoop fs -ls samplesales/

Veriyi hdfs’e yazdık şimdi ise Pyspark arayüzümüzü başlatıp veriyi spark ile işlemeye başlayalım.

/user/spark-2.1.0-bin-hadoop2.7/bin/pyspark --master yarn-client --num-executors 4 --driver-memory 2g --executor-memory 4g

Spark başarılı bir şekilde başladı. Bu aşamada kullanacağımız kütüphaneleri import edelim.

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
from pyspark.sql import Row
from pyspark.sql import HiveContext
from pyspark.sql.functions import *

hive_context = HiveContext(sc)
sqlContext = SQLContext(sc)

Şimdi ise örnek versetimizi tutacağımız satır formatını oluşturalım.

RecSales = RecSales = Row('ordernumber','quantityordered','priceeach','orderlinenumber','sales','orderdate','status','qtr_id','month_id','year_id','productline','msrp','productcode','customername','phone','addressline1','addressline2','city','state','postalcode','country','territory','contactlastname','contactfirstname','dealsize')

Sıra hdfs’e aldığımız verimizi okumaya ve bir dataframe içerisine yazmaya geldi. Veri içerisinde kolonlar “,” ayracı ile ayrildığı için bunu ayracı veriyi yüklerken belirtiyoruz. Veriyi data frame içine aldıktan sonra bir isim verip temp table olarak işaretliyoruz. Bu isimi daha sonra hive_context veya sqlContext içerisinde SQL yazarken kullanabileceğiz.

dataSales = sc.textFile("/user/samplesales/")
header = dataSales.first()
dataSales= dataSales.filter(lambda line: line != header)
recSales = dataSales.map(lambda l: l.split(","))
dataSales = recSales.map(lambda l: RecSales(l[0],l[1],l[2],l[3],l[4],l[5],l[6],l[7],l[8],l[9],l[10],l[11],l[12],l[13],l[14],l[15],l[16],l[17],l[18],l[19],l[20],l[21],l[22],l[23],l[24]))
dfRecSales = hive_context.createDataFrame(dataSales)
dfRecSales.registerTempTable("sales")

Veriyi başarılı bir şekilde hdfs’ten okuduk ve bir data frame objesi içerisine yazdık. Şimdi Spark SQL ile yüklediğimiz veriye basit bir kaç sorgu atalım.

hive_context.sql("select count(*) from sales").collect()

hive_context.sql("select * from sales").head()

hive_context.sql("select ordernumber,priceeach  from sales").head(2)

Evet şimdi satışları territory (bölge) alanına göre gruplayalım ve sonuçları bir Hive tablosuna yazalım.

dfterriroty = hive_context.sql("select territory,sum(priceeach) total from sales group by territory")
dfterriroty.registerTempTable("sumterr")
dfterriroty.show()

Elde ettiğimiz sonucu bir hive tablosu yaratıp içerisine yazalım.

hive_context.sql("CREATE TABLE IF NOT EXISTS territory_sum (territory String, total INT)")

hive_context.sql("insert into territory_sum select * from sumterr")

Son olarak Hive’a yazılan datayı kontrol edelim.

hive_context.sql("select * from territory_sum").show()

hive_context.sql("select count(*) from territory_sum").show()

hive_context.sql("select count(*) from sumterr").show()

Oracle Cloud Based Machine Learning Alt Yapısı

Herkese Selam,

Oracle Cloud üzerinde analitik çalışmalar yapabilmemiz adına bazı güzel özellikler geliştirmiş durumda. Cloud üzerinde Oracle Advanced Analytics package’i ile gelen bir çok özelliği cloud üzerinde kullanabilmemiz sağlanmış durumda.  Oracle Cloud üzerinden ulaşılabilen bu alt yapıda Cloud veritabanımız üzerinde sorgular çalıştırabilmemiz yanında ML notebook’lar yaratıp analitik çalışmalarımızı gerçekleştirebiliyoruz. Kullanıcıların adaptasyonunu hızlandırmak için ise bir çok örnek çalışma içeriye koyulmuş durumda.  

Bu alt yapının sunmuş olduğu en güzel özellik şüphesiz ki zepplin based bir ML notebook alt yapısının bizlere sunulmuş olması. Bu notebook mekanizması ile web based bir şekilde analizlerimizi yapmamız mümkün. Veriyi görselleştirmede ve analiz etmede oldukça güçlü bir alt yapı sunulduğunu söylemek mümkün.

Oracle Advanced Analytics’in bize sunmuş olduğu onlarca algoritma mevcut. Veritabanızmıda bulunan veriyi veritabanı dışarısına çıkartmadan bize sunulan notebook alt yapısı ile istediğimiz ML modelini datamız üzerinde kullanıp makine öğrenmesi uygulamaları geliştirebilmekteyiz.

Regression, Classification, Anomaly Detection, Clustering, Market Basket Analysis ve Attribute Importance gibi bir çok ML modelini bu alt yapı ile kullanabilmekteyiz.

Daha detaylı bilgi sahibi olabilmek için trial bir Oracle Cloud hesabı açıp bu özellikleri deneyebiliyoruz. Oracle 1 aylık 250 Euro krediye sahip bir hesabı kullanımımıza açıyor.

Ücretisiz bir Oracle Cloud hesabı açmak için linki takip edebilirsiniz. [Oracle Cloud]

Ayrıca Oracle Cloud ML servisini kullanabilmek için güzel bir dokumantasyon hazırlanmış durumda. Bu dokumantasyonu kullanarak Oracle Cloud ML servisini ayağı kaldırıp kullanmaya başlayabiliriz.

https://oracle.github.io/learning-library/workshops/adwc4dev/?version=Self-Guided&page=L300.md

Ayrıca aşağıdaki video izlenerek alt yapının nasıl kullanılacağı ile ilgili fikir edinilebilir. Ek kaynak olarak şu sunuma da bakılabilir.

Hibrit Partitioned Tablolar

Herkese Selam,

Partitioning Oracle 8 sürümünden bu yana hayatımızda olan ve benim kişisel olarak Oracle veritabanının en beğendiğim özellikleri arasında. Oracle’da partition alt yapısı büyük hacimli veriler tuttuğumuz tablolarda veriye erişim konusunda sorgu performanslarımızı arttıran, özellikle veri ambarı ortamlarında vazgeçemediğimiz özelliklerin en başında geliyor. Tablolarımızı doğru fiziksel partition tasarımları yaparak büyük hacimli veriyi rahatlıkla sorgulayabilmekteyiz.

Oracle’ın her major veya minor sürüm güncellemelerinde partitioning alt yapısı ile ilgili bir çok yenilik ve güncelleme oldu.  Bu güncellemeler 19c sürümü ile de devam etti. Oracle 19c ile beraber partitioning tarafında yeni bir özellik daha geldi. Bu özelliğin adı hibrit partitioned tablolar.

Partitioning bilindiği üzere büyük tabloları, küçük parçalara bölerek daha kolay ulaşılabilir ve yönetilebilir hale getiriyordu. Bunun yanı sıra bir çok farklı tipte (List-Hash-Range) partitionlar oluşturmamıza sistem olanak sağlıyordu. Ancak kurumsal dünyadaki verinin büyüklüğündeki devasa büyümeler, yasal regülasyonlar ve karmaşıklığın artması bu tarafta yapılacak geliştirmelerin zorunluluğunu arttırdı. Özellikle verinin sürekli online olma ihtiyacı önemli zorluklardan bir tanesi olarak karşımıza çıkmakta.

Yeni gelen hibrit partitioned tablolar ile artık DBA’ler büyük hacimli tablolarda eskisi gibi partitionlama işlemlerini yapabilecekleri gibi veritabanı içerisinde hangi partitionların hızlı bir şekilde sorgulamaya ve güncellemeye açık olacağını seçebilmekteler. Aynı zamanda partitionlara read only erişim vermek gibi bir opsiyonda yine yapılabilmekteler. Bunun yanı sıra yeni gelen bu özellik ile beraber cold ve eski verileri tutan partitionlar farklı external storage’larda saklanabiliyor. Örneğin iki yıldan daha eski datayı tutan partitionları standart file sistem de dosya şeklinde saklayabileceğim gibi hdfs gibi storage cost’u düşük ortamlarda da tutabiliyorum.

Oracle 19c Kullanmaya Başlayalım

Herkese Selam,

Oracle kısa bir süre önce yeni DB sürümü olan 19c’yi duyurdu. Oracle 19c yine Cloud base çalışan ve otonom olduğu iddiası ile ilgileri üzerine çeken ve bunların yanında yeni bir çok özelliğe sahip olan bir veritabanı yazılımı olmuş gibi görünüyor.

Oracle 19c ile ilgili bilgi sahibi olmak için linki takip edebilirsiniz.

Oracle, 19c sürümünü hızlıca deneyebilmek adına ise çok güzel bir iş daha yapmış ve LiveSQL platformuna 18c’yi hemen entegre etmiş.  LiveSQL  platformu üzerinden 18c veritabanını hızlıca deneyimleyebilmek mümkün.

Spark ile Oracle ve Cassandra Üzerinde Veritabanı İşlemleri

Herkese Selam,

Bu yazıda Spark ile Cassandra veritabanına veri yazma ve veri okuma işlemi gerçekleştireceğim. Umarım farkındalık anlamında faydalı bir yazı olur.

 

Günümüzde veri kaynaklarının ve veri miktarlarının hızla artması, geleneksel yöntemler ile toplanan verinin işlenmesini zorlaştırmıştır. Toplanan veri miktarının artması aynı zamanda verinin işlenme ihtiyacını her geçen gün arttırmıştır. Bu ihtiyaç ve zorlukları takiben büyük verinin hızlı analiz edilmesi ve depolanması ile ilgili çeşitli çözümler üretilmiştir. Spark verinin işlenmesi tarafında günümüzde kullanılan en yaygın ve performanslı bir çözüm iken Cassandra‘da büyük verinin depolanması ve sorgulanması konusunda endüstride yaygın kullanılan veritabanlarından biridir. Bu makalede bu iki güncel teknolojiyi bir arada kullanmayı deneyeceğiz.

Öncelikle bu örneği yaparken kullandığım yazılımlar hakkında bilgi vermek istiyorum.

İşletim Sistemi: SUSE

Spark: Spark 2.1.0

Oracle Database : Oracle 11g R2

Cassandra: Cassandra 3.4

Hadoop: Hadoop 2.7

İlk olarak Spark ile Oracle veritabanına bağlanıp örnek bir veri çekelim ve bunu hdfs’e yazalım.  Bu işlemi yapabilmek için sistemimizde ojdbc6.jar dosyasının olması gerekmekte. Bunu indirmek için aşağıdaki linki kullanabilirsiniz.

http://www.oracle.com/technetwork/apps-tech/jdbc-112010-090769.html

Şimdi Oracle’dan okuyacağımız tabloyu veritabanında yaratalım ve içerisine örnek data koyalım.

CREATE TABLE EMP
(
   EMPNO      NUMBER,
   ENAME      VARCHAR (10),
   JOB        VARCHAR (9),
   MGR        NUMBER,
   SAL        NUMBER,
   COMM       NUMBER,
   DEPTNO     NUMBER
);

INSERT INTO EMP VALUES
(7369, 'SMITH', 'CLERK', 7902, 800, 50, 20);

INSERT INTO EMP VALUES
(7499, 'ALLEN', 'SALESMAN', 7698, 1600, 300, 30);

INSERT INTO EMP VALUES
(7521, 'WARD', 'SALESMAN', 7698, 1250, 500, 30);

INSERT INTO EMP VALUES
(7566, 'JONES', 'MANAGER', 7839, 2975, NULL, 20);

INSERT INTO EMP VALUES
(7654, 'MARTIN', 'SALESMAN', 7698, 1250, 1400, 30);

INSERT INTO EMP VALUES
(7698, 'BLAKE', 'MANAGER', 7839, 2850, NULL, 30);

INSERT INTO EMP VALUES
(7782, 'CLARK', 'MANAGER', 7839, 2450, NULL, 10);

INSERT INTO EMP VALUES
(7788, 'SCOTT', 'ANALYST', 7566, 3000, NULL, 20);

INSERT INTO EMP VALUES
(7839, 'KING', 'PRESIDENT', NULL, 5000, NULL, 10);

INSERT INTO EMP VALUES
(7844, 'TURNER', 'SALESMAN', 7698, 1500, 0, 30);

INSERT INTO EMP VALUES
(7876, 'ADAMS', 'CLERK', 7788, 1100, NULL, 20);

INSERT INTO EMP VALUES
(7900, 'JAMES', 'CLERK', 7698, 950, NULL, 30);

INSERT INTO EMP VALUES
(7902, 'FORD', 'ANALYST', 7566, 3000, NULL, 20);

INSERT INTO EMP VALUES
(7934, 'MILLER', 'CLERK', 7782, 1300, NULL, 10);

COMMIT;

Şimdi Apache Spark’ı Pyspark arayüzü ile (Python Interface) linux terminalinden başlatıyoruz.

/spark-2.1.0-bin-hadoop2.7/bin/pyspark 
--jars "/home/jars/ojdbc6.jar" 
--master yarn-client 
--num-executors 10 
--driver-memory 16g 
--executor-memory 8g

Evet Apache Spark’ı başlattık. Şimdi veritabanından okuma yapacağımız Python kodunu yazalım ve çalıştıralım.

empDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:oracle:thin:username/password@//hostname:portnumber/SID") \
    .option("dbtable", "hr.emp") \
    .option("user", "db_user_name") \
    .option("password", "password") \
    .option("driver", "oracle.jdbc.driver.OracleDriver") \
    .load()

empDF.show()

empDF.select('EMPNO','ENAME','JOB','MGR','SAL','COMM','DEPTNO').write.format('com.databricks.spark.csv').save('/employees/')

Evet veriyi Oracle’dan Spark ile okuyup hdfs’e yazdık. Şimdi bu veriyi Cassandra veritabanında saklayacağımız tabloyu cassandra’da yaratalım.

CREATE TABLE emp(
empno int,
ename text,	
job text,
mgr text,
sal text,
comm text,
deptno text,
primary key(empno)	
);

Şimdi yeni bir pyspark başlatalım ve hdfs’e yazdığımız veriyi okuyup cassandra veritabanına yazalım. Ancak bunu yapmadan önce cassandra veritabanı ile bağlantı kurabilmek için bazı .jar dosyalarına ihtiyacımız bulunuyor. Bu .jar dosyalarını indirdikten sonra pyspark’ı bu jar dosyalarını kullanarak başlatıyoruz.

* spark-cassandra-connector-2.4.0-s_2.11.jar
* jsr166e-1.1.0.jar
* pyspark-cassandra-0.8.0.jar

/spark-2.1.0-bin-hadoop2.7/bin/pyspark \
  --jars /jar_s/spark-cassandra-connector-2.4.0-s_2.11.jar,/jar_s/jsr166e-1.1.0.jar\
  --py-files /jar_s/pyspark-cassandra-0.8.0.jar \
  --conf spark.cassandra.connection.host= CASSANDRA_IP_ADDRESS

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
from pyspark.sql import Row
from pyspark.sql import HiveContext
from pyspark.sql.functions import *


hive_context = HiveContext(sc)
sqlContext = SQLContext(sc)

RecEmp = Row('empno','ename','job','mgr','sal','comm','deptno')


dataEmp = sc.textFile("/employees/")
recEmp = dataEmp.map(lambda l: l.split(","))
dataEmp = recEmp.map(lambda l: RecEmp(float(l[0]),l[1],l[2],(l[3]),(l[4]),(l[5]),(l[6])))
dfEmp = hive_context.createDataFrame(dataEmp)
dfEmp.registerTempTable("emp")


spark = SparkSession.builder \
  .appName('SparkCassandraApp') \
  .config('spark.cassandra.connection.host', 'CASSANDRA_IP_ADDRESS') \
  .config('spark.cassandra.connection.port', 'CASSANDRA_PORT') \
  .config('spark.cassandra.auth.username','CASSANDRA_USER') \
  .config('spark.cassandra.auth.password','CASSANDRA_PASS') \
  .master('local[2]') \
  .getOrCreate()

Hdfs’e daha önce yazdığımız veriyi okudum ve cassandra bağlantılarını kurdum. Şimdi hdfs’den okuduğumuz veriyi cassandra veri tabanına yazalım.

dfEmp.write\
    .format("org.apache.spark.sql.cassandra")\
    .mode('append')\
    .options(table="emp", keyspace="test")\
    .save()

Evet data cassandra’ya başarılı bir şekilde yazıldı. Şimdi cassandra üzerinden veriyi kontrol edelim.

select * from emp;

Verinin başarılı bir şekilde yazıldığını gördük. Şimdi cassandraya yazdığımız veriyi Spark ile okuyup. Pyspark ile console üzerinde gösterelim.

ds = sqlContext \
  .read \
  .format('org.apache.spark.sql.cassandra') \
  .options(table='emp', keyspace='test') \
  .load()

ds.show()

Sklearn ile Machine Learning Modellerinin Saklanması, Yüklenmesi ve Kullanılması

Herkese Selam,

Oluşturduğumuz makine öğrenmesi tabanlı modellerin saklanması ve daha sonra yüklenip kullanılma ihtiyacı sıkça karşılaştığımız durumlardan biri. Bu yazıda sklearn ile oluşturduğumuz modelleri nasıl saklayacağımızı ve daha sonra nasıl yükleyip kullanacağımızdan bahsedeceğim. Umarım farkındalık anlamında faydalı bir yazı olur.

Öncelikle örnek bir model oluşturalım. Bu örnekte oluşturacağımız model basit bir text classification yapan bir model olacak. Bu modelli oluştururken kullanacağım veri seti ile ilgili bilgiler aşağıdaki gibidir.

Veri Seti : Sentiment Labelled Sentences Data Set

Kaynak : UCI Machine Learning Libarary

İçerik : Bu veri seti 3 farklı internet sitesi (Amazon, Yelp, Imdb) üzerinden toplanmış kullanıcı yorumları ile oluşturulmuştur. Bu yorumlar restoran, film ve ürün değerlendirmelerinden oluşmaktadır. Veri setindeki her bir kayıt iki farklı duygu ile etiketlenmiştir. Bunlar 1: Olumlu/Pozitif 2:Olumsuz/Negatif

Şimdi modeli oluşturma adımına geçelim.
(Bu modelin detaylı anlatımı için linki takip edebilirsiniz. https://emrahmete.wordpress.com/2018/11/25/dogal-dil-isleme-nlp-ile-sentiment-duygu-analizi-tespiti/
)

import pandas as pd
import numpy as np
import pickle
import sys
import os
import io
import re
from sys import path
import numpy as np
import pickle
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelBinarizer
import matplotlib.pyplot as plt
from string import punctuation, digits
from IPython.core.display import display, HTML
from nltk.corpus import stopwords
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
from nltk.tokenize import RegexpTokenizer

#Amazon Data
input_file = "../data/amazon_cells_labelled.txt"
amazon = pd.read_csv(input_file,delimiter='\t',header=None)
amazon.columns = ['Sentence','Class']

#Yelp Data
input_file = "../data/yelp_labelled.txt"
yelp = pd.read_csv(input_file,delimiter='\t',header=None)
yelp.columns = ['Sentence','Class']

#Imdb Data
input_file = "../data/imdb_labelled.txt"
imdb = pd.read_csv(input_file,delimiter='\t',header=None)
imdb.columns = ['Sentence','Class']


#combine all data sets
data = pd.DataFrame()
data = pd.concat([amazon, yelp, imdb])
data['index'] = data.index

#Text Preprocessing
columns = ['index','Class', 'Sentence']
df_ = pd.DataFrame(columns=columns)

#lower string
data['Sentence'] = data['Sentence'].str.lower()

#remove email adress
data['Sentence'] = data['Sentence'].replace('[a-zA-Z0-9-_.]+@[a-zA-Z0-9-_.]+', '', regex=True)

#remove IP address
data['Sentence'] = data['Sentence'].replace('((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\.|$)){4}', '', regex=True)

#remove punctaitions and special chracters
data['Sentence'] = data['Sentence'].str.replace('[^\w\s]','')

#remove numbers
data['Sentence'] = data['Sentence'].replace('\d', '', regex=True)

#remove stop words
for index, row in data.iterrows():
    word_tokens = word_tokenize(row['Sentence'])
    filtered_sentence = [w for w in word_tokens if not w in stopwords.words('english')]
    df_ = df_.append({"index": row['index'], "Class":  row['Class'],"Sentence": " ".join(filtered_sentence[0:])}, ignore_index=True)

data = df_

#Split test and training data set
X_train, X_test, y_train, y_test = train_test_split(data['Sentence'].values.astype('U'),data['Class'].values.astype('int32'), test_size=0.10, random_state=0)
classes  = data['Class'].unique()

#Creating Model
from sklearn.metrics import confusion_matrix
from sklearn.model_selection import cross_val_score
from sklearn.metrics import accuracy_score
from sklearn.neural_network import MLPClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import SGDClassifier


#grid search result
vectorizer = TfidfVectorizer(analyzer='word',ngram_range=(1,2), max_features=50000,max_df=0.5,use_idf=True, norm='l2') 
counts = vectorizer.fit_transform(X_train)
vocab = vectorizer.vocabulary_
classifier = SGDClassifier(alpha=1e-05,max_iter=50,penalty='elasticnet')
targets = y_train
classifier = classifier.fit(counts, targets)
example_counts = vectorizer.transform(X_test)
predictions = classifier.predict(example_counts)

Evet modelimizi oluşturduk. Şimdi ise yazımıza konu olan bu modeli saklamak/kaydetmek kısmına geldik. Modelin saklanma işlemini pickle kütüphanesi ile yapacağız.

# Model and Vocabulary Save
pickle.dump(classifier,open("sentiment_classifier.pkl","wb"))
pickle.dump(vocab,open("vocab_sentiment_classifier.pkl","wb"))

Evet modelimizi kayıt ettik. Şimdi kayıt ettiğimiz modeli tekrar geri yükleyip örnek bir metnin sentimentini bulalım.

# Reload Model and Vocabulary
vec = open("sentiment_classifier.pkl", 'rb')
loaded_model = pickle.load(vec)

vcb = open("vocab_sentiment_classifier.pkl", 'rb')
loaded_vocab = pickle.load(vcb)

Load işlemi başarı ile tamamlandı. Şimdi load ettiğimiz modeli kullanarak bir örnek yapalım.

# Single Value Prediction
examples = 'this is the greatest film that I have ever seen'

#lower string
examples = examples.lower()

#remove email adress
examples = re.sub(r'[a-zA-Z0-9-_.]+@[a-zA-Z0-9-_.]+', '', examples)

#remove IP address
examples = re.sub(r'((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\.|$)){4}', '', examples)

#remove punctaitions and special chracters
examples = re.sub(r'[^\w\s]', '', examples)

#remove numbers
examples = re.sub(r'\d', '', examples)

#remove stop words
examples = ttp.clean_turkish_stop_words(examples)

examples = [examples]

from sklearn.feature_extraction.text import TfidfTransformer
count_vect = TfidfVectorizer(analyzer='word',ngram_range=(1,2), max_features=50000,max_df=0.5,use_idf=True, norm='l2',vocabulary=loaded_vocab)
tfidf_transformer = TfidfTransformer()
x_count = count_vect.fit_transform(examples)
predicted = loaded_model.predict(x_count)

result_category = predicted[0]

if result_category == 1:
    print('Positive')
else:
    print('Negative')

Input : ‘this is the greatest film that I have ever seen’
Result : Positive

Evet test ettiğimiz gibi pozitif bir duygu ile yazdığımız metini sistem’de positive olarak buldu.
Böylelikle daha önceden kayıt ettiğimiz bir modeli, yükleyip başarılı bir şekilde kullanmış olduk.