Lab01 - Apache Spark [ ver. ChO.2025.11.03.003 ]

Apache Spark

Tematyka zajęć:

1. Apache Spark - środowiska uruchomieniowe
2. Pliki RDD, transformacje i akcje w Apache Spark
3. Środowisko Jupyter. Skrypty w języku Scala - struktury RDD
4. Środowisko Jupyter. Scala - struktury DataFrames
5. Środowisko Jupyter. Scala - przetwarzanie plików CSV
6. Środowisko Jupyter. Python - obsługa struktur RDD
7. Środowisko Jupyter. Python - obsługa struktur DataFrames
8. Środowisko Jupyter. Python - przetwarzanie plików tekstowych
9. Środowisko Jupyter - Python - przetwarzanie danych z relacyjnej bazy danych
10. Zadania do realizacji

Pliki do realizacji zadań:

• Plik dane1.csv (plik dostępny po zalogowaniu na stronie)
• Plik error.csv (plik dostępny po zalogowaniu na stronie)
• Plik access.csv (plik dostępny po zalogowaniu na stronie)

A. Apache Spark - narzędzie Spark i środowiska uruchomieniowe

1. Program Spark - punkt startowy do analizy danych BigData:
   • tworzymy aplikacje w lokalnym środowisku korzystając z interfejsów wysokiego poziomu,
   • interaktywne korzystanie i stosowanie złożonych algorytmów,
   • różne metody przetwarzania danych: zapytania SQL, przetwarzanie tekstów, systemy uczące i czy przetwarzające grafy,
   • przygotowane aplikacje można uruchomić na wielu węzłach uruchomionych w ramach klastra Spark.
2. Oprogramowanie Apache Spark można uruchomić w ramach systemu Linux i Windows jako aplikacje jednostanowiskowe. Do realizacji projektów w środowisku jednostanowiskowym można skorzystać z platformy Jupyter w której zintegrowano program Spark w ramach narzędzia pyspark
3. Oprogramowanie można zainstalować lokalnie lub korzystając z przygotowanego kontenera Docker.
   • Środowisko Windows - instalacja środowiska uruchomieniowego Java, instalacja oprogramowania Apache Spark oraz oprogramowania Anaconda (środwisko do realizacji projektów w języku python). Przydatne linki:
     • Guide to install Spark and use PySpark from Jupyter in Windows
     • Install PySpark to run in Jupyter Notebook on Windows
     • Get Started with PySpark and Jupyter Notebook in 3 Minutes
   • Kontener z oprogramowaniem Spark i Jupyter - instalacja w systemie Windows, wymagana możliwość uruchamiania kontenerów w modzie Linux (Windows 10). Przydatne linki:
     • Getting Started with PySpark for Big Data Analytics using Jupyter Notebooks and Jupyter Docker Stacks
     • Tutorial: Running a Dockerized Jupyter Server for Data Science
   • Instalacja opragomowania w systemie Linux ( zarówno natywna jak i w kontenerze Docker)
4. Środowisko uruchomieniowe Jupyter.

B. Pliki RDD, transformacje i akcje w Spark

1. Środowisko Apache Spark zawiera kilka elementów. Rysnek 1 przedstawia ekosystem programu Apache Spark.

   

2. Przetwarzanie zadań w środowisku Apache Spark (rys.2).

   

