Vážení čtenáři, přeji vám hezký den!
V tomto článku hlavní konzultant obchodního oddělení Big Data Solutions společnosti „Neoflex“ podrobně popisuje možnosti vytváření vitrín proměnné struktury s využitím Apache Spark.
V rámci projektu analýzy dat se často vyskytuje úkol vytváření vitrín na základě slabě strukturovaných dat.
Obvykle se jedná o logy nebo odpovědi různých systémů, které jsou uchovávány ve formátu JSON nebo XML. Data se exportují do Hadoopu, a poté je třeba vytvořit vitrínu. Přístup k vytvořené vitríně můžeme například organizovat prostřednictvím Impala.
V tomto případě je schéma cílové vitríny předem neznámé. Navíc schéma nemůže být sestaveno předem, protože závisí na datech a my se zabýváme těmito slabě strukturovanými daty.
Například, dnes se loguje taková odpověď:
{source: "app1", error_code: ""}a zítra od stejného systému přijde taková odpověď:
{source: "app1", error_code: "error", description: "Chyba sítě"}V důsledku toho by do vitríny mělo být přidáno další pole — description, a nikdo neví, zda přijde nebo ne.
Úkol vytvoření vitríny na takových datech je poměrně standardní a Spark pro to má řadu nástrojů. Pro analýzu zdrojových dat je podpora jak pro JSON, tak pro XML a pro neznámá schémata je k dispozici podpora schemaEvolution.
Na první pohled se řešení zdá být jednoduché. Stačí vzít složku s JSON a přečíst ji do dataframe. Spark vytvoří schéma a vnořené údaje převede na struktury. Poté je nutné vše uložit do parquet, který je podporován také v Impala, a zaregistrovat vitrínu v Hive metastore.
Vypadá to opravdu jednoduše.
Nicméně z krátkých příkladů v dokumentaci není jasné, co dělat s řadou problémů v praxi.
V dokumentaci je popsán přístup, který není zaměřen na vytváření vitrín, ale na čtení JSON nebo XML do dataframe.
A sice, je zde jednoduchý způsob, jak přečíst a analyzovat JSON:
df = spark.read.json(path...)To stačí k tomu, aby byly data dostupná pro Spark.
V praxi je scénář mnohem složitější, než jen přečíst JSON soubory ze složky a vytvořit dataframe. Situace je taková: už existuje určitá vitrína, každý den přicházejí nová data, která je třeba přidat do vitríny, přičemž je třeba si pamatovat, že schéma se může lišit.
Běžná schéma pro vytváření vitríny je následující:
Krok 1. Data jsou načítána do Hadoopu s následným denním doplňováním a ukládají se do nové partition. Vzniká tak složka rozdělená podle dnů se zdrojovými daty.
Krok 2. Během inicializačního načítání je tato složka čtena a zpracovávána pomocí Spark. Vytvořený dataframe je uložen ve formátu přístupném pro analýzu, například ve formátu parquet, který lze poté importovat do Impala. Tak se vytváří cílová datová vitrína se všemi daty, která se do této chvíle nashromáždila.
Krok 3. Vytvoří se načítání, které každý den aktualizuje vitrínu.
Nastává otázka inkrementálního načítání, potřeba partitionování vitríny a otázka podpory celkové schématu vitríny.
Uvedeme příklad. Předpokládejme, že byl realizován první krok budování úložiště a nastavena export JSON souborů do složky.
Vytvořit z nich dataframe, aby bylo možné jej později uložit jako vitrínu, není problém. To je ten první krok, který snadno najdete v dokumentaci Spark:
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)Vypadá to, že všechno je v pořádku.
Přečetli jsme a zpracovali JSON, poté ukládáme dataframe jako parquet, registrujeme v Hive jakýmkoli vhodným způsobem:
df.write.format(“parquet”).option('path','').saveAsTable('') Získáváme vitrínu.
Ale následující den přibyly nové údaje ze zdroje. Máme složku s JSON a vitrínu vytvořenou na základě této složky. Po načtení další dávky dat ze zdroje ve vitríně chybí údaje za jeden den.
Logickým řešením bude partitionovat vitrínu podle dnů, což umožní každý následující den přidat novou partition. Tento mechanismus je také dobře znám, Spark umožňuje zapisovat partitiony odděleně.
Nejprve provedeme inicializační načítání, ukládáme data, jak bylo popsáno výše, pouze s přidáním partitionování. Tento krok se nazývá inicializace vitríny a provádí se pouze jednou:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)
Následující den načteme pouze novou partition:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")
Zůstává pouze znovu registrovat v Hive, aby se aktualizoval schéma.
Nicméně, zde a vznikají problémy.
První problém. Dříve nebo později vytvořený parquet nebude možné přečíst. To souvisí s tím, jak se parquet a JSON liší v přístupu k prázdným polím.
Pojďme si přiblížit typickou situaci. Například, včera přišel JSON:
Den 1: {"a": {"b": 1}},
a dnes vypadá stejný JSON takto:
Den 2: {"a": null}
Představme si, že máme dvě oddělené partition, kde je jeden řádek.
Když čteme původní data jako celek, Spark dokáže určit typ a pochopí, že „a“ je pole typu „struktura“, s vnořeným polem „b“ typu INT. Ale pokud byla každá partition uložena zvlášť, vznikne parquet s neslučitelnými schématy partition:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Tato situace je dobře známá, proto byla speciálně přidána možnost — při parsování původních dat odstraňovat prázdná pole:
df = spark.read.json("...", dropFieldIfAllNull=True)
V tomto případě bude parquet tvořen partition, které budou možné přečíst společně.
Ačkoliv ti, kteří to dělali v praxi, se tu s hořkým úsměvem zasmějí. Proč? Protože pravděpodobně vzniknou další situace. Nebo tři. Nebo čtyři. První, která skoro určitě nastane, numerické typy budou v různých JSON souborech vypadat odlišně. Například, {intField: 1} a {intField: 1.1}. Pokud se tyto pole ocitnou v jedné partition, pak sloučení schématu všechno přečte správně a provede k nejpřesnějšímu typu. Ale pokud budou v různých, jedno bude mít intField: int, a druhé intField: double.
Pro zpracování této situace existuje následující příznak:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Teď máme složku, kde jsou partition, které lze přečíst do jednoho dataframe a validní parquet celé vitríny. Ano? Ne.
Je třeba si připomenout, že jsme registrovali tabulku v Hive. Hive není citlivý na velikost písmen v názvech polí, ale parquet je citlivý. Proto partition se schématy: field1: int, a Field1: int jsou pro Hive stejné, ale pro Spark nikoli. Je nutné nezapomenout převést názvy polí na malá písmena.
Po tomto se zdá, že všechno je v pořádku.
Nicméně, ne vše je tak jednoduché. Objevuje se druhý, také dobře známý problém. Protože každá nová partition se ukládá zvlášť, v složce partition budou ležet pomocné soubory Spark, například příznak úspěšnosti operace _SUCCESS. To povede k chybě při pokusu o parquet. Abychom tomu zabránili, je potřeba nastavit konfiguraci, která zakáže Spark zapisovat do složky pomocné soubory:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Vypadá to, že do cílového úložiště se každý den přidává nová paritce Parquet, kde jsou uložena analyzovaná data za den. Předem jsme se postarali o to, aby nedocházelo ke konfliktu typů dat v paritcích.
Ale máme zde třetí problém. Nyní není známa celková schéma a navíc je v tabulce Hive nesprávná schéma, protože každá nová paritce pravděpodobně vznesla zkreslení do schématu.
Je potřeba znovu zaregistrovat tabulku. To lze provést jednoduše: znovu přečíst Parquet úložiště, vzít schéma a vytvořit na jeho základě DDL, se kterým znovu zaregistrovat složku v Hive jako externí tabulku tím, že aktualizujeme schéma cílového úložiště.
Před námi je čtvrtý problém. Když jsme tabulku registrovali poprvé, opírali jsme se o Spark. Nyní to děláme sami a musíme mít na paměti, že pole Parquet mohou začínat znaky, které nejsou platné pro Hive. Například Spark vyhazuje řádky, které nemohl analyzovat do pole „corrupt_record“. Takové pole nebude možné zaregistrovat v Hive bez toho, abychom ho escapovali.
Vědíc to, získáváme schéma:
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)
Kód ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") vytváří bezpečný DDL, tedy namísto:
create table tname (_field1 string, 1field string)
S takovými názvy polí jako „_field1, 1field“ se vytváří bezpečný DDL, kde jsou názvy polí escapovány: create table `tname` (`_field1` string, `1field` string).
Objevuje se otázka: jak správně získat dataframe s plnou schématem (v kódu pf)? Jak získat tento pf? To je pátý problém. Přečíst schéma všech paritcí ze složky s Parquet soubory cílového úložiště? To je nejbezpečnější metoda, ale náročná.
Schéma již existuje v Hive. Novou schéma lze získat spojením schématu celé tabulky a nové partition. To znamená, že je třeba vzít schéma tabulky z Hive a sloučit ho s novým schématem partition. To lze udělat přečtením testových metadat z Hive, jejich uložením do dočasné složky a načtením obou partition najednou pomocí Sparku.
V podstatě máme vše, co potřebujeme: původní schéma tabulky v Hive a novou partition. Data máme také. Potřebujeme pouze získat nové schéma, ve kterém se spojuje schéma datového skladu a nová pole z vytvořené partition:
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/*")
Dále vytváříme DDL pro registraci tabulky, jak bylo uvedeno v předchozím fragmentu.
Pokud je celý řetězec správně funkční, to znamená — proběhla inicializační nahrávka a tabulka v Hive byla správně vytvořena, získáváme aktualizované schéma tabulky.
A poslední problém spočívá v tom, že do tabulky Hive nelze jednoduše přidat partition, protože by byla poškozena. Je třeba donutit Hive opravit svou strukturu partition:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Jednodužší úkol načítání JSON a vytvoření na jeho základě datového skladu vyústí v překonání celé řady nespecifikovaných obtíží, jejichž řešení je třeba hledat zvlášť. A i když jsou tato řešení jednoduchá, jejich hledání zabere hodně času.
Abychom realizovali konstrukci datového skladu, bylo třeba:
- Přidat partition do datového skladu, a zbavit se systémových souborů
- Vyřešit prázdná pole v původních datech, která byla typizována Sparkem
- Převést jednoduché typy na řetězec
- Převést názvy polí na malá písmena
- Oddělit export dat a registraci tabulky v Hive (vytvoření DDL)
- Nezapomenout escapovat názvy polí, které mohou být nekompatibilní s Hive
- Naučit se aktualizovat registraci tabulky v Hive
Závěrem bychom chtěli poznamenat, že řešení pro budování datových skladů skrývá mnoho nebezpečných překážek. Proto při obtížích s realizací je lepší obrátit se na zkušeného partnera s úspěšnou expertizou.
Děkujeme za přečtení tohoto článku, doufáme, že informace budou užitečné.
Zdroj: habr.com
Koupit spolehlivý hosting pro webové stránky s ochranou proti DDoS, VPS VDS servery 🔥 Koupit spolehlivý hosting pro webové stránky s ochranou proti DDoS, VPS VDS servery | ProHoster
