Szanowni Czytelnicy, życzę miłego dnia!
W tym artykule główny konsultant w dziedzinie rozwiązań Big Data w firmie „Neoflex” szczegółowo opisuje różne opcje budowy zmiennych struktur danych przy użyciu Apache Spark.
W ramach projektu analizy danych często pojawia się potrzeba budowy witryn na podstawie słabo ustrukturyzowanych danych.
Zazwyczaj są to logi lub odpowiedzi różnych systemów, przechowywane w formacie JSON lub XML. Dane są eksportowane do Hadoop, a następnie należy z nich stworzyć witrynę. Dostęp do utworzonej witryny można zorganizować na przykład poprzez Impala.
W tym przypadku schemat docelowej witryny jest z góry nieznany. Co więcej, schemat nie może być również opracowany z wyprzedzeniem, ponieważ zależy od danych, a mamy do czynienia z tymi właśnie słabo ustrukturyzowanymi danymi.
Na przykład dzisiaj zarejestrowano taką odpowiedź:
{source: "app1", error_code: ""}a jutro z tego samego systemu przychodzi taka odpowiedź:
{source: "app1", error_code: "error", description: "Błąd sieci"}W rezultacie do witryny powinno dodać się dodatkowe pole — description, a to, czy przyjdzie, czy nie, nikt nie wie.
Zadanie stworzenia witryny na takich danych jest dość standardowe, a Spark oferuje szereg narzędzi do tego celu. Do parsowania danych źródłowych dostępne są wsparcie zarówno dla JSON, jak i XML, a dla schematu, którego nie można z góry określić, przewidziano wsparcie dla schemaEvolution.
Na pierwszy rzut oka rozwiązanie wydaje się proste. Należy wziąć folder z JSON i odczytać go w dataframe. Spark stworzy schemat, a zagnieżdżone dane zostaną przekształcone w struktury. Następnie wszystko należy zapisać w formacie parquet, który jest wspierany także przez Impala, rejestrując witrynę w Hive metastore.
Wydaje się, że wszystko jest proste.
Jednak z krótkich przykładów w dokumentacji nie wynika, jak radzić sobie z szeregiem problemów w praktyce.
Dokumentacja opisuje podejście nie do tworzenia witryny, ale do odczytu JSON lub XML w dataframe.
A konkretnie, przedstawia tylko, jak odczytać i sparsować JSON:
df = spark.read.json(path...)To wystarczy, aby udostępnić dane dla Sparka.
W praktyce jednak scenariusz jest znacznie bardziej skomplikowany niż tylko odczytanie plików JSON z folderu i stworzenie dataframe. Sytuacja wygląda następująco: już istnieje określona witryna, każdego dnia przychodzą nowe dane, które należy dodać do witryny, nie zapominając, że schemat może się różnić.
Typowy schemat budowy witryny wygląda tak:
Krok 1. Dane są ładowane do Hadoop z codziennym doładowaniem i umieszczane w nowej partycji. Powstaje folder podzielony według dni z danymi źródłowymi.
Krok 2. Podczas inicjalizacyjnego ładowania ten folder jest odczytywany i analizowany przez Spark. Uzyskany dataframe jest zapisywany w formacie dostępnym do analizy, na przykład w Parquet, który można następnie zaimportować do Impali. W ten sposób powstaje docelowa witryna ze wszystkimi danymi, które do tego momentu się zgromadziły.
Krok 3. Tworzy się ładowanie, które będzie codziennie aktualizować witrynę.
Pojawia się pytanie o inkrementalne ładowanie, potrzebę partycjonowania witryny oraz kwestię wsparcia dla ogólnej schemy witryny.
Przykład. Załóżmy, że pierwszy krok budowy magazynu został zrealizowany i skonfigurowano eksport plików JSON do folderu.
Nie ma problemu w utworzeniu dataframe z nich, aby następnie zapisać jako witrynę. To pierwszy krok, który można łatwo znaleźć w dokumentacji Sparka:
df = spark.read.option("mergeSchema", True).json(".../*")
df.printSchema()
root
|-- a: long (nullable = true)
|-- b: string (nullable = true)
|-- c: struct (nullable = true) |
|-- d: long (nullable = true)Wydaje się, że wszystko jest w porządku.
Odczytaliśmy i sparsowaliśmy JSON, następnie zapisujemy dataframe jako Parquet, rejestrując go w Hive w dowolny wygodny sposób:
df.write.format(“parquet”).option('path','').saveAsTable('') Otrzymujemy witrynę.
Jednak następnego dnia do źródła dodano nowe dane. Mamy folder z JSON, oraz witrynę utworzoną na podstawie tego folderu. Po załadowaniu następnej partii danych z źródła, brakuje danych za jeden dzień w witrynie.
Logicznym rozwiązaniem będzie partycjonowanie witryny według dni, co pozwoli na dodawanie nowej partycji każdego dnia. Mechanizm ten jest dobrze znany, Spark pozwala na zapisywanie partycji oddzielnie.
Najpierw wykonujemy inicjalizacyjne ładowanie, zapisując dane tak, jak opisano powyżej, dodając tylko partycjonowanie. To działanie nazywa się inicjalizacją witryny i wykonywane jest tylko raz:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable)
Następnego dnia ładujemy tylko nową partycję:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable +"\/date_load=" + date_load + "\/" )
Pozostaje tylko ponownie zarejestrować w Hive, aby zaktualizować schemę.
Jednak tutaj pojawiają się problemy.
Pierwszy problem. Prędzej czy później utworzony plik parquet nie będzie możliwy do odczytania. Jest to związane z różnymi podejściami do pustych pól w parquet i JSON.
Rozważmy typową sytuację. Na przykład, wczoraj otrzymujemy JSON:
Dzień 1: {"a": {"b": 1}},
a dzisiaj ten sam JSON wygląda tak:
Dzień 2: {"a": null}
Załóżmy, że mamy dwie różne partycje, w każdej z nich po jednym wierszu.
Kiedy czytamy dane źródłowe w całości, Spark potrafi określić typ i zrozumie, że „a” jest polem typu „struktura”, z zagnieżdżonym polem „b” typu INT. Jednak, jeśli każda partycja była zapisana osobno, otrzymujemy parquet z niespójnymi schematami partycji:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Ta sytuacja jest dobrze znana, dlatego specjalnie dodano opcję — podczas parsowania danych źródłowych usunięcie pustych pól:
df = spark.read.json("...", dropFieldIfAllNull=True)
W takim przypadku parquet będzie składał się z partycji, które można będzie odczytać razem.
Chociaż ci, którzy mieli z tym do czynienia w praktyce, uśmiechną się gorzko. Dlaczego? Ponieważ najprawdopodobniej wystąpią jeszcze dwie sytuacje. Lub trzy. Lub cztery. Pierwsza, która wystąpi prawie na pewno, to różne sposoby przedstawiania typów numerycznych w różnych plikach JSON. Na przykład, {intField: 1} i {intField: 1.1}. Jeśli takie pola znajdą się w jednej partycji, to połączenie schematów odczyta wszystko prawidłowo, prowadząc do najbardziej dokładnego typu. A jeśli w różnych, to w jednej będzie intField: int, a w drugiej intField: double.
Aby obsłużyć tę sytuację, istnieje następująca flaga:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Teraz mamy folder, w którym znajdują się partycje, które można odczytać w jednolitym dataframe i poprawnym parquet całej witryny. Tak? Nie.
Trzeba pamiętać, że rejestrowaliśmy tabelę w Hive. Hive nie uwzględnia wielkości liter w nazwach pól, a parquet jest na to wrażliwy. Dlatego partycje ze schematami: field1: int, i Field1: int dla Hive są takie same, a dla Sparka już nie. Należy nie zapomnieć, aby przekształcić nazwy pól na małe litery.
Po tym wszystkim wydaje się, że wszystko jest w porządku.
Jednak nie jest to takie proste. Pojawia się druga, również dobrze znana problematyka. Ponieważ każda nowa partycja jest zapisywana osobno, w folderze partycji będą znajdować się pliki pomocnicze Sparka, na przykład flaga powodzenia operacji _SUCCESS. To spowoduje błąd podczas próby odczytu parquet. Aby tego uniknąć, należy skonfigurować ustawienia, zabraniając Sparkowi dopisywania plików pomocniczych do folderu:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Wygląda na to, że każdego dnia do folderu docelowego dodawana jest nowa partycja parquet, w której znajdują się przetworzone dane z danego dnia. Z góry zadbaliśmy o to, aby nie było partycji z konfliktem typów danych.
Jednak mamy trzeci problem. Teraz ogólny schemat nie jest znany, co więcej, w tabeli Hive jest niepoprawny schemat, ponieważ każda nowa partycja najprawdopodobniej wprowadziła zniekształcenie w schemacie.
Musimy ponownie zarejestrować tabelę. Można to zrobić w prosty sposób: ponownie odczytać partycje parquet, wziąć schemat i stworzyć na jego podstawie DDL, z którym zarejestrujemy folder w Hive jako tabelę zewnętrzną, aktualizując schemat folderu docelowego.
Przed nami czwarty problem. Gdy rejestrowaliśmy tabelę po raz pierwszy, opieraliśmy się na Sparku. Teraz robimy to sami i musimy pamiętać, że pola parquet mogą zaczynać się od znaków, które są niedozwolone dla Hive. Na przykład, Spark odrzuca wiersze, których nie mógł zparsować w polu „corrupt_record”. Takie pole nie będzie mogło być zarejestrowane w Hive bez wcześniejszego zescapowania.
Znając to, otrzymujemy schemat:
f_def = ""
for f in pf.dtypes:
if f[0] != "date_load":
f_def = f_def + "," + f[0].replace("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<")
table_define = "CREATE EXTERNAL TABLE jsonevolvtable (" + f_def[1:] + " ) "
table_define = table_define + "PARTITIONED BY (date_load string) STORED AS PARQUET LOCATION '\/user\/admin\/testJson\/testSchemaEvolution\/pq\/'"
hc.sql("drop table if exists jsonevolvtable")
hc.sql(table_define)
Kod ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") tworzy bezpieczny DDL, czyli zamiast:
create table tname (_field1 string, 1field string)
Z takimi nazwami pól jak „_field1, 1field”, tworzony jest bezpieczny DDL, gdzie nazwy pól są zescapowane: create table `tname` (`_field1` string, `1field` string).
Pojawia się pytanie: jak prawidłowo uzyskać dataframe z pełnym schematem (w kodzie pf)? Jak uzyskać ten pf? To piąty problem. Odczytywać schemat wszystkich partycji z folderu z plikami parquet docelowej witryny? To najbezpieczniejsza metoda, ale wymaga dużo zasobów.
Schemat jest już dostępny w Hive. Nowy schemat można uzyskać, łącząc schemat całej tabeli z nową partycją. Oznacza to, że należy wziąć schemat tabeli z Hive i połączyć go ze schematem nowej partycji. Można to zrobić, odczytując metadane testowe z Hive, zapisując je w tymczasowym folderze i odczytując za pomocą Sparka obie partycje jednocześnie.
W zasadzie mamy wszystko, czego potrzebujemy: oryginalny schemat tabeli w Hive i nową partycję. Mamy również dane. Pozostaje jedynie uzyskać nowy schemat, który łączy schemat witryny z nowymi polami z utworzonej partycji:
from pyspark.sql import HiveContext
from pyspark.sql.functions import lit
hc = HiveContext(spark)
df = spark.read.json("...", dropFieldIfAllNull=True)
df.write.mode("overwrite").parquet(".../date_load=12-12-2019")
pe = hc.sql("select * from jsonevolvtable limit 1")
pe.write.mode("overwrite").parquet(".../fakePartiton/")
pf = spark.read.option("mergeSchema", True).parquet(".../date_load=12-12-2019/*", ".../fakePartiton/*")
Następnie tworzymy DDL rejestracji tabeli, jak w poprzednim fragmencie.
Jeśli cały proces działa poprawnie, a mianowicie — była inicjalizująca ładowanie, a tabela jest poprawnie utworzona w Hive, to otrzymujemy zaktualizowany schemat tabeli.
Ostatni problem polega na tym, że nie można po prostu dodać partycji do tabeli Hive, ponieważ może to ją uszkodzić. Konieczne jest zmuszenie Hive do naprawienia struktury partycji:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Proste zadanie odczytu JSON i stworzenie na jego podstawie witryny przekształca się w pokonywanie szeregu niejawnych trudności, rozwiązania których trzeba szukać oddzielnie. I chociaż te rozwiązania są proste, to ich znalezienie zajmuje dużo czasu.
Aby zrealizować budowę witryny, trzeba było:
- Dodawać partycje do witryny, pozbywając się plików pomocniczych
- Rozwiązać problem z pustymi polami w danych wejściowych, które Spark typował
- Przekształcić proste typy na stringi
- Przekształcić nazwy pól na małe litery
- Oddzielić eksport danych i rejestrację tabeli w Hive (utworzenie DDL)
- Nie zapomnieć o wyciekaniu nazw pól, które mogą być niezgodne z Hive
- Nauczyć się aktualizować rejestrację tabeli w Hive
Podsumowując, należy zauważyć, że rozwiązanie budowy witryn skrywa wiele pułapek. Dlatego w przypadku wystąpienia trudności w realizacji lepiej zwrócić się do doświadczonego partnera z sukcesami w tej dziedzinie.
Dziękujemy za przeczytanie tego artykułu, mamy nadzieję, że informacje okażą się przydatne.
Źródło: habr.com
Kup solidny hosting stron z ochroną przed DDoS, serwery VPS VDS 🔥 Kup solidny hosting stron z ochroną przed DDoS, serwery VPS VDS | ProHoster
