Szkielety obliczeniowe, takie jak Apache Spark, zostały szeroko zaadoptowane do budowania aplikacji danych o dużej skali. Dla firmy Uber dane są podstawą podejmowania strategicznych decyzji i rozwoju produktów. Aby pomóc nam lepiej wykorzystywać te dane, zarządzamy masowymi wdrożeniami Sparka w naszych globalnych biurach inżynieryjnych.

Pomimo że Spark sprawia, że technologia danych staje się bardziej dostępna, właściwy dobór zasobów przydzielonych do aplikacji Spark i optymalizacja wydajności operacyjnej naszej infrastruktury danych wymaga bardziej szczegółowego wglądu w te systemy, a mianowicie ich wzorców wykorzystania zasobów.

Aby wydobyć te wzorce bez zmiany kodu użytkownika, zbudowaliśmy i udostępniliśmy JVM Profiler, rozproszony profiler do zbierania metryk wydajności i wykorzystania zasobów oraz przekazywania ich do dalszej analizy. Chociaż został on stworzony dla Sparka, jego ogólna implementacja sprawia, że można go zastosować do każdej usługi lub aplikacji opartej na maszynie wirtualnej Java (JVM). Przeczytaj dalej, aby dowiedzieć się, jak Uber używa tego narzędzia do profilowania naszych aplikacji Spark, a także jak wykorzystać je w swoich własnych systemach.

Wyzwania związane z profilowaniem

Na co dzień firma Uber obsługuje dziesiątki tysięcy aplikacji działających na tysiącach maszyn. Wraz z rozwojem naszego stosu technologicznego szybko zdaliśmy sobie sprawę, że nasze istniejące rozwiązanie do profilowania i optymalizacji wydajności nie będzie w stanie spełnić naszych potrzeb. W szczególności, chcieliśmy mieć możliwość:

Korrelowania metryk przez dużą liczbę procesów na poziomie aplikacji

W środowisku rozproszonym wiele aplikacji Spark działa na tym samym serwerze, a każda aplikacja Spark ma dużą liczbę procesów (np. tysiące executorów) działających na wielu serwerach, jak pokazano na rysunku 1:

Nasze dotychczasowe narzędzia mogły monitorować jedynie metryki na poziomie serwera i nie mierzyły metryk dla poszczególnych aplikacji. Potrzebowaliśmy rozwiązania, które mogłoby zbierać metryki dla każdego procesu i korelować je między procesami dla każdej aplikacji. Dodatkowo, nie wiemy, kiedy te procesy zostaną uruchomione i jak długo potrwają. Aby móc zbierać metryki w tym środowisku, profiler musi być uruchamiany automatycznie z każdym procesem.

Uczynić zbieranie metryk nieinwazyjnym dla dowolnego kodu użytkownika

W swoich obecnych implementacjach, biblioteki Spark i Apache Hadoop nie eksportują metryk wydajnościowych, jednak często zdarzają się sytuacje, w których musimy zbierać te metryki bez zmiany kodu użytkownika lub frameworka. Na przykład, jeśli doświadczamy wysokich opóźnień na węźle NameNode rozproszonego systemu plików Hadoop (HDFS), chcemy sprawdzić opóźnienia obserwowane z każdej aplikacji Sparka, aby upewnić się, że te problemy nie zostały powielone. Ponieważ kody klienckie NameNode są osadzone wewnątrz naszej biblioteki Sparka, modyfikacja kodu źródłowego w celu dodania tej specyficznej metryki jest kłopotliwa. Aby nadążyć za ciągłym wzrostem naszej infrastruktury danych, musimy być w stanie wykonać pomiary dowolnej aplikacji w dowolnym momencie i bez wprowadzania zmian w kodzie. Co więcej, implementacja bardziej nieinwazyjnego procesu zbierania metryk umożliwiłaby nam dynamiczne wstrzykiwanie kodu do metod Java podczas ładowania.

Wprowadzenie JVM Profiler

Aby sprostać tym dwóm wyzwaniom, zbudowaliśmy i udostępniliśmy nasz JVM Profiler. Istnieją pewne istniejące narzędzia open source, takie jak statsd-jvm-profiler firmy Etsy, które mogą zbierać metryki na poziomie poszczególnych aplikacji, ale nie zapewniają one możliwości dynamicznego wstrzykiwania kodu do istniejących binarek Javy w celu zbierania metryk. Zainspirowani niektórymi z tych narzędzi, zbudowaliśmy nasz profiler z jeszcze większymi możliwościami, takimi jak profilowanie dowolnych metod/argumentów Javy.

Co robi JVM Profiler?

