1. Przegląd
Z tego ćwiczenia dowiesz się, jak utworzyć potok przetwarzania danych za pomocą Apache Spark z Dataproc na Google Cloud Platform. Odczytywanie danych z jednej lokalizacji pamięci masowej, przekształcanie ich i zapisywanie w innej lokalizacji pamięci masowej to typowy przypadek użycia w naukach o danych i inżynierii danych. Typowe transformacje obejmują zmianę zawartości danych, usunięcie niepotrzebnych informacji i zmianę typów plików.
Z tego ćwiczenia dowiesz się więcej o Apache Spark, uruchomisz przykładowy potok przy użyciu Dataproc z PySpark (interfejs API Pythona Apache Spark), BigQuery, Google Cloud Storage i danych z Reddita.
2. Wprowadzenie do Apache Spark (opcjonalnie)
Według strony internetowej „Apache Spark to ujednolicony mechanizm analityczny do przetwarzania danych na dużą skalę”. Umożliwia analizowanie i przetwarzanie danych równolegle i w pamięci, co pozwala na masowe obliczenia równoległe na wielu różnych maszynach i węzłach. Został on pierwotnie wydany w 2014 r. jako ulepszenie tradycyjnego MapReduce i nadal jest jedną z najpopularniejszych platform do wykonywania obliczeń na dużą skalę. Apache Spark jest napisany w języku Scala i ma interfejsy API w językach Scala, Java, Python i R. Zawiera wiele bibliotek, takich jak Spark SQL do wykonywania zapytań SQL dotyczących danych, Spark Streaming do strumieniowania danych, MLlib do uczenia maszynowego i GraphX do przetwarzania grafów. Wszystkie te biblioteki działają w silniku Apache Spark.
Spark może działać samodzielnie lub korzystać z usługi zarządzania zasobami, takiej jak Yarn, Mesos lub Kubernetes, w celu skalowania. W tych ćwiczeniach z programowania będziesz używać Dataproc, który korzysta z Yarn.
Dane w Sparku były pierwotnie wczytywane do pamięci w postaci tzw. RDD, czyli odpornego rozproszonego zbioru danych. W ramach rozwoju Sparka dodano 2 nowe typy danych w stylu kolumnowym: zbiór danych (Dataset), który jest typowany, oraz ramkę danych (Dataframe), która nie jest typowana. Ogólnie rzecz biorąc, RDD doskonale sprawdzają się w przypadku każdego rodzaju danych, natomiast zbiory danych i ramki danych są zoptymalizowane pod kątem danych w formie tabeli. Zbiory danych są dostępne tylko w interfejsach Java API i Scala API, więc w tym ćwiczeniu użyjemy interfejsu PySpark Dataframe API. Więcej informacji znajdziesz w dokumentacji Apache Spark.
3. Przypadek użycia
Inżynierowie danych często potrzebują łatwego dostępu do danych dla analityków danych. Dane są jednak często początkowo nieuporządkowane (w obecnym stanie trudno je wykorzystać do analizy) i wymagają oczyszczenia, zanim będą przydatne. Przykładem takich danych są informacje pobrane z internetu, które mogą zawierać nietypowe kodowania lub zbędne tagi HTML.
W tym module wczytasz zbiór danych z BigQuery w postaci postów z Reddita do klastra Sparka hostowanego w Dataproc, wyodrębnisz przydatne informacje i zapiszesz przetworzone dane jako skompresowane pliki CSV w Google Cloud Storage.
Główny specjalista ds. danych w Twojej firmie chce, aby jego zespoły zajęły się różnymi problemami związanymi z przetwarzaniem języka naturalnego. Chce on przeanalizować dane z subredditu „r/food”. Utworzysz potok do zrzutu danych, zaczynając od wypełnienia wstecznego od stycznia 2017 r. do sierpnia 2019 r.
4. Dostęp do BigQuery za pomocą interfejsu BigQuery Storage API
Pobieranie danych z BigQuery za pomocą metody tabledata.list API może być czasochłonne i nieefektywne w przypadku dużych ilości danych. Ta metoda zwraca listę obiektów JSON i wymaga sekwencyjnego odczytywania po jednej stronie naraz, aby odczytać cały zbiór danych.
BigQuery Storage API znacznie usprawnia dostęp do danych w BigQuery dzięki zastosowaniu protokołu opartego na RPC. Obsługuje równoległe odczytywanie i zapisywanie danych, a także różne formaty serializacji, takie jak Apache Avro i Apache Arrow. W praktyce oznacza to znacznie większą wydajność, zwłaszcza w przypadku większych zbiorów danych.
W tym ćwiczeniu użyjesz spark-bigquery-connector do odczytywania i zapisywania danych między BigQuery a Sparkiem.
5. Tworzenie projektu
Zaloguj się w konsoli Google Cloud Platform na stronie console.cloud.google.com i utwórz nowy projekt:
Następnie musisz włączyć płatności w konsoli Cloud, aby móc korzystać z zasobów Google Cloud.
Wykonanie tego samouczka nie powinno kosztować więcej niż kilka dolarów, ale może okazać się droższe, jeśli zdecydujesz się wykorzystać więcej zasobów lub pozostawisz je uruchomione. W ostatniej sekcji tego ćwiczenia dowiesz się, jak wyczyścić projekt.
Nowi użytkownicy Google Cloud Platform mogą skorzystać z bezpłatnego okresu próbnego, w którym mają do dyspozycji środki w wysokości 300 USD.
6. Konfigurowanie środowiska
Teraz skonfigurujesz środowisko, wykonując te czynności:
- Włączanie interfejsów Compute Engine, Dataproc i BigQuery Storage API
- Konfigurowanie ustawień projektu
- Tworzenie klastra Dataproc
- Tworzenie zasobnika Google Cloud Storage
Włączanie interfejsów API i konfigurowanie środowiska
Otwórz Cloud Shell, klikając przycisk w prawym górnym rogu Cloud Console.
Po załadowaniu Cloud Shell uruchom te polecenia, aby włączyć interfejsy Compute Engine API, Dataproc API i BigQuery Storage API:
gcloud services enable compute.googleapis.com \
dataproc.googleapis.com \
bigquerystorage.googleapis.com
Ustaw identyfikator projektu. Aby go znaleźć, otwórz stronę wyboru projektu i wyszukaj swój projekt. Może się różnić od nazwy projektu.
Uruchom to polecenie, aby ustawić identyfikator projektu:
gcloud config set project <project_id>
Ustaw region projektu, wybierając go z listy tutaj. Przykładem może być us-central1.
gcloud config set dataproc/region <region>
Wybierz nazwę klastra Dataproc i utwórz dla niej zmienną środowiskową.
CLUSTER_NAME=<cluster_name>
Tworzenie klastra Dataproc
Utwórz klaster Dataproc, wykonując to polecenie:
gcloud beta dataproc clusters create ${CLUSTER_NAME} \
--worker-machine-type n1-standard-8 \
--num-workers 8 \
--image-version 1.5-debian \
--initialization-actions gs://dataproc-initialization-actions/python/pip-install.sh \
--metadata 'PIP_PACKAGES=google-cloud-storage' \
--optional-components=ANACONDA \
--enable-component-gateway
Wykonanie tego polecenia zajmie kilka minut. Aby rozłożyć polecenie na części:
Spowoduje to rozpoczęcie tworzenia klastra Dataproc o nazwie podanej wcześniej. Korzystanie z interfejsu beta API umożliwi korzystanie z funkcji Dataproc w wersji beta, takich jak Component Gateway.
gcloud beta dataproc clusters create ${CLUSTER_NAME}
Ustawia to typ maszyny, której mają używać instancje robocze.
--worker-machine-type n1-standard-8
Spowoduje to ustawienie liczby instancji roboczych w klastrze.
--num-workers 8
Spowoduje to ustawienie wersji obrazu Dataproc.
--image-version 1.5-debian
Spowoduje to skonfigurowanie działań inicjowania, które będą używane w klastrze. W tym przypadku uwzględniasz działanie inicjowania pip.
--initialization-actions gs://dataproc-initialization-actions/python/pip-install.sh
Są to metadane, które mają być uwzględnione w klastrze. W tym miejscu podajesz metadane pipdziałania inicjującego.
--metadata 'PIP_PACKAGES=google-cloud-storage'
Spowoduje to zainstalowanie w klastrze komponentów opcjonalnych.
--optional-components=ANACONDA
Włączy to bramę komponentów, która umożliwia korzystanie z bramy komponentów Dataproc do wyświetlania popularnych interfejsów, takich jak Zeppelin, Jupyter czy historia Spark.
--enable-component-gateway
Bardziej szczegółowe wprowadzenie do Dataproc znajdziesz w tych ćwiczeniach z programowania.
Tworzenie zasobnika Google Cloud Storage
Do przechowywania danych wyjściowych zadania potrzebny jest zasobnik Google Cloud Storage. Wybierz unikalną nazwę zasobnika i uruchom to polecenie, aby utworzyć nowy zasobnik. Nazwy zasobników są unikalne we wszystkich projektach Google Cloud dla wszystkich użytkowników, więc może być konieczne kilkukrotne wypróbowanie różnych nazw. Jeśli nie otrzymasz odpowiedzi ServiceException, zasobnik zostanie utworzony.
BUCKET_NAME=<bucket_name>
gsutil mb gs://${BUCKET_NAME}
7. Wstępna analiza danych
Zanim przeprowadzisz wstępne przetwarzanie, dowiedz się więcej o rodzaju danych, z którymi masz do czynienia. W tym celu poznasz 2 metody eksploracji danych. Najpierw wyświetlisz niektóre dane pierwotne za pomocą interfejsu internetowego BigQuery, a potem obliczysz liczbę postów w każdym subreddicie za pomocą PySpark i Dataproc.
Korzystanie z interfejsu internetowego BigQuery
Zacznij od wyświetlenia danych w interfejsie internetowym BigQuery. W konsoli Cloud kliknij ikonę menu, przewiń w dół i kliknij „BigQuery”, aby otworzyć interfejs internetowy BigQuery.
Następnie uruchom to polecenie w edytorze zapytań w interfejsie BigQuery w przeglądarce. Spowoduje to zwrócenie 10 pełnych wierszy danych ze stycznia 2017 r.:
select * from fh-bigquery.reddit_posts.2017_01 limit 10;
Przewiń stronę, aby zobaczyć wszystkie dostępne kolumny i przykłady. Zobaczysz zwłaszcza 2 kolumny, które zawierają tekstową treść każdego posta: „title” (tytuł) i „selftext” (treść posta). Zwróć też uwagę na inne kolumny, takie jak „created_utc”, która zawiera czas UTC utworzenia posta, oraz „subreddit”, która zawiera nazwę subredditu, w którym znajduje się post.
Wykonywanie zadania PySpark
Aby skopiować repozytorium z przykładowym kodem i przejść do odpowiedniego katalogu, uruchom w Cloud Shell te polecenia:
cd
git clone https://github.com/GoogleCloudPlatform/cloud-dataproc
Za pomocą PySpark możesz określić liczbę postów w każdym subreddicie. Możesz otworzyć edytor Cloud i przeczytać skrypt cloud-dataproc/codelabs/spark-bigquery przed jego wykonaniem w następnym kroku:
W edytorze Cloud kliknij przycisk „Otwórz terminal”, aby wrócić do Cloud Shell i uruchomić to polecenie, aby wykonać pierwsze zadanie PySpark:
cd ~/cloud-dataproc/codelabs/spark-bigquery
gcloud dataproc jobs submit pyspark --cluster ${CLUSTER_NAME} \
--jars gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar \
--driver-log-levels root=FATAL \
counts_by_subreddit.py
To polecenie umożliwia przesyłanie zadań do Dataproc za pomocą interfejsu Jobs API. W tym miejscu wskazujesz typ zlecenia jako pyspark. Możesz podać nazwę klastra, parametry opcjonalne i nazwę pliku zawierającego zadanie. W tym przypadku podajesz parametr --jars, który umożliwia dołączenie spark-bigquery-connector do zadania. Możesz też ustawić poziomy wyjściowe logów za pomocą --driver-log-levels root=FATAL, co spowoduje pominięcie wszystkich danych wyjściowych logów z wyjątkiem błędów. Dzienniki Sparka są zwykle dość zaszumione.
Wykonanie tych poleceń zajmie kilka minut, a dane wyjściowe powinny wyglądać mniej więcej tak:
8. Poznawanie interfejsów Dataproc i Spark
Podczas uruchamiania zadań Spark w Dataproc masz dostęp do 2 interfejsów, w których możesz sprawdzać stan zadań i klastrów. Pierwszy to interfejs Dataproc, który znajdziesz, klikając ikonę menu i przewijając w dół do Dataproc. Możesz tu sprawdzić bieżącą dostępną pamięć, pamięć oczekującą i liczbę instancji roboczych.
Możesz też kliknąć kartę zadań, aby zobaczyć ukończone zadania. Szczegóły zadań, takie jak logi i dane wyjściowe, możesz wyświetlić, klikając identyfikator zadania. 
Możesz też wyświetlić interfejs Spark. Na stronie zadania kliknij strzałkę wstecz, a potem kliknij Interfejsy internetowe. W sekcji Brama komponentu powinno być widocznych kilka opcji. Wiele z nich można włączyć w sekcji Komponenty opcjonalne podczas konfigurowania klastra. W tym module kliknij „Serwer historii usługi Spark”.
Powinno się otworzyć to okno:
Wszystkie ukończone zadania będą widoczne w tym miejscu. Możesz kliknąć dowolny identyfikator application_id, aby dowiedzieć się więcej o danym zadaniu. Podobnie możesz kliknąć „Pokaż niekompletne zgłoszenia” u dołu strony docelowej, aby wyświetlić wszystkie obecnie uruchomione zadania.
9. Uruchamianie zadania uzupełniania
Teraz uruchomisz zadanie, które wczytuje dane do pamięci, wyodrębnia potrzebne informacje i zapisuje dane wyjściowe w zasobniku Google Cloud Storage. Wyodrębnisz „tytuł”, „treść” (czysty tekst) i „sygnaturę czasową utworzenia” każdego komentarza w serwisie Reddit. Następnie przekonwertuj te dane na plik CSV, spakuj go i wczytaj do zasobnika z identyfikatorem URI gs://${BUCKET_NAME}/reddit_posts/RRRR/MM/food.csv.gz.
Możesz ponownie otworzyć edytor Cloud, aby przeczytać kod dla cloud-dataproc/codelabs/spark-bigquery/backfill.sh, który jest skryptem otoki do wykonania kodu w cloud-dataproc/codelabs/spark-bigquery/backfill.py.
cd ~/cloud-dataproc/codelabs/spark-bigquery
bash backfill.sh ${CLUSTER_NAME} ${BUCKET_NAME}
Wkrótce zobaczysz kilka komunikatów o zakończeniu zadań. Wykonanie zadania może zająć do 15 minut. Możesz też sprawdzić zasobnik pamięci, aby potwierdzić, że dane zostały wyeksportowane, używając narzędzia gsutil. Gdy wszystkie zadania zostaną wykonane, uruchom to polecenie:
gsutil ls gs://${BUCKET_NAME}/reddit_posts/*/*/food.csv.gz
Powinny się wyświetlić te dane wyjściowe:
Gratulacje! Udało Ci się uzupełnić dane komentarzy w Reddicie. Jeśli chcesz dowiedzieć się, jak tworzyć modele na podstawie tych danych, przejdź do ćwiczenia Spark-NLP.
10. Czyszczenie
Aby uniknąć niepotrzebnych opłat na koncie Google Cloud Platform po ukończeniu tego samouczka:
- Usuń zasobnik Cloud Storage dla utworzonego środowiska.
- Usuń środowisko Dataproc.
Jeśli projekt został utworzony specjalnie na potrzeby tego ćwiczenia, możesz go też usunąć:
- W konsoli GCP otwórz stronę Projekty.
- Z listy projektów wybierz projekt, który chcesz usunąć, i kliknij Usuń.
- W polu wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.
Licencja
Ten utwór jest dostępny na licencji Creative Commons Attribution 3.0 Generic oraz Apache 2.0.