{"id":1203,"date":"2019-11-02T15:17:51","date_gmt":"2019-11-02T14:17:51","guid":{"rendered":"http:\/\/wchmurze.cloud\/?p=1203"},"modified":"2019-11-13T21:08:17","modified_gmt":"2019-11-13T20:08:17","slug":"colab-zaiskrzyl-podczas-instalacji-apache-sparka","status":"publish","type":"post","link":"https:\/\/wchmurze.cloud\/index.php\/2019\/11\/02\/colab-zaiskrzyl-podczas-instalacji-apache-sparka\/","title":{"rendered":"Colab zaiskrzy\u0142 podczas instalacji Apache Sparka"},"content":{"rendered":"

Jednym z podstawowych problem\u00f3w m\u0142odego adepta sztuki torturowania danych i\u00a0 analizy du\u017cych zbior\u00f3w jest spory pr\u00f3g wej\u015bcia.\u00a0 Gdzie si\u0119 tego mo\u017cna nauczy\u0107 ?\u00a0 Jednym z bardziej znanych narz\u0119dzi jest Apache Spark. Okazuje si\u0119, \u017ce\u00a0 uruchomienie tego oprogramowania\u00a0 w us\u0142udze Google Colab jest banalne. Systemem operacyjnych z jakiego korzystamy jest Ubuntu 18.04. Oczywi\u015bcie taka konfiguracja nie umo\u017cliwi nam przetworzenia ogromnych ilo\u015bci danych, ale pozwoli na rozpoznanie narz\u0119dzi, podstawowych\u00a0 poj\u0119\u0107 i zbudowania nowych kompetencji.<\/p>\n

 <\/p>\n

Po uruchomieniu notatnika\u00a0 w j\u0119zyku Python 3.<\/p>\n

W pierwszej kom\u00f3rce wklejamy<\/p>\n

!wget https:\/\/raw.githubusercontent.com\/djkormo\/colab-examples\/master\/spark\/install.bash -O install.bash\r\n<\/pre>\n
--2019-10-31 22:32:44--  https:\/\/raw.githubusercontent.com\/djkormo\/colab-examples\/master\/spark\/install.bash<\/a>\r\nResolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.0.133, 151.101.64.133, 151.101.128.133, ...\r\nConnecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.0.133|:443... connected.\r\nHTTP request sent, awaiting response... 200 OK\r\nLength: 382 [text\/plain]\r\nSaving to: \u2018install.bash\u2019\r\n\r\ninstall.bash        100%[===================>]     382  --.-KB\/s    in 0s      \r\n\r\n2019-10-31 22:32:44 (59.8 MB\/s) - \u2018install.bash\u2019 saved [382\/382]<\/pre>\n
\u015aci\u0105gni\u0119ty w ten spos\u00f3b plik instalacyjny zawiera zaledwie kilka linii kodu<\/p>\n
!cat install.bash\r\n<\/pre>\n
#!\/bin\/bash\r\n\r\ncd \/\r\n\r\nmkdir -p content\r\n\r\ncd content\/ \r\n\r\napt-get install openjdk-8-jdk-headless\r\n\r\nwget http:\/\/apache.crihan.fr\/dist\/spark\/spark-2.4.4\/spark-2.4.4-bin-hadoop2.7.tgz -O spark-2.4.4-bin-hadoop2.7.tgz\r\n\r\ntar xf spark-2.4.4-bin-hadoop2.7.tgz\r\n\r\npip install -q findspark\r\npip install spark-nlp\r\n\r\nexport JAVA_HOME=\/usr\/lib\/jvm\/java-11-openjdk-amd64\r\nSPARK_HOME=\/content\/spark-2.4.4-bin-hadoop2.7\r\n\r\n<\/pre>\n
 <\/p>\n
Instalujemy JDK dla Javy 1.8, nast\u0119pnie \u015bci\u0105gamy z repozytorium Apache Sparka w postaci pliku .tgz, rozpakowujemy zawarto\u015b\u0107 i ustawiamy dwie zmienne \u015brodowiskowe.<\/p>\n
Uruchamiany skrypt instalacyjny<\/p>\n
!bash install.bash<\/pre>\n
Nadesz\u0142a pora na uruchomienie przyk\u0142adowego kawa\u0142ka kodu<\/p>\n
Najpierw uruchamiamy sesj\u0119 Sparka<\/p>\n
import os\r\nos.environ['JAVA_HOME'] = '\/usr\/lib\/jvm\/java-8-openjdk-amd64'\r\nos.environ['SPARK_ME'] = '\/content\/spark-2.4.4-bin-hadoop2.7'\r\nimport findspark\r\nfindspark.init(\"spark-2.4.4-bin-hadoop2.7\")\r\n\r\nfrom pyspark.sql import SparkSession\r\n\r\nspark = SparkSession \\\r\n    .builder \\\r\n    .appName(\"First App\") \\\r\n    .master(\"local[*]\")\\\r\n    .getOrCreate()\r\n\r\nprint(spark)\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
pyspark.sql.session.SparkSession object at 0x7fe1c00a5128<\/pre>\n
Potem uruchamiamy kontekst Sparka.<\/p>\n
from pyspark import SparkContext\r\nsc = SparkContext.getOrCreate()\r\nprint(sc)\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
SparkContext master=local[*] appName=First App<\/pre>\n
Musimy mie\u0107 \u015bwiadomo\u015b\u0107, \u017ce ka\u017cda zmiana kodu, instalacja danych, zastosowane algorytmy s\u0105 uruchamiane na klastrze obliczeniowym. Dost\u0119p do infrastruktury jest za pomoc\u0105 kontekstu Sparka. Patrz\u0105c z wysokiego poziomu, ka\u017cda aplikacja Sparka, sk\u0142ada si\u0119 z programu sterownika, kt\u00f3ry uruchamia na klastrze r\u00f3\u017cne r\u00f3wnoleg\u0142e procesy.Program sterownika ma\u00a0 dost\u0119p do Sparka przez obiektu kontekstowy (SparkContext). Maj\u0105\u0107 dost\u0119p do kontekstu mamy ju\u017c mo\u017cliwo\u015b\u0107 budowy obiekt\u00f3w zawieraj\u0105cych dane, kt\u00f3re pocz\u0105tkowo mog\u0105 si\u0119 nam wydawa\u0107 jako zwyk\u0142e i znane z Pythona struktury danych. Nic bardziej mylnego, obiekty znane jako\u00a0 Resilent Distributed Datasets (RDD) lub DataFrame (DF) s\u0105 umieszczane r\u00f3wnolegle na wielu w\u0119z\u0142ach zwanych\u00a0 wykonawcami (worker node).<\/p>\n
Na RDD b\u0119dziemy wykonywa\u0107 r\u00f3\u017cne metody, kt\u00f3re dzielimy na transformacje<\/em> i akcje<\/em>. Transformacje przekszta\u0142caj\u0105 nam RDD w inne RDD, za\u015b akcje generuj\u0105 z RDD wynik kt\u00f3ry mo\u017ce by\u0107 zwr\u00f3cony do sterownika b\u0105d\u017a zapisany w HDFS\u2019ie lub innym obs\u0142ugiwanym \u017ar\u00f3dle danych. Wszystkie transformacje s\u0105 leniwe (lazy), co oznacza , b\u0119d\u0105 wykonane wy\u0142\u0105cznie wtedy, gdy b\u0119dzie to konieczne. Zamiast manipulacji danymi mamy tu jedynie rejestracj\u0119 metadanych, kt\u00f3re pozwalaj\u0105 na zbudowanie RDD.<\/p>\n
from pyspark.sql import SQLContext\r\nurl = \"https:\/\/raw.githubusercontent.com\/guru99-edu\/R-Programming\/master\/adult_data.csv\"\r\nfrom pyspark import SparkFiles\r\nsc.addFile(url)\r\nsqlContext = SQLContext(sc)\r\nprint(sqlContext)\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
pyspark.sql.context.SQLContext object at 0x7fe1cf605438<\/pre>\n
Wczytajmy nasz plik jaka ramk\u0119 danych (dataframe)<\/p>\n
df = sqlContext.read.csv(SparkFiles.get(\"adult_data.csv\"), header=True, inferSchema= True)\t\r\n<\/pre>\n
Sprawd\u017amy jaka jest struktura tych danych.<\/p>\n
df.printSchema()\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
root\r\n |-- x: integer (nullable = true)\r\n |-- age: integer (nullable = true)\r\n |-- workclass: string (nullable = true)\r\n |-- fnlwgt: integer (nullable = true)\r\n |-- education: string (nullable = true)\r\n |-- educational-num: integer (nullable = true)\r\n |-- marital-status: string (nullable = true)\r\n |-- occupation: string (nullable = true)\r\n |-- relationship: string (nullable = true)\r\n |-- race: string (nullable = true)\r\n |-- gender: string (nullable = true)\r\n |-- capital-gain: integer (nullable = true)\r\n |-- capital-loss: integer (nullable = true)\r\n |-- hours-per-week: integer (nullable = true)\r\n |-- native-country: string (nullable = true)\r\n |-- income: string (nullable = true)\r\n<\/pre>\n
Zr\u00f3bmy kilka przyk\u0142ad\u00f3w z zastosowanie RDD.<\/p>\n
Utw\u00f3rzmy pierwszy obiekt RDD o nazwie rdd-1, do kt\u00f3rego za\u0142adowali\u015bmy list\u0119 sk\u0142adaj\u0105c\u0105 si\u0119 z 10 cyfr. To prosty obiekt, ale na jego podstawie poka\u017c\u0119 podstawowe transformacje i akcje.\u00a0 Utworzenie obiektu z danych wewn\u0119trznych (nie z plik\u00f3w) umo\u017cliwia nam funkcja parallelize<\/strong>.<\/p>\n
rdd_1=sc.parallelize([0,1,2,3,4,5,6,7,8,9])\r\nprint(rdd_1)\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:195<\/pre>\n
Pierwsz\u0105 transformacj\u0119, kt\u00f3r\u0105 mo\u017cna zrobi\u0107 jest map<\/strong>.<\/p>\n
Map bierze ka\u017cdy element zbioru RDD jako element wej\u015bciowy przekszta\u0142ca go za pomoc\u0105 funkcji w element wyj\u015bciowy. W poni\u017cszym przyk\u0142adzie przemno\u017cyli\u015bmy elementy przez liczb\u0119 dwa. Dla os\u00f3b, kt\u00f3re nie znaj\u0105 sk\u0142adni lambda, mamy tu do czynienia z budow\u0105 funkcji kt\u00f3ra przekszta\u0142ca element wej\u015bciowy na element wej\u015bciowy. Nie wiemy jaki jest typ elementu wej\u015bciowego, nie musi to by\u0107\u00a0 typ skalarny. Nie ma potrzeby budowania tego w postaci def + cia\u0142o funkcji. Sk\u0142adnia wygl\u0105da w uproszczeniu tak lambda wejscie: wyjscie,<\/strong> gdzie wyj\u015bcie zawiera algorytm.\u00a0 My\u015bl\u0119, ze po kilku poni\u017cszych przyk\u0142adach stanie si\u0119 to bardziej czytelne.<\/p>\n
rdd_2=rdd_1.map(lambda x: x*2)\r\nprint(rdd_2)<\/pre>\n
Na wyj\u015bciu:<\/p>\n
PythonRDD[1] at RDD at PythonRDD.scala:53<\/pre>\n
Wykonajmy akcj\u0119 collect<\/strong> na obu zbiorach, dzi\u0119ki czemu b\u0119dzie mo\u017cna zobaczy\u0107 ich zawarto\u015b\u0107. Uwaga, nie nale\u017cy stosowa\u0107 tej akcji na bardzo du\u017cych zbiorach danych, gdy\u017c mog\u0105 si\u0119 one nie zmie\u015bci\u0107 w lokalnej pami\u0119ci. Akcje powinny by\u0107 wykonywane tylko kiedy s\u0105 konieczne (!).<\/p>\n
rdd_1.collect()\r\nrdd_2.collect()\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]\r\n[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]\r\n<\/pre>\n
Nast\u0119pn\u0105 transformacj\u0119, z kt\u00f3rej mo\u017cemy skorzysta\u0107 jest flatMap<\/strong>.<\/p>\n
Funkcja\u00a0 dla ka\u017cdego elementu na wej\u015bciu zwraca kolekcj\u0119 element\u00f3w, w tym przypadku rozk\u0142adamy ka\u017cdy z napis\u00f3w na wyrazy przyjmuj\u0105\u0107 spacj\u0119 jako separator.<\/p>\n
rdd_lines = sc.parallelize([\"Pan Tadeusz\",\"Pan Samochodzik i Templariusze\", \"Poznajemy Sparka\", \"Dwadzie\u015bcia tysi\u0119cy mil podmorskiej \u017ceglugi\"])\r\nrdd_words = rdd_lines.flatMap(lambda line: line.split(\" \"))\r\nprint(rdd_words.collect())\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
['Pan', 'Tadeusz', 'Pan', 'Samochodzik', 'i', 'Templariusze', 'Poznajemy', 'Sparka', 'Dwadzie\u015bcia', 'tysi\u0119cy', 'mil', 'podmorskiej', '\u017ceglugi']<\/pre>\n
Kolejn\u0105 transformacj\u0105 jest operacja filtrowania filter<\/strong>, kt\u00f3ra tworzy podzbi\u00f3r spe\u0142niaj\u0105cy warunki funkcji lambda zwrac\u0105cej tylko warto\u015bci logiczne. W naszym przypadku wymagamy, by element by\u0142 mniejszy od sze\u015bciu<\/p>\n
rdd_numbers=sc.parallelize([0,1,2,3,4,5,6,7,8,9])\r\nrdd_numbers_filtered=rdd_numbers.filter(lambda x:<6)\r\nprint(rdd_numbers_filtered.collect())\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
[0, 1, 2, 3, 4, 5]\r\n<\/pre>\n
Kolejna transformacja o nazwie distinct<\/strong> umo\u017cliwia usuni\u0119cie zduplikowanych warto\u015bci w zbiorze.<\/p>\n
rdd_nonunique = sc.parallelize([1,2,2,3,4,4,5,6,7,8,8,9,9,9])\r\nrdd_unique = rdd_nonunique.distinct()\r\nprint(rdd_unique.collect())\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
[2, 4, 6, 8, 1, 3, 5, 7, 9]\r\n<\/pre>\n
Jak wyda\u0107 podw\u00f3jne liczby dwa, cztery, osiem i potr\u00f3jna dziewi\u0119\u0107 zosta\u0142y zredukowane do jednego wyst\u0105pienia..<\/p>\n
Istnieje mo\u017cliwo\u015b\u0107 utworzenia mniejszego zbioru danych za pomoc\u0105 transformacji sample<\/strong>, kt\u00f3ra przyjmuje trzy argumenty<\/p>\n
(czy zast\u0119powa\u0107, u\u0142amek liczby element\u00f3w w zbiorze wyj\u015bciowym\/wej\u015bciowym, ziarno dla generatora liczb pseudolosowych)<\/p>\n
rdd_numbers = sc.parallelize([0,1,2,3,4,5,6,7,8,9])\r\nrdd_samples = rdd_numbers.sample(False, 0.5,123)\r\nprint(rdd_samples.collect())\r\n<\/pre>\n
Na wyj\u015bciu.<\/p>\n
[1, 2, 3, 4, 6, 7]\r\n<\/pre>\n
Ustawienie ziarna na 123 oznacza, \u017ce z tego samego zbioru wej\u015bciowego zostanie wylosowany za ka\u017cdym razem taki sam zbi\u00f3r wyj\u015bciowy. Nale\u017cy o tym pami\u0119ta\u0107.<\/p>\n
 <\/p>\n
Wszystkie powy\u017csze przyk\u0142ady dotyczy\u0142y jednego zbioru RDD.<\/p>\n
Utw\u00f3rzmy dwa zbiory rdd_1 i\u00a0 rdd_2. Elementy w nich mog\u0105 si\u0119 pokrywa\u0107. Transformacja union<\/strong> umo\u017cliwia utworzenie nowego obiektu zawieraj\u0105cego ich sum\u0119.<\/p>\n
Sk\u0142adnia wygl\u0105da nast\u0119puj\u0105co\u00a0 rdd_wy_wyjsciu=rdd_pierwsze.union(rdd_drugie)<\/strong><\/p>\n
rdd_1 = sc.parallelize([1,2,3,4,5])\r\nrdd_2 = sc.parallelize([4,5,6,7,9])\r\nunion = rdd_1.union(rdd_2)\r\nprint(union.collect())<\/pre>\n
W posobny spos\u00f3b mo\u017cna skorzysta\u0107 z operacji intersetion<\/strong> (cz\u0119\u015b\u0107 wsp\u00f3lna), substract<\/strong> (r\u00f3\u017cnica, czyli elementy, kt\u00f3re s\u0105 w pierwszym obiekcie, a nie ma ich w drugim), cartesian<\/strong> (iloczyn kartezja\u0144ski). Sk\u0142adnia polece\u0144 jest analogiczna jak dla union<\/strong>.<\/p>\n
intersection = rdd_1.intersection(rdd_2)\r\nprint(\"intersection:\",intersection.collect())\r\nsubtract = rdd_1.subtract(rdd_2)\r\nprint(\"subtract:\",subtract.collect())\r\ncartesian = rdd_1.cartesian(rdd_2)\r\nprint(\"cartesian:\",cartesian.collect())<\/pre>\n
Na wyj\u015bciu:<\/p>\n
intersection: [4, 5]\r\nsubtract: [1, 2, 3]\r\ncartesian: [(1, 4), (1, 5), (2, 4), (2, 5), (1, 6), (1, 7), (2, 6), (2, 7), (1, 9), (2, 9), (3, 4), (3, 5), (4, 4), (4, 5), (5, 4), (5, 5), (3, 6), (3, 7), (4, 6), (4, 7), (3, 9), (4, 9), (5, 6), (5, 7), (5, 9)\r\n<\/pre>\n
Nadesz\u0142a pora na prezentacj\u0119 kilku akcji.<\/p>\n
Akcja reduce<\/strong> umo\u017cliwia zamian\u0119 wielu element\u00f3w zbioru na jeden. W naszym przyk\u0142adzie\u00a0 jest to mno\u017cenie<\/p>\n
rdd_numbers = sc.parallelize([1,2,3,4,5,6,7,8,9])\r\nrdd_multis = rdd_numbers.reduce(lambda x, y: x * y)\r\nprint(rdd_multis)\r\nprint(type(rdd_multis))\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
362880\r\n<class 'int'>\r\n<\/pre>\n
Istnieje te\u017c mo\u017cliwo\u015b\u0107 prostego przemno\u017cenia przez sta\u0142\u0105 warto\u015b\u0107. Umo\u017cliwia to operacja fold<\/strong>, kt\u00f3rej pierwszym elementem jest warto\u015b\u0107 przez kt\u00f3r\u0105 b\u0119dzie przemno\u017cony wynik.<\/p>\n
rdd_numbers = sc.parallelize([1,2,3,4,5])\r\nrdd_zero = rdd_numbers.fold(0, lambda x, y: x * y)\r\nprint(rdd_zero)\r\nprint(type(rdd_zero))\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
0\r\n<class 'int'>\r\n<\/pre>\n
Ten sam kod, ale zamiast zera zastosowali\u015bmy warto\u015b\u0107 minus jeden.<\/p>\n
rdd_numbers = sc.parallelize([1,2,3,4,5])\r\nrdd_minus = rdd_numbers.fold(-1, lambda x, y: x * y)\r\nprint(rdd_minus)\r\nprint(type(rdd_minus))\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
-120\r\n<class 'int'>\r\n<\/pre>\n
Na koniec zostawi\u0119 dwie akcje count<\/strong> oraz take<\/strong>.<\/p>\n
 <\/p>\n
Akcja count<\/strong> zwraca liczb\u0119 element\u00f3w danego obiektu RDD.<\/p>\n
rdd_numbers = sc.parallelize([1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20])\r\nrdd_numbers.count()\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
20\r\n<\/pre>\n
Akcja take<\/strong> zwraca pierwszych x element\u00f3w z obiektu RDD. W naszym przypadku\u00a0 jest to liczba dziesi\u0119\u0107.<\/p>\n
rdd_numbers = sc.parallelize([1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20])\r\nrdd_numbers.take(10)\r\n<\/pre>\n
Na wyj\u015bciu:<\/p>\n
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]\r\n<\/pre>\n
 <\/p>\n
Pokaza\u0142em jedynie kilka przyk\u0142ad\u00f3w wykorzystuj\u0105cych stos technologiczny Apache Sparka. \u015arodowisko Colab mo\u017cna wykorzysta\u0107 jednak jeszcze efektywniej korzystaj\u0105c z SSH i VNC. Wtedy istnieje mo\u017cliwo\u015b\u0107 uruchamiania gotowych skrypt\u00f3w i kontroli przy wykorzystaniu Spark GUI i\u00a0 prosty podgl\u0105d obiekt\u00f3w<\/p>\n
\n
    \n
  1. Jobs<\/a><\/li>\n
  2. Stages<\/a><\/li>\n
  3. Storage<\/a><\/li>\n
  4. Environment<\/a><\/li>\n
  5. Executors<\/a><\/li>\n<\/ol>\n<\/div>\n
     <\/p>\n
    O tym b\u0119dzie kolejny wpis, gdzie spr\u00f3bujemy przeanalizowa\u0107 prac\u0119 nad du\u017cym zbiorem.<\/p>\n
     <\/p>\n
    Przyk\u0142ady umie\u015bci\u0142em w repozytorium w postaci pliku .ipynb<\/p>\n
    https:\/\/github.com\/djkormo\/colab-examples\/blob\/master\/spark\/Colab_apache_spark_intro_primer.ipynb<\/a><\/p>\n
     <\/p>\n
    Literatura:<\/p>\n
     <\/p>\n
    \n
    Install Apache Spark on Ubuntu 19.04\/18.04 & Debian 10\/9\/8<\/a><\/p><\/blockquote>\n