Spark schemaEvolution პრაქტიკაში

ძვირფასო მკითხველებო, კარგი დღე!

ამ სტატიაში Neoflex-ის Big Data Solutions ბიზნეს ზონის წამყვანი კონსულტანტი დეტალურად აღწერს Apache Spark-ის გამოყენებით ცვლადი სტრუქტურის ვიტრინების აგების ვარიანტებს.

როგორც მონაცემთა ანალიზის პროექტის ნაწილი, ხშირად ჩნდება მაღაზიის ვიტრინების აშენების ამოცანა, რომელიც ეფუძნება თავისუფლად სტრუქტურირებულ მონაცემებს.

ჩვეულებრივ, ეს არის ჟურნალები, ან პასუხები სხვადასხვა სისტემებიდან, შენახული როგორც JSON ან XML. მონაცემები აიტვირთება Hadoop-ში, შემდეგ თქვენ უნდა ააწყოთ მაღაზიის ვიტრინა მათგან. ჩვენ შეგვიძლია მოვაწყოთ წვდომა შექმნილ ვიტრინაზე, მაგალითად, Impala-ს მეშვეობით.

ამ შემთხვევაში, სამიზნე ვიტრინის სქემა წინასწარ არ არის ცნობილი. უფრო მეტიც, სქემის წინასწარ შედგენა ასევე შეუძლებელია, რადგან ეს დამოკიდებულია მონაცემებზე და საქმე გვაქვს ამ ძალიან თავისუფლად სტრუქტურირებულ მონაცემებთან.

მაგალითად, დღეს დარეგისტრირდა შემდეგი პასუხი:

{source: "app1", error_code: ""}

და ხვალ იგივე სისტემიდან მოდის შემდეგი პასუხი:

{source: "app1", error_code: "error", description: "Network error"}

შედეგად ვიტრინას უნდა დაემატოს კიდევ ერთი ველი - აღწერა და არავინ იცის მოვა თუ არა.

ასეთ მონაცემებზე ვიტრინის შექმნის ამოცანა საკმაოდ სტანდარტულია და Spark-ს ამისათვის არაერთი ინსტრუმენტი აქვს. წყაროს მონაცემების გასაანალიზებლად არის როგორც JSON, ასევე XML მხარდაჭერა, ხოლო ადრე უცნობი სქემისთვის უზრუნველყოფილია schemaEvolution-ის მხარდაჭერა.

ერთი შეხედვით, გამოსავალი მარტივი ჩანს. თქვენ უნდა აიღოთ საქაღალდე JSON-ით და წაიკითხოთ იგი მონაცემთა ჩარჩოში. Spark შექმნის სქემას, გადააქცევს წყობილ მონაცემებს სტრუქტურებად. გარდა ამისა, ყველაფერი უნდა შეინახოთ პარკეტში, რომელსაც ასევე მხარს უჭერს Impala, ვიტრინის დარეგისტრირებით Hive metastore-ში.

როგორც ჩანს, ყველაფერი მარტივია.

თუმცა, დოკუმენტაციაში მოყვანილი მოკლე მაგალითებიდან არ არის ნათელი, რა უნდა გააკეთოს პრაქტიკაში არსებულ მთელ რიგ პრობლემებთან.

დოკუმენტაცია აღწერს მიდგომას არა ვიტრინის შესაქმნელად, არამედ JSON ან XML მონაცემთა ჩარჩოში წაკითხვისთვის.

კერძოდ, ის უბრალოდ აჩვენებს, თუ როგორ უნდა წაიკითხოთ და გააანალიზოთ JSON:

df = spark.read.json(path...)

ეს საკმარისია იმისთვის, რომ მონაცემები ხელმისაწვდომი გახდეს Spark-ისთვის.

პრაქტიკაში, სკრიპტი ბევრად უფრო რთულია, ვიდრე უბრალოდ საქაღალდედან JSON ფაილების წაკითხვა და მონაცემთა ჩარჩოს შექმნა. სიტუაცია ასე გამოიყურება: უკვე არის გარკვეული ვიტრინა, ყოველდღე შემოდის ახალი მონაცემები, ისინი უნდა დაემატოს ვიტრინას, არ უნდა დაგვავიწყდეს, რომ სქემა შეიძლება განსხვავდებოდეს.

ვიტრინის აშენების ჩვეულებრივი სქემა შემდეგია:

ნაბიჯი 1. მონაცემები იტვირთება Hadoop-ში შემდგომი ყოველდღიური გადატვირთვით და ემატება ახალ დანაყოფს. გამოდის საქაღალდე თავდაპირველი მონაცემებით დაყოფილი დღის მიხედვით.