3. Zadania w ramach technologii Spark wykonywane są poprzez akcje i transformacje. W trakcie przetwarzania danych Spark tworzy pliki RDD ( Resilient Distributed Datasets ), które są niemodyfikowalne. Poprzez transforamcję otrzymujemy kolejny plik RDD. Do zapisu lub prezenatcji wykorzystujemy akcje. Istnieje możliwość odczytu danych z różnych źródeł tj. standardowe pliki CSV, pliki systemu HDFS, bazy danych czy dane strumieniowe. Po przetworzeniu danych system Spark umożliwia zapis danych do standardowych plików, plików systemu HDFS lub do baz danych typu HBase czy Cassandra.

   Przykładowe transformacje plików w systemie Spark.

   map (func)

   Zwraca nowy rozproszony zestaw danych utworzony przez przepuszczenie każdego elementu źródła przez funkcję func.

   flatMap (func)

   Stosuje funkcję na każdym elemencie w RDD i zwraca RDD z zawartością zwróconych iteratorów. Często stosowana do wyodrębnienia słów.

   filter (func)

   Zwraca nowy zestaw danych utworzony przez wybranie tych elementów źródła, dla których func zwraca true

   sample (withRe placement, fraction, [seed])

   Próbkowanie RDD z zastępowaniem lub bez.

   distinct ([numTasks]))

   Zwraca nowy zestaw danych, który zawiera różne elementy źródłowego zestawu danych

   union (otherDataset)

   Zwraca nowy zestaw danych, który zawiera połączenie elementów w źródłowym zestawie danych i zawartych w argumencie.

   intersection (otherDataset)

   Zwraca nowy RDD, który zawiera przecięcie elementów zawartych w źródłowym zbiorze danych i zawartym w argumencie.

   subtract (otherDataset)

   Zwraca nowy RDD, który usuwa zawartość RDD zawartego w argumencie.

   join (otherDataset, [numTasks])

   Gdy jest wywoływany w zestawach danych typu (K, V) i (K, W), zwraca zestaw danych par (K, (V, W)) ze wszystkimi parami elementów dla każdego klucza. Połączenia zewnętrzne są obsługiwane przez leftOuterJoin, rightOuterJoin i fullOuterJoin.

   groupByKey ([numTasks])

   reduceByKey (func, [numTasks])

   Łączy elementy z takim samym kluczem.

   aggregateByKey (zeroValue) (seqOp, combOp, [numTasks])

   sortByKey (asceding=True, numPartitions=None, keyfunc =<function <lambda>>)

   Zwraca RDD posortowany według klucza.

   keys()

   Zwraca RDD złożony z samych kluczy.

   values()

   Zwraca RDD złożony z samych wartości.

   Przykładowe akcje realizowane w ramach programu Apache Spark.

   collect ()

   Zwraca wszystkie elementy zestawu danych jako tablicę w programie sterownika. Jest to zwykle przydatne po filtrze lub innej operacji, która zwraca wystarczająco mały podzbiór danych.

   count ()

   Zwraca liczbę elementów w zestawie danych.

   countByValue ()

   Liczba wystąpień każdego elementu w RDD.

   take(num)

   Zwraca num elementów z RDD.

   takeSample(withReplacement, num, [seed])

   Zwraca num elementów w sposób losowy.

   first()

   Zwraca pierwszy element.

   reduce (func)

   Agreguje elementy zestawu danych za pomocą funkcji func (która pobiera dwa argumenty i zwraca jeden). Funkcja powinna być przemienna i asocjacyjna, aby mogła być poprawnie przetwarzana równolegle.

   countByKey()

   Liczy elementy każdego klucza.

   lookup(key)

   Zwraca wszystkie wartości związane z podanym kluczem.

C. Środowisko Jupyter. Skrypty w języku Scala - struktury RDD

• Na potrzeby zajęć została uruchomiana maszyna wirtualna udostępniającą serwis Spark wraz z notatnikiem Jupyter. Dostęp do serwisu jest z sieci wydziałowej pod adresem http://172.20.45.100:10000/. Konto i hasło użytkownika zostanie podane na zajęciach.

  W ramach notatnika Jupyter tworzymy nowy projekt wybierając z menu przycisku "New" - Apache Toree - Scala.

  

• Pierwszy skrypt w powłoce SparkContext.

  val data = Array.range(1,30)
  val rdd = sc.parallelize(data)
  rdd.collect() 
  

• Wybór elementów z RDD - first(), take() i takeSample().

  rdd.first()
  

  rdd.take(3)
  

  rdd.takeSample(true,10)
  

• Przetwarzanie danych z wykorzystaniem transformacji map().

  val rdd2 = rdd.map( x => x*x )
  rdd2.collect()
  

  val rdd2 = rdd.map( x => List(x,x) )
  rdd2.collect()
  

  val rdd2 = rdd.flatMap( x => List(x,x))
  rdd2.collect()
  

• Przetwarzanie danych z wykorzystaniem akcji reduce().

  val value = rdd.reduce( (t1,t2) => t1 + t2)
  

  val rdda = sc.parallelize( List( "aa","bb","cc","dd","ee","ff","gg" ) )
  val value = rdda.reduce( (t1, t2) => t1+t2)
  

• Przetwarzanie danych z wykorzystaniem akcji count().

  rdd.count()
  

• Przetwarzanie danych z wykorzystaniem transformacji union(), distinct() i itersection().

  val data1 = Array.range(1,21)
  val data2 = Array.range(19,25)
  val rdd1 = sc.parallelize(data1)
  val rdd2 = sc.parallelize(data2)
  val rdd3 = rdd1.union(rdd2)
  rdd3.collect()
  

  val rdd4 = rdd3.distinct()
  rdd4.collect()
  

  rdd3.count()
  

  rdd4.count()
  

  val rdd5 = rdd1.intersection(rdd2)
  rdd5.collect()
  

• Przetwarzanie danych z wykorzystaniem akcji countByKey().

  val rdda1 = sc.parallelize(List("aa","bb","cc","dd","aa","cc","ee","ff","dd","dd","aa"))
  val rdda2 = rdda1.map( k => (k,1))
  rdda2.countByKey()
  

