Cienījamie lasītāji, labdien!
Šajā rakstā vadošais konsultants Big Data Solutions biznesa jomā uzņēmumā «Neoflex» padziļināti apraksta, kā veidot mainīgas struktūras datu vitrīnas, izmantojot Apache Spark.
Datu analīzes projektā bieži rodas uzdevums veidot vitrīnas, pamatojoties uz vāji strukturētiem datiem.
Parasti tie ir žurnāli vai dažādu sistēmu atbildes, kuras tiek saglabātas JSON vai XML formātā. Dati tiek izsūtīti Hadoop, no tiem tālāk jāveido vitrīna. Pieejamību izveidotajai vitrīnai var organizēt, piemēram, izmantojot Impala.
Šajā gadījumā mērķa vitrīnas shēma iepriekš nav zināma. Vairāk nekā tas, shēma arī nevar tikt sastādīta iepriekš, jo tā ir atkarīga no datiem, un mums ir darīšana ar šiem vāji strukturētajiem datiem.
Piemēram, šodien tiek žurnēta šāda atbilde:
{source: "app1", error_code: ""}un rīt no šīs pašas sistēmas nāk šāda atbilde:
{source: "app1", error_code: "error", description: "Tīkla kļūda"}Rezultātā vitrīna būs jāpievieno vēl viena lauks — description, un vai tas tiks saņemts vai nē, neviens nezina.
Uzdevums veidot vitrīnu no šādiem datiem ir diezgan standarta, un Spark tam ir daudzi rīki. Izejdatņu datu parsēšanai tiek atbalstīti gan JSON, gan XML, un nezināmai iepriekš shēmai ir paredzēts schemaEvolution atbalsts.
No pirmā skatiena risinājums izskatās vienkārši. Jāņem mape ar JSON un jāizlasa dataframe. Spark izveidos shēmu, iegultos datus pārvēršot struktūrās. Pēc tam viss ir jāglabā parquet formātā, kas tiek atbalstīts arī Impala, reģistrējot vitrīnu Hive metastore.
Izskatās, ka viss ir vienkārši.
Tomēr īsajos piemēros dokumentācijā nav skaidrs, ko darīt ar dažām problēmām praksē.
Dokumentācijā tiek aprakstīts pieejas veids, kā nevis veidot vitrīnu, bet lasīt JSON vai XML dataframe.
Proti, vienkārši tiek parādīts, kā izlasīt un parsēt JSON:
df = spark.read.json(path...)Tas ir pietiekami, lai padarītu datus pieejamus Spark.
Praksē scenārijs ir daudz sarežģītāks nekā vienkārši izlasīt JSON failus no mapes un izveidot dataframe. Situācija izskatās tā: jau ir izveidota noteikta vitrīna, katru dienu nāk jauni dati, un tos ir jāpievieno vitrīnai, neaizmirstot, ka shēma var atšķirties.
Parasta vitrīnas veidošanas shēma ir šāda:
1. solis. Dati tiek ielādēti Hadoop, ar ikdienas turpmāko ielādi, un tiek glabāti jaunā partijā. Tā rezultātā tiek veidota dienu sadalītā mape ar sākotnējiem datiem.
2. solis. Ievades ielādes laikā šī mape tiek nolasīta un analizēta ar Spark palīdzību. Iegūtais dataframe tiek saglabāts formātā, kas ir pieejams analīzei, piemēram, parquet formātā, kuru pēc tam var importēt Impala. Tādējādi tiek izveidota mērķa vitrīna ar visiem datiem, kas ir uzkrājušies līdz šim.
3. solis. Tiek izveidota ielāde, kas katru dienu atjauninās vitrīnu.
Rodās jautājums par inkrementālo ielādi, nepieciešamību dalīt vitrīnu partijās un par kopējās vitrīnas shēmas atbalstu.
Piemēram, pieņemsim, ka ir veikts pirmais solis datu krātuves veidošanā un ir iestatīta JSON failu izgāde mapē.
Izveidot no tiem dataframe, lai vēlāk saglabātu to kā vitrīnu, nav problēma. Tas ir tieši tas pirmais solis, kuru viegli var atrast Spark dokumentācijā:
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)Izskatās, ka viss ir labi.
Mēs izlasījām un analizējām JSON, tālāk saglabājam dataframe kā parquet, reģistrējot Hive jebkurā ērtā veidā:
df.write.format(“parquet”).option('path','').saveAsTable('') Iegūstam vitrīnu.
Bet nākamajā dienā no avota pievienojās jauni dati. Mums ir mape ar JSON, un vitrīna, kas izveidota, pamatojoties uz šo mapē. Pēc nākamās datu partiijas ielādes vitrīnā trūkst datu par vienu dienu.
Loģisks risinājums būtu dalīt vitrīnu pa dienām, kas ļautu katru nākamo dienu pievienot jaunu partiju. Šis mehānisms arī ir labi zināms, Spark ļauj ierakstīt partijas atsevišķi.
Vispirms mēs veicam inicializējošo ielādi, saglabājot datus tā, kā aprakstīts iepriekš, pievienojot tikai dalīšanu pa partijām. Šī darbība tiek saukta par vitrīnas inicializāciju un tiek veikta tikai vienu reizi:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)
Nākamajā dienā ielādējam tikai jauno partiju:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")
Paliek tikai pārrakstīt Hive, lai atjauninātu shēmu.
Tomēr šeit arī rodas problēmas.
Pirmā problēma. Agrāk vai vēlāk iegūtais parquet nevarēs tikt nolasīts. Tas saistīts ar to, kā atšķirīgi parquet un JSON apstrādā tukšos laukus.
Apskatīsim tipisku situāciju. Piemēram, vakar saņemam JSON:
Diena 1: {"a": {"b": 1}},
bet šodien šis pats JSON izskatās tā:
Diena 2: {"a": null}
Pieņemsim, ka mums ir divas dažādas partitīcijas, katrā no kurām ir viena rinda.
Lasot sākotnējos datus kopumā, Spark spēs noteikt tipu un sapratīs, ka "a" ir lauks ar strukturētu tipu, ar iegulto lauku "b" tipo INT. Bet, ja katra partitīcija tika saglabāta atsevišķi, tad iznāk parquet ar nesaderīgām partitīciju shēmām:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Šī situācija ir labi zināma, tāpēc ir speciāli pievienota iespēja – parsējot sākotnējos datus, dzēst tukšos laukus:
df = spark.read.json("...", dropFieldIfAllNull=True)
Šajā gadījumā parquet sastāvēs no partitīcijām, kuras varēs nolasīt kopā.
Tomēr tie, kas to darījuši praksē, šeit grūti pasmaidīs. Kāpēc? Jo, visticamāk, radīsies vēl divas situācijas. Vai trīs. Vai četras. Pirmā, kas gandrīz noteikti radīsies, skaitliskie tipi dažādos JSON failos izskatīsies atšķirīgi. Piemēram, {intField: 1} un {intField: 1.1}. Ja šādi lauki iekļūst vienā partitīcijā, tad shēmas saplūšana nolasīs visu pareizi, pārvēršot uz visprecīzāko tipu. Bet, ja atsevišķi, tad vienā būs intField: int, bet otrā intField: double.
Lai apstrādātu šo situāciju, ir šāds karogs:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Tagad mums ir mape, kur atrodas partitīcijas, kuras var nolasīt vienotā dataframe un derīgā parquet visai izstādei. Jā? Nē.
Jāatceras, ka mēs reģistrējām tabulu Hive. Hive nav jutīgs pret lielo un mazo burtu lietojumu lauku nosaukumos, bet parquet ir jutīgs. Tāpēc partitīcijas ar shēmām: field1: int un Field1: int Hive ir vienādas, bet Spark nav. Jāatceras pārvērst lauku nosaukumus uz mazo burtu.
Pēc tam šķiet, ka viss ir labi.
Tomēr ne viss ir tik vienkārši. Radās otrā, arī labi zināmā problēma. Tā kā katra jaunā partitīcija tiek saglabāta atsevišķi, tad partitīcijas mapē atradīsies Spark servera faili, piemēram, operācijas veiksmīguma karogs _SUCCESS. Tas radīs kļūdu, mēģinot nolasīt parquet. Lai to novērstu, jākonfigurē iestatījums, aizliedzot Spark pievienot mapei servera failus:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Izskatās, ka tagad katru dienu mērķa vitrinā tiek pievienota jauna parquet partīcija, kurā atrodas dienā parsētie dati. Mēs iepriekš esam parūpējušies par to, lai nebūtu partīciju ar datu tipu konfliktiem.
Bet mums ir trešā problēma. Tagad kopējā shēma nav zināma, turklāt Hive tabula satur nepareizu shēmu, jo katra jauna partīcija, visticamāk, ir izraisījusi shēmas deformāciju.
Jāreģistrē tabula no jauna. To var izdarīt vienkārši: atkārtoti izlasīt parquet vitrinā, ņemt shēmu un izveidot uz tās pamata DDL, ar kuru no jauna reģistrēt mapīti Hive kā ārējo tabulu, atjauninot mērķa vitrinā esošo shēmu.
Mums rodas ceturtā problēma. Kad mēs pirmo reizi reģistrējām tabulu, mēs balstījāmies uz Spark. Tagad to darām paši, un jāatceras, ka parquet lauki var sākties ar simboliem, kuri nav atļauti Hive. Piemēram, Spark izmet rindas, kuras nevarēja parsēt laukā «corrupt_record». Šādu lauku nebūs iespējams reģistrēt Hive, nerakstot to ar bāru.
Zinot to, iegūstam shēmu:
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)
Kods ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") veido drošu DDL, proti, tā vietā,
create table tname (_field1 string, 1field string)
Ar tādiem lauku nosaukumiem kā «_field1, 1field», veido drošu DDL, kurā lauku nosaukumi ir noregulēti: create table `tname` (`_field1` string, `1field` string).
Radās jautājums: kā pareizi iegūt dataframe ar pilnu shēmu (kods pf)? Kā iegūt šo pf? Tā ir piektā problēma. Vai pārskatīt visu partīciju shēmu no mapītes ar parquet failiem mērķa vitrinā? Šī metode ir visdrošākā, bet arī smaga.
Shēma jau ir Hive. Jaunu shēmu var iegūt, apvienojot visas tabulas shēmu ar jauno partīciju. Tāpēc jāņem tabulas shēma no Hive un jāpievieno tā jaunās partīcijas shēmai. To var izdarīt, nolasot testa metadatus no Hive, saglabājot tos pagaidu mapē un nolasot, izmantojot Spark, abas partīcijas vienlaikus.
Pamatā ir viss, kas nepieciešams: sākotnējā tabulas shēma Hive un jaunā partīcija. Dati mums arī ir. Vēl paliek tikai iegūt jauno shēmu, kurā apvienojas vitrīnes shēma un jauni lauki no izveidotās partīcijas:
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/*")
Tālāk izveidojam DDL tabulas reģistrācijas, kā iepriekšējā fragmentā.
Ja visa ķēde darbojas pareizi, proti — ir bijusi inicializējoša ielāde, un Hive ir pareizi izveidota tabula, tad mēs saņemam atjauninātu tabulas shēmu.
Un pēdējā problēma ir, ka nav iespējams tik vienkārši pievienot partīciju Hive tabulai, jo tā tiks salauzta. Ir nepieciešams likt Hive salabot savu partīciju struktūru:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Vienkārša uzdevuma nolasīt JSON un izveidot vitrīnu uz tā pamata izrādās, ka jāiziet cauri virknei neizteiktām grūtībām, risinājumus kuriem jāmeklē atsevišķi. Un, lai gan šie risinājumi ir vienkārši, to meklēšanai aiziet daudz laika.
Lai īstenotu vitrīnas veidošanu, bija jā:
- Pievienot partīcijas vitrīnai, atbrīvojoties no palīgfailiem
- Izzināt tukšos laukus sākotnējajos datos, kurus Spark tipizēja
- Pārvērst vienkāršos tipus par virknēm
- Pārvērst lauku nosaukumus par maziem burtiem
- Apsargāt datu izsūtīšanu un tabulas reģistrāciju Hive (DDL izveide)
- Neaizmirst aizsargāt lauku nosaukumus, kuri var būt savstarpēji nesavienojami ar Hive
- Iemācīties atjaunināt tabulas reģistrāciju Hive
Kopsavilkumā jāatzīmē, ka risinājums vitrīnu veidošanā slēpj daudz zemūdens akmeņu. Tādēļ, ja rodas grūtības īstenošanā, labāk vērsties pie pieredzējuša partnera ar veiksmīgu ekspertīzi.
Paldies par šīs raksta izlasīšanu, ceram, ka informācija būs noderīga.
Avots: habr.com
Iegādājieties uzticamu tīmekļa hostingu ar DDoS aizsardzību, VPS VDS serveriem 🔥 Iegādājieties uzticamu tīmekļa hostingu ar DDoS aizsardzību, VPS VDS serveriem | ProHoster