ნაბიჯი 2. საწყისი ჩატვირთვის დროს ეს საქაღალდე იკითხება და ანალიზდება Spark-ის მიერ. შედეგად მიღებული მონაცემთა ჩარჩო ინახება გასაანალიზებელ ფორმატში, მაგალითად, პარკეტში, რომელიც შემდეგ შეიძლება იმპორტირებული იყოს Impala-ში. ეს ქმნის სამიზნე გამოფენას ყველა მონაცემით, რომელიც დაგროვდა ამ მომენტამდე.

ნაბიჯი 3. იქმნება ჩამოტვირთვა, რომელიც განაახლებს მაღაზიის ვიტრინას ყოველდღე.
ჩნდება დამატებითი დატვირთვის საკითხი, ვიტრინის დაყოფის აუცილებლობა და ვიტრინის ზოგადი სქემის შენარჩუნების საკითხი.

ავიღოთ მაგალითი. დავუშვათ, რომ საცავის აშენების პირველი ნაბიჯი განხორციელდა და JSON ფაილები აიტვირთება საქაღალდეში.

მათგან მონაცემთა ჩარჩოს შექმნა, შემდეგ მისი ვიტრინის სახით შენახვა პრობლემა არ არის. ეს არის პირველი ნაბიჯი, რომელიც ადვილად შეგიძლიათ ნახოთ 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)

ყველაფერი კარგად ჩანს.

ჩვენ ვკითხულობთ და გავაანალიზებთ JSON-ს, შემდეგ ვინახავთ მონაცემთა ჩარჩოს პარკეტად და ვარეგისტრირებთ მას Hive-ში ნებისმიერი მოსახერხებელი გზით:

df.write.format(“parquet”).option('path','<External Table Path>').saveAsTable('<Table Name>')

ჩვენ ვიღებთ ფანჯარას.

მაგრამ, მეორე დღეს, წყაროდან ახალი მონაცემები დაემატა. ჩვენ გვაქვს საქაღალდე JSON-ით და ამ საქაღალდიდან შექმნილი ვიტრინა. წყაროდან მონაცემების შემდეგი ჯგუფის ჩატვირთვის შემდეგ, მონაცემთა მარტს აკლია ერთი დღის მონაცემები.

ლოგიკური გამოსავალი იქნება ვიტრინის დაყოფა დღის განმავლობაში, რაც საშუალებას მოგცემთ დაამატოთ ახალი დანაყოფი ყოველ მეორე დღეს. ამის მექანიზმი ასევე ცნობილია, Spark საშუალებას გაძლევთ დაწეროთ ტიხრები ცალკე.

პირველ რიგში, ჩვენ ვაკეთებთ საწყის დატვირთვას, ვინახავთ მონაცემებს, როგორც ზემოთ იყო აღწერილი, ვამატებთ მხოლოდ დაყოფას. ამ მოქმედებას ეწოდება მაღაზიის ინიციალიზაცია და კეთდება მხოლოდ ერთხელ:

df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)

მეორე დღეს ჩვენ ვტვირთავთ მხოლოდ ახალ დანაყოფის:

df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")

რჩება მხოლოდ ხელახლა დარეგისტრირება Hive-ში სქემის განახლებისთვის.
თუმცა, სწორედ აქ ჩნდება პრობლემები.

პირველი პრობლემა. ადრე თუ გვიან, მიღებული პარკეტი წაუკითხავი იქნება. ეს იმის გამო ხდება, რომ პარკეტი და JSON განსხვავებულად ეპყრობიან ცარიელ ველებს.

განვიხილოთ ტიპიური სიტუაცია. მაგალითად, გუშინ JSON ჩამოდის:

День 1: {"a": {"b": 1}},

და დღეს იგივე JSON ასე გამოიყურება:

День 2: {"a": null}

ვთქვათ, გვაქვს ორი განსხვავებული დანაყოფი, თითოეულს აქვს ერთი ხაზი.
როდესაც ჩვენ წავიკითხავთ მთლიან წყაროს მონაცემებს, Spark შეძლებს განსაზღვროს ტიპი და მიხვდება, რომ "a" არის "სტრუქტურის" ტიპის ველი, რომელსაც აქვს ჩასმული ველი "b" ტიპის INT. მაგრამ, თუ თითოეული დანაყოფი ცალკე იყო შენახული, მაშინ ვიღებთ პარკეტს შეუთავსებელი დანაყოფის სქემებით:

df1 (a: <struct<"b": INT>>)
df2 (a: STRING NULLABLE)

ეს სიტუაცია კარგად არის ცნობილი, ამიტომ სპეციალურად დაემატა ვარიანტი - წყაროს მონაცემების გაანალიზებისას ამოიღეთ ცარიელი ველები:

df = spark.read.json("...", dropFieldIfAllNull=True)

ამ შემთხვევაში, პარკეტი შედგება ტიხრებისგან, რომელთა წაკითხვა შესაძლებელია ერთად.
თუმცა, ვინც ამას პრაქტიკაში აკეთებდა, აქ მწარედ გაიღიმება. რატომ? დიახ, რადგან, სავარაუდოდ, კიდევ ორი ​​სიტუაცია იქნება. ან სამი. ან ოთხი. პირველი, რომელიც თითქმის აუცილებლად მოხდება, არის ის, რომ რიცხვითი ტიპები განსხვავებულად გამოიყურება სხვადასხვა JSON ფაილში. მაგალითად, {intField: 1} და {intField: 1.1}. თუ ასეთი ველები გვხვდება ერთ დანაყოფში, მაშინ სქემის შერწყმა წაიკითხავს ყველაფერს სწორად, რაც გამოიწვევს ყველაზე ზუსტ ტიპს. მაგრამ თუ სხვადასხვაში, მაშინ ერთს ექნება intField: int, ხოლო მეორეს ექნება intField: double.

ამ სიტუაციის მოსაგვარებლად არის შემდეგი დროშა:

df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)

ახლა ჩვენ გვაქვს საქაღალდე, სადაც არის დანაყოფები, რომელთა წაკითხვა შესაძლებელია ერთ მონაცემთა ჩარჩოში და მთელი ვიტრინის მოქმედ პარკეტში. დიახ? არა.

უნდა გვახსოვდეს, რომ ცხრილი დავარეგისტრირეთ Hive-ში. Hive არ არის რეგისტრირებული ველების სახელებში, ხოლო პარკეტი არის რეგისტრირებული. ამიტომ, დანაყოფები სქემებით: field1: int და Field1: int იგივეა Hive-სთვის, მაგრამ არა Spark-ისთვის. არ დაგავიწყდეთ ველების სახელების მცირე რეზერვში გადაყვანა.

ამის შემდეგ თითქოს ყველაფერი კარგადაა.

თუმცა, ყველაფერი ასე მარტივი არ არის. არის მეორე, ასევე ცნობილი პრობლემა. ვინაიდან ყოველი ახალი დანაყოფი ცალკე ინახება, დანაყოფის საქაღალდე შეიცავს Spark სერვისის ფაილებს, მაგალითად, _SUCCESS ოპერაციის წარმატების დროშას. ეს გამოიწვევს შეცდომას პარკეტის მცდელობისას. ამის თავიდან ასაცილებლად, თქვენ უნდა დააკონფიგურიროთ კონფიგურაცია, რათა თავიდან აიცილოთ Spark-ს საქაღალდეში სერვისის ფაილების დამატება:

hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")

როგორც ჩანს, ახლა ყოველდღე ახალი პარკეტის დანაყოფი ემატება სამიზნე გამოფენის საქაღალდეს, სადაც განთავსებულია დღის გაანალიზებული მონაცემები. ჩვენ წინასწარ ვიზრუნეთ, რომ არ არსებობდეს დანაყოფები მონაცემთა ტიპის კონფლიქტით.

მაგრამ მესამე პრობლემა გვაქვს. ახლა ზოგადი სქემა არ არის ცნობილი, უფრო მეტიც, Hive-ის ცხრილს აქვს არასწორი სქემა, რადგან ყოველი ახალი დანაყოფი, სავარაუდოდ, სქემაში დამახინჯებას აყენებს.

ცხრილი ხელახლა უნდა დაარეგისტრიროთ. ეს შეიძლება გაკეთდეს მარტივად: ხელახლა წაიკითხეთ ვიტრინის პარკეტი, აიღეთ სქემა და შექმენით მასზე დაყრდნობით DDL, რომლითაც ხელახლა დაარეგისტრირეთ საქაღალდე Hive-ში, როგორც გარე ცხრილი, განაახლეთ სამიზნე ვიტრინის სქემა.

მეოთხე პრობლემა გვაქვს. როდესაც ცხრილი პირველად დავარეგისტრირეთ, სპარკს დავეყრდენით. ახლა ჩვენ თვითონ ვაკეთებთ ამას და უნდა გვახსოვდეს, რომ პარკეტის ველები შეიძლება დაიწყოს სიმბოლოებით, რომლებიც დაუშვებელია Hive-სთვის. მაგალითად, Spark აგდებს ხაზებს, რომლებსაც ვერ აანალიზებს "corrupt_record" ველში. ასეთი ველი არ შეიძლება დარეგისტრირდეს Hive-ში გაქცევის გარეშე.

ამის ცოდნა, ჩვენ ვიღებთ სქემას:

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)