• Przetwarzanie danych z wykorzystaniem akcji join(), leftOuterJoin(), rightOuterJoin().

  val rddr1 = sc.parallelize(List("aa","bb","cc","dd","ee","ff","gg","aa")).map( k => (k,1))
  val rddr2 = sc.parallelize(List("aa","cc","mm","rr","tt")).map( k => (k,1))
  rddr1.join(rddr2).collect()
  

  rddr1.leftOuterJoin(rddr2).collect()
  

  rddr1.rightOuterJoin(rddr2).collect()
  

D. Środowisko Jupyter. Scala - struktury DataFrames

• Tworzenie zbioru DataFrames z pliku RDD.

  val data = Seq(("Java", "20000"), ("Python", "100000"), ("Scala", "3000"))
  val rddt = sc.parallelize(data)
  rddt.collect()
  

  val dfFromrddt = rddt.toDF()
  dfFromrddt.printSchema()
  

  val dfFromrddt = rddt.toDF("language","users_count")
  dfFromrddt.printSchema()
  

  val columns = Seq("language","users_count")
  val dfFromrddt2 = spark.createDataFrame(rddt).toDF(columns:_*)
  

  dfFromrddt2.show()
  

E. Środowisko Jupyter. Scala - przetwarzanie plików CSV

• Wczytanie pliku tekstowego CSV do struktury RDD (poprzez mechanizm dla pliku tekstowego).

  val rdd_csv2 = sc.textFile("Dane/dane1.csv")
  rdd_csv2.collect()
  

  rdd_csv2.count()
  

• Przetworzenie rekordów do struktury tablicy.

  val rdd_array = rdd_csv2.map( line => line.split('|'))
  rdd_array.collect()
  

• Wyszukanie informacji w dokumencie.

  rdd_array.filter(line => line.contains("Robert") ).collect()
  

  rdd_csv2.filter(line => line.contains("Robert") ).collect()
  

• Wczytanie pliku tekstowego CSV do struktury RDD - DF (poprzez mechanizm dla pliku CSV).

  val rdd_csv = spark.read.option("header","false").csv("Dane/dane1.csv")
  rdd_csv.collect()
  

  rdd_csv.count()
  

  rdd_csv.show()
  

  rdd_csv.printSchema()
  

• Zapis i odczyt pliku tekstowego. Sprawdzamy w systemie plików utoworzony plik.

  rdd.collect()
  

  rdd.saveAsTextFile("../Files/test.txt")
  

  val rdd_txt = sc.textFile("../Files/test.txt")
  rdd_txt.collect()
  

F. Środowisko Jupyter. Python - obsługa struktur RDD

• Sprawdzenie poprawności działania środowiska pyspark (brak błędów - poprawne działanie).

  import findspark
  findspark.init()
  

  from pyspark import SparkConf
  from pyspark import SparkContext
  sc = SparkContext.getOrCreate(SparkConf().setMaster("local[*]"))
  

• Pierwszy skrypt w powłoce SparkContext.

  rdd = sc.parallelize(range(30))
  rdd.collect() 
  

• Pierwsze elementy z zbioru danych.

  rdd.first()
  

• Kolejne elementy z zbioru ( listy, tablicy ).

  rdd.take(3)
  

• Wybrane losowo elementy z zbioru (listy, tablicy), sprawdzić dla parametru TRUE i FALSE.

  rdd.takeSample(True,10)
  

• Przetwarzanie danych z wykorzystaniem transformacji map().

  rdd2 = rdd.map(lambda x: x*x)
  rdd2.collect()
  

• Przetwarzanie danych z wykorzystaniem transformacji map().

  rdd2 = rdd.map(lambda x: [x,x])
  rdd2.collect()
  

• Przetwarzanie danych z wykorzystaniem transformacji map().

  rdd2 = rdd.flatMap(lambda x: [x,x])
  rdd2.collect()
  

• Przetwarzanie danych z wykorzystaniem akcji reduce().

  val = rdd.reduce(lambda t1, t2: t1+t2)
  print (val)
  

  rdda = sc.parallelize(['aa','bb','cc','dd','ee','ff','gg'])
  val = rdda.reduce(lambda t1, t2: t1+t2)
  print (val)
  

• Przetwarzanie danych z wykorzystaniem akcji count().

  rdd.count()
  

• Przetwarzanie danych z wykorzystaniem transformacji union().

  array1 = [("physics",85),("maths",75),("chemistry",95)]
  array2 = [("physics",65),("maths",45),("chemistry",85)]
  rdd_arr1 = sc.parallelize(array1)
  rdd_arr2 = sc.parallelize(array2)
  rdd_arr1.union(rdd_arr2).collect()
  

  rdd1 = sc.parallelize(range(1,20))
  rdd2 = sc.parallelize(range(10,25))
  rdd3 = rdd1.union(rdd2)
  rdd3.collect()
  

  rdd3.count()
  

• Przetwarzanie danych z wykorzystaniem transformacji distinct().

  rdd4 = rdd3.distinct()
  rdd4.collect()
  

  rdd4.count()
  

• Przetwarzanie danych z wykorzystaniem transformacji intersection().

  rdd4 = rdd1.intersection(rdd2)
  rdd4.collect()
  

• Przetwarzanie danych z wykorzystaniem akcji countByKey().

  rdd11a = sc.parallelize(('aa','bb','cc','dd','aa','cc','ee','ff','dd','dd','aa'))
  rdd11b = rdd11a.map(lambda k: (k,1))
  rdd11b.countByKey().items()
  

• Przetwarzanie danych z wykorzystaniem akcji join(), leftOuterJoin(), rightOuterJoin().

  rdda1 = sc.parallelize(('aa','bb','cc','dd','ee','ff','gg','aa')).map(lambda k: (k,1))
  rdda2 = sc.parallelize(('aa','cc','mm','rr','tt')).map(lambda k: (k,1))
  rdda1.join(rdda2).collect()
  

  rdda1.leftOuterJoin(rdda2).collect()
  

  rdda1.rightOuterJoin(rdda2).collect()
  

G. Środowisko Jupyter. Python - obsługa struktur DataFrames

• Tworzenie zbioru DataFrames z pliku RDD.

  from pyspark.sql.dataframe import DataFrame
  from pyspark.sql import SparkSession
  spark = SparkSession(sc)
  

  rdd = sc.parallelize([(1, 2, 3, 'a b c'),
               (4, 5, 6, 'd e f'),
               (7, 8, 9, 'g h i')])
  df = rdd.toDF(['col1', 'col2', 'col3','col4'])
  df.show()  
  df.printSchema()
  

• Tworzenie zbioru DataFrames z danych źródłowych.

  dfe = spark.createDataFrame([
                          ('1', 'Joe',   '70000', '1'),
                          ('2', 'Henry', '80000', '2'),
                          ('3', 'Sam',   '60000', '2'),
                          ('4', 'Max',   '90000', '1')],
                          ['Id', 'Name', 'Sallary','DepartmentId']
                         )
  dfe.show()
  dfe.printSchema()
  

• Tworzenie zbioru DataFrames z danych źródłowych (typy danych).

  from pyspark.sql.functions import udf, array
  from pyspark.sql.types import StringType
  from datetime import datetime as Date
  data = [
  [10,'Direct Sales',Date(2019,1,1)],
  [12,'Direct Sales',Date(2019,1,2)],
  [20,'Online Sales',Date(2019,1,1)],
  [25,'Online Sales',Date(2019,1,2)],
  ]
  df = spark.createDataFrame(data , ['Revenue','Department','Date']) 
  df.show()
  

H. Środowisko Jupyter. Python - przetwarzanie plików tekstowych

• Wczytanie pliku tekstowego CSV do struktury RDD.

  rdd_csv = sc.textFile('Files/dane1.csv')
  rdd_csv.collect()
  

  rdd_csv.count()
  

• Przetworzenie rekordów do struktury tablicy.

  rdd_array = rdd_csv.map(lambda line: line.split('|'))
  rdd_array.collect()
  

• Wyszukanie informacji w dokumencie.

  rdd_array.filter(lambda line: 'Robert' in line).collect()
  

• Grupowanie danych: lista osób na poszczególnych stopniach studiów.

  rdd_study = rdd_array.map(lambda n: (str(n[3]),str(n[1]))).groupByKey()
  rdd_study.map(lambda x: {x[0]: list(x[1])}).collect()
  

• Wczytanie danych z pliku CSV do struktury DataFrames.

  df = spark.read.format('csv').options(header='false',delimiter='|').load("../Files/dane1.csv",header=False)
  df.show(10)
  df.printSchema()
  

• Przetwarzanie dokumentu typu log, Wczytanie pliku tekstowego do struktury RDD.

  rdd_log = sc.textFile('../Files/error.csv')
  rdd_log.collect()
  

  rdd_log.count()
  

• Wyszukanie informacji w dokumencie.

  rdd_log.filter(lambda s: '[php7:error]' in s ).take(10)
  

J. Zadania do realizacji

Zadanie 1

W ramach zadania należy opracować polecenie tworzącą listę osób na poszczególnych stopniach studiów wg. wzoru: [stopień] { [Nazwisko, Imię, semestr], ... }. Rysunek 3 przedstawia przykładowy wynik polecenia.

Zadanie 2

W ramach zadania należy opracować polecenie zwracające statystykę imion. Rysunek 4 przedstawia przykładowy wynik polecenia.