JVM Profiler składa się z trzech kluczowych funkcji, które ułatwiają zbieranie metryk wydajności i wykorzystania zasobów, a następnie dostarczają te metryki (np. Apache Kafka) do innych systemów w celu dalszej analizy:

• Agent Java: Poprzez włączenie agenta Java do naszego profilera, użytkownicy mogą zbierać różne metryki (np. użycie CPU/pamięci) i ślady stosu dla procesów JVM w sposób rozproszony.
• Zaawansowane możliwości profilowania: Nasz JVM Profiler pozwala na śledzenie dowolnych metod Javy i argumentów w kodzie użytkownika bez wprowadzania rzeczywistych zmian w kodzie. Ta funkcja może być używana do śledzenia opóźnień wywołań HDFS NameNode RPC dla aplikacji Spark i identyfikowania powolnych wywołań metod. Może również śledzić ścieżki plików HDFS, które czyta lub zapisuje każda aplikacja Spark, aby zidentyfikować gorące pliki do dalszej optymalizacji.
• Raportowanie analityki danych: W firmie Uber używamy profilera do raportowania metryk do tematów Kafka i tabel Apache Hive, dzięki czemu analityka danych jest szybsza i łatwiejsza.

Typowe przypadki użycia

Nasz JVM Profiler obsługuje wiele różnych przypadków użycia, przede wszystkim umożliwiając instrumentowanie dowolnego kodu Java. Używając prostej zmiany konfiguracji, JVM Profiler może być dołączony do każdego executora w aplikacji Spark i zbierać metryki uruchomienia metod Java. Poniżej omawiamy niektóre z tych przypadków użycia:

• Right-size executor: Używamy metryk pamięci z JVM Profiler, aby śledzić rzeczywiste wykorzystanie pamięci dla każdego executora, dzięki czemu możemy ustawić odpowiednią wartość dla argumentu Spark “executor-memory”.
• Monitorowanie opóźnienia HDFS NameNode RPC: Profilujemy metody na klasie org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB w aplikacji Spark i identyfikujemy długie opóźnienia na wywołaniach NameNode. Monitorujemy ponad 50 tysięcy aplikacji Spark każdego dnia z kilkoma miliardami takich wywołań RPC.
• Monitorowanie zdarzeń porzucania sterowników: Profilujemy metody takie jak org.apache.spark.scheduler.LiveListenerBus.onDropEvent, aby śledzić sytuacje, podczas których kolejka zdarzeń sterownika Spark staje się zbyt długa i upuszcza zdarzenia.
• Śledzenie pochodzenia danych: Profilujemy argumenty ścieżki pliku w metodzie org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.getBlockLocations i org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.addBlock, aby śledzić, jakie pliki są odczytywane i zapisywane przez aplikację Spark.

Szczegóły implementacji i rozszerzalność

Aby implementacja była jak najbardziej bezproblemowa, JVM Profiler ma bardzo prostą i rozszerzalną konstrukcję. Ludzie mogą łatwo dodać dodatkowe implementacje profilera, aby zebrać więcej metryk, a także wdrożyć własne niestandardowe raporty do wysyłania metryk do różnych systemów do analizy danych.

Kod JVM Profiler jest ładowany do procesu Java poprzez argument agenta Java po uruchomieniu procesu. Składa się on z trzech głównych części:

• Class File Transformer: przetwarza kod bajtowy metod Java wewnątrz procesu w celu profilowania dowolnego kodu użytkownika i zapisywania metryk w wewnętrznym buforze metrycznym.
• Metric Profileers
  • CPU/Memory Profiler: zbiera metryki użycia CPU/Memory przez JMX i wysyła je do raportów.
  • Profiler czasu trwania metody: odczytuje metryki czasu trwania metody (opóźnienia) z bufora metryk i wysyła je do raportujących.
  • Method Argument Profiler: odczytuje wartości argumentów metody z bufora metryk i wysyła do reporterów.
• Reporterzy
  • Console Reporter: zapisuje metryki w konsoli wyjściowej.
  • Reporter Kafka: wysyła metryki do tematów Kafka.

Jak rozszerzyć JVM Profiler, aby wysyłał metryki poprzez własny reporter

Użytkownicy mogą zaimplementować własne reportery i określić je z opcją -javaagent, jak:

java

-javaagent:jvm-profiler-0.0.5.jar=reporter=com.uber.profiling.reporters.CustomReporter

Integracja z infrastrukturą danych Ubera

Zintegrowaliśmy nasze metryki JVM Profiler z wewnętrzną infrastrukturą danych Ubera, aby umożliwić:

1. Analizę danych obejmującą cały klaster: Metryki są najpierw podawane do Kafki i pobierane do HDFS, a następnie użytkownicy zadają pytania za pomocą Hive/Presto/Spark.
2. Debugowanie aplikacji Spark w czasie rzeczywistym: Używamy Flink do agregowania danych dla pojedynczej aplikacji w czasie rzeczywistym i zapisywania do naszej bazy danych MySQL, a następnie użytkownicy mogą przeglądać metryki za pomocą interfejsu internetowego.

Używanie JVM Profiler

Poniżej przedstawiamy instrukcje, jak używać naszego JVM Profiler do śledzenia prostej aplikacji Java:

Po pierwsze, git klonujemy projekt:

Budujemy projekt, uruchamiając następujące polecenie:

Następnie wywołujemy plik JAR z wynikiem budowania (np.target/jvm-profiler-0.0.5.jar) i uruchamiamy aplikację wewnątrz JVM Profiler za pomocą następującego polecenia:

Polecenie uruchamia przykładową aplikację Java i raportuje jej wydajność oraz metryki wykorzystania zasobów na konsoli wyjściowej. Na przykład:

Profiler może również wysyłać metryki do tematu Kafka za pomocą polecenia takiego jak poniższe:

Użyj profilera do profilowania aplikacji Spark

Teraz, przejdźmy przez to jak uruchomić profiler z aplikacją Spark.

Zakładając, że mamy już klaster HDFS, wgrywamy plik JAR JVM Profiler do naszego HDFS:

Następnie używamy wiersza poleceń spark-submit, aby uruchomić aplikację Spark z profilerem:

Przykłady zapytań metrycznych

W Uber, wysyłamy nasze metryki do tematów Kafka i programujemy potoki danych w tle, aby automatycznie wprowadzić je do tabel Hive. Użytkownicy mogą skonfigurować podobne potoki i używać SQL do zapytań o metryki profilera. Mogą również napisać własne repetytory, które będą wysyłać metryki do bazy danych SQL, takiej jak MySQL i stamtąd zadawać pytania. Poniżej znajduje się przykładowy schemat tabeli Hive:

Poniżej przedstawiamy przykładowy wynik uruchomienia poprzedniego zapytania SQL, który pokazuje metryki pamięci i CPU dla każdego procesu dla executorów Spark:

role	processUuid	maxHeapMemoryMB	avgProcessCpuLoad
executor	6d3aa4ee-.4233-4cc7-a628-657e1a87825d	2805.255325	7.61E-11
executor	21f418d1-6d4f-440d-b28a-4eba1a3bb53d	3993.969582	9.56E-11
executor	a5da6e8c-149b-4722-8fa8-74a16baabcf8	3154.484474	8.18E-11
executor	a1994e27-59bd-4cda-9de3-745ead954a27	2561.847374	8.58E-11

Wyniki i następne kroki

Zastosowaliśmy JVM Profiler do jednej z największych aplikacji Spark firmy Uber (która używa ponad 1000 executorów), a w procesie tym zmniejszyliśmy przydział pamięci dla każdego executora o 2 GB, przechodząc z 7 GB do 5 GB. Tylko dla tej aplikacji Spark zaoszczędziliśmy 2TB pamięci.

Zastosowaliśmy również JVM Profiler do wszystkich aplikacji Hive on Spark wewnątrz Ubera i znaleźliśmy kilka możliwości poprawy wydajności wykorzystania pamięci. Rysunek 3 poniżej pokazuje jeden z rezultatów, który znaleźliśmy: około 70 procent naszych aplikacji używało mniej niż 80 procent przydzielonej pamięci. Nasze odkrycia wskazywały, że dla większości z tych aplikacji moglibyśmy przydzielić mniej pamięci i zwiększyć jej wykorzystanie o 20 procent.

W miarę dalszego rozwoju naszego rozwiązania oczekujemy dodatkowej redukcji pamięci w naszych maszynach JVM.

JVM Profiler jest niezależnym projektem open source i zachęcamy innych programistów do korzystania z tego narzędzia i wnoszenia wkładu (np. składania pull requestów)!

Nasz zespół Big Data Engineering w Seattle, Palo Alto i San Francisco pracuje nad narzędziami i systemami rozszerzającymi cały ekosystem Hadoop, w tym HDFS, Hive, Presto i Spark. Tworzymy technologie na bazie tej rodziny oprogramowania open source, aby pomóc nam w podejmowaniu lepszych decyzji biznesowych opartych na danych. Jeśli brzmi to dla Ciebie atrakcyjnie, sprawdź nasze oferty pracy i rozważ dołączenie do naszego zespołu!

Główne zdjęcie kredytowe: Seahorse, Roatan, Honduras autorstwa Conora Myhrvolda.

Zapisz się do naszego biuletynu, aby być na bieżąco z najnowszymi innowacjami od Uber Engineering.

.