კოდი ("_corrupt_record", "`_corrupt_record") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("მასივი<`", "მასივი<") ხდის უსაფრთხო DDL-ს, ანუ ნაცვლად:

create table tname (_field1 string, 1field string)

ველების სახელებით, როგორიცაა "_field1, 1field", იქმნება უსაფრთხო DDL, სადაც ველების სახელები არ არის: შექმენით ცხრილი `tname` (`_field1` string, `1field` string).

ჩნდება კითხვა: როგორ სწორად მივიღოთ მონაცემთა ჩარჩო სრული სქემით (pf კოდით)? როგორ მივიღოთ ეს pf? ეს მეხუთე პრობლემაა. ხელახლა წაიკითხეთ ყველა დანაყოფის სქემა საქაღალდედან სამიზნე ვიტრინის პარკეტის ფაილებით? ეს მეთოდი ყველაზე უსაფრთხო, მაგრამ რთულია.

სქემა უკვე Hive-შია. თქვენ შეგიძლიათ მიიღოთ ახალი სქემა მთელი ცხრილისა და ახალი დანაყოფის სქემის კომბინაციით. ასე რომ, თქვენ უნდა აიღოთ ცხრილის სქემა Hive-დან და დააკავშიროთ იგი ახალი დანაყოფის სქემასთან. ეს შეიძლება გაკეთდეს Hive-დან ტესტის მეტამონაცემების წაკითხვით, დროებით საქაღალდეში შენახვით და ორივე დანაყოფის ერთდროულად წასაკითხად Spark-ის გამოყენებით.

სინამდვილეში, არის ყველაფერი, რაც გჭირდებათ: ცხრილის ორიგინალური სქემა Hive-ში და ახალი დანაყოფი. ჩვენც გვაქვს მონაცემები. რჩება მხოლოდ ახალი სქემის მიღება, რომელიც აერთიანებს მაღაზიის ვიტრინის სქემას და ახალ ველებს შექმნილი დანაყოფიდან:

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/*")

შემდეგი, ჩვენ ვქმნით ცხრილის რეგისტრაციის DDL-ს, როგორც წინა ფრაგმენტში.
თუ მთელი ჯაჭვი მუშაობს სწორად, კერძოდ, იყო ინიციალიზაციის დატვირთვა და ცხრილი სწორად შეიქმნა Hive-ში, მაშინ მივიღებთ განახლებულ ცხრილის სქემას.

და ბოლო პრობლემა ის არის, რომ თქვენ არ შეგიძლიათ უბრალოდ დაამატოთ დანაყოფი Hive მაგიდაზე, რადგან ის გაფუჭდება. თქვენ უნდა აიძულოთ Hive შეასწოროს მისი დანაყოფის სტრუქტურა:

from pyspark.sql import HiveContext
hc = HiveContext(spark) 
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)

JSON-ის წაკითხვისა და მასზე დაფუძნებული ვიტრინის შექმნის მარტივი ამოცანა იწვევს არაერთი იმპლიციტური სირთულეების გადალახვას, რომელთა გადაწყვეტილებებიც ცალკე უნდა მოძებნოთ. და მიუხედავად იმისა, რომ ეს გადაწყვეტილებები მარტივია, მათ პოვნას დიდი დრო სჭირდება.

ვიტრინის მშენებლობის განსახორციელებლად მომიწია:

• დაამატეთ ტიხრები ვიტრინაში, მოიცილეთ სერვისის ფაილები
• გაუმკლავდეთ ცარიელ ველებს წყაროს მონაცემებში, რომლებიც Spark-მა აკრიფა
• გადაიტანეთ მარტივი ტიპები სტრინგზე
• გადაიყვანეთ ველების სახელები პატარა რეზერვებში
• მონაცემთა ცალკე ატვირთვა და ცხრილის რეგისტრაცია Hive-ში (DDL თაობა)
• არ დაგავიწყდეთ ველების სახელების გაქცევა, რომლებიც შეიძლება შეუთავსებელი იყოს Hive-თან
• ისწავლეთ როგორ განაახლოთ ცხრილის რეგისტრაცია Hive-ში

შეჯამებით, ჩვენ აღვნიშნავთ, რომ მაღაზიის ვიტრინების აშენების გადაწყვეტილება სავსეა მრავალი ხარვეზით. ამიტომ, განხორციელების სირთულეების შემთხვევაში, სჯობს დაუკავშირდით გამოცდილ პარტნიორს წარმატებული გამოცდილებით.

გმადლობთ ამ სტატიის წაკითხვისთვის, ვიმედოვნებთ, რომ ინფორმაცია თქვენთვის სასარგებლო იქნება.

წყარო: www.habr.com