Spark schemaEvolution ໃນການປະຕິບັດ

ຜູ້ອ່ານທີ່ຮັກແພງ, ມື້ທີ່ດີ!

ໃນບົດຄວາມນີ້, ທີ່ປຶກສາຊັ້ນນໍາຂອງພື້ນທີ່ທຸລະກິດ Big Data Solutions ຂອງ Neoflex ອະທິບາຍລາຍລະອຽດກ່ຽວກັບທາງເລືອກໃນການກໍ່ສ້າງໂຄງສ້າງທີ່ປ່ຽນແປງໄດ້ໂດຍໃຊ້ Apache Spark.

ເປັນສ່ວນຫນຶ່ງຂອງໂຄງການການວິເຄາະຂໍ້ມູນ, ວຽກງານຂອງການສ້າງຫນ້າຮ້ານໂດຍອີງໃສ່ຂໍ້ມູນທີ່ມີໂຄງສ້າງວ່າງມັກຈະເກີດຂື້ນ.

ປົກກະຕິແລ້ວເຫຼົ່ານີ້ແມ່ນບັນທຶກ, ຫຼືຄໍາຕອບຈາກລະບົບຕ່າງໆ, ບັນທຶກເປັນ JSON ຫຼື XML. ຂໍ້ມູນຖືກອັບໂຫຼດໃສ່ Hadoop, ຫຼັງຈາກນັ້ນທ່ານຈໍາເປັນຕ້ອງສ້າງຫນ້າຮ້ານຈາກພວກເຂົາ. ພວກເຮົາສາມາດຈັດການເຂົ້າເຖິງງານວາງສະແດງທີ່ສ້າງຂຶ້ນ, ຕົວຢ່າງ, ຜ່ານ Impala.

ໃນກໍລະນີນີ້, schema ຂອງຫນ້າຮ້ານເປົ້າຫມາຍແມ່ນບໍ່ຮູ້ຈັກລ່ວງຫນ້າ. ຍິ່ງໄປກວ່ານັ້ນ, ໂຄງການດັ່ງກ່າວຍັງບໍ່ສາມາດຖືກແຕ້ມຂຶ້ນລ່ວງຫນ້າ, ເນື່ອງຈາກວ່າມັນຂຶ້ນກັບຂໍ້ມູນ, ແລະພວກເຮົາກໍາລັງຈັດການກັບຂໍ້ມູນທີ່ມີໂຄງສ້າງທີ່ວ່າງຫຼາຍ.

ຕົວຢ່າງ, ໃນມື້ນີ້ຄໍາຕອບຕໍ່ໄປນີ້ຖືກບັນທຶກ:

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

ແລະມື້ອື່ນຈາກລະບົບດຽວກັນມາຄໍາຕອບຕໍ່ໄປນີ້:

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

ດັ່ງນັ້ນ, ຫນຶ່ງພາກສະຫນາມເພີ່ມເຕີມຄວນໄດ້ຮັບການເພີ່ມໃສ່ໃນ showcase - ຄໍາອະທິບາຍ, ແລະບໍ່ມີໃຜຮູ້ວ່າມັນຈະມາຫຼືບໍ່.

ວຽກງານຂອງການສ້າງຫນ້າຮ້ານໃນຂໍ້ມູນດັ່ງກ່າວແມ່ນມາດຕະຖານທີ່ສວຍງາມ, ແລະ Spark ມີເຄື່ອງມືຈໍານວນຫນຶ່ງສໍາລັບການນີ້. ສໍາລັບການແຍກວິເຄາະແຫຼ່ງຂໍ້ມູນ, ມີການສະຫນັບສະຫນູນທັງ JSON ແລະ XML, ແລະສໍາລັບ schema ທີ່ບໍ່ຮູ້ຈັກກ່ອນຫນ້ານີ້, ການສະຫນັບສະຫນູນສໍາລັບ schemaEvolution ແມ່ນສະຫນອງໃຫ້.

ຢູ່ glance ທໍາອິດ, ການແກ້ໄຂເບິ່ງຄືວ່າງ່າຍດາຍ. ທ່ານຈໍາເປັນຕ້ອງເອົາໂຟນເດີທີ່ມີ JSON ແລະອ່ານມັນເຂົ້າໄປໃນກອບຂໍ້ມູນ. Spark ຈະສ້າງ schema, ປ່ຽນຂໍ້ມູນ nested ເປັນໂຄງສ້າງ. ນອກຈາກນັ້ນ, ທຸກສິ່ງທຸກຢ່າງຕ້ອງໄດ້ຮັບການບັນທຶກໄວ້ໃນ parquet, ເຊິ່ງຍັງໄດ້ຮັບການສະຫນັບສະຫນູນໃນ Impala, ໂດຍການລົງທະບຽນຫນ້າຮ້ານໃນ Hive metastore.

ທຸກສິ່ງທຸກຢ່າງເບິ່ງຄືວ່າງ່າຍດາຍ.

ຢ່າງໃດກໍ່ຕາມ, ມັນບໍ່ແມ່ນຄວາມຊັດເຈນຈາກຕົວຢ່າງສັ້ນໆໃນເອກະສານສິ່ງທີ່ຕ້ອງເຮັດກັບບັນຫາຈໍານວນຫນຶ່ງໃນການປະຕິບັດ.

ເອກະສານອະທິບາຍວິທີການບໍ່ສ້າງຫນ້າຮ້ານ, ແຕ່ອ່ານ JSON ຫຼື XML ເຂົ້າໄປໃນກອບຂໍ້ມູນ.

ຄື, ມັນພຽງແຕ່ສະແດງໃຫ້ເຫັນວິທີການອ່ານແລະວິເຄາະ JSON:

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

ນີ້ແມ່ນພຽງພໍທີ່ຈະເຮັດໃຫ້ຂໍ້ມູນທີ່ມີໃຫ້ກັບ Spark.

ໃນທາງປະຕິບັດ, script ແມ່ນສັບສົນຫຼາຍກ່ວາພຽງແຕ່ອ່ານໄຟລ໌ JSON ຈາກໂຟນເດີແລະສ້າງກອບຂໍ້ມູນ. ສະຖານະການເບິ່ງຄືວ່ານີ້: ມີຫນ້າຮ້ານທີ່ແນ່ນອນແລ້ວ, ຂໍ້ມູນໃຫມ່ເຂົ້າມາທຸກໆມື້, ພວກເຂົາຕ້ອງໄດ້ຮັບການເພີ່ມໃສ່ຫນ້າຮ້ານ, ຢ່າລືມວ່າໂຄງການອາດຈະແຕກຕ່າງກັນ.

ໂຄງ​ການ​ປົກ​ກະ​ຕິ​ສໍາ​ລັບ​ການ​ກໍ່​ສ້າງ showcase ເປັນ​ດັ່ງ​ຕໍ່​ໄປ​ນີ້​:

ຂັ້ນຕອນ 1. ຂໍ້ມູນໄດ້ຖືກໂຫລດເຂົ້າໄປໃນ Hadoop ດ້ວຍການໂຫຼດໃຫມ່ປະຈໍາວັນຕໍ່ມາແລະເພີ່ມເຂົ້າໄປໃນພາທິຊັນໃຫມ່. ມັນ turns ອອກໂຟນເດີທີ່ມີຂໍ້ມູນເບື້ອງຕົ້ນແບ່ງສ່ວນໂດຍມື້.

ຂັ້ນຕອນ 2. ໃນລະຫວ່າງການໂຫຼດເບື້ອງຕົ້ນ, ໂຟນເດີນີ້ຈະຖືກອ່ານ ແລະວິເຄາະໂດຍ Spark. ກອບຂໍ້ມູນຜົນໄດ້ຮັບຖືກບັນທຶກໄວ້ໃນຮູບແບບທີ່ສາມາດວິເຄາະໄດ້, ສໍາລັບການຍົກຕົວຢ່າງ, ໃນ parquet, ເຊິ່ງຫຼັງຈາກນັ້ນສາມາດຖືກນໍາເຂົ້າເຂົ້າໄປໃນ Impala. ນີ້ສ້າງການສະແດງເປົ້າຫມາຍທີ່ມີຂໍ້ມູນທັງຫມົດທີ່ໄດ້ສະສົມເຖິງຈຸດນີ້.

ຂັ້ນຕອນ 3. ການດາວໂຫຼດແມ່ນຖືກສ້າງຂຶ້ນເພື່ອອັບເດດໜ້າຮ້ານທຸກໆມື້.
ມີຄໍາຖາມກ່ຽວກັບການເພີ່ມການໂຫຼດ, ຄວາມຕ້ອງການທີ່ຈະແບ່ງສ່ວນຂອງ showcase, ແລະຄໍາຖາມຂອງການຮັກສາລະບົບທົ່ວໄປຂອງ showcase.

ໃຫ້​ເຮົາ​ໃຊ້​ຕົວ​ຢ່າງ. ໃຫ້ເວົ້າວ່າຂັ້ນຕອນທໍາອິດຂອງການສ້າງ repository ໄດ້ຖືກປະຕິບັດ, ແລະໄຟລ໌ 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, ຫຼັງຈາກນັ້ນພວກເຮົາບັນທຶກ dataframe ເປັນ parquet, ລົງທະບຽນມັນຢູ່ໃນ Hive ໃນວິທີທີ່ສະດວກ:

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

ພວກເຮົາໄດ້ຮັບປ່ອງຢ້ຽມ.

ແຕ່, ໃນມື້ຕໍ່ມາ, ຂໍ້ມູນໃຫມ່ຈາກແຫຼ່ງໄດ້ຖືກເພີ່ມ. ພວກເຮົາມີໂຟນເດີທີ່ມີ JSON, ແລະຕູ້ວາງສະແດງທີ່ສ້າງຂຶ້ນຈາກໂຟນເດີນີ້. ຫຼັງ​ຈາກ​ການ​ໂຫຼດ​ຊຸດ​ຕໍ່​ໄປ​ຂອງ​ຂໍ້​ມູນ​ຈາກ​ແຫຼ່ງ​ຂໍ້​ມູນ​, data mart ແມ່ນ​ຂາດ​ຂໍ້​ມູນ​ທີ່​ມີ​ມູນ​ຄ່າ​ຂອງ​ມື້​ຫນຶ່ງ​.

ການແກ້ໄຂຢ່າງມີເຫດຜົນຈະເປັນການແບ່ງສ່ວນໜ້າຮ້ານໃນແຕ່ລະມື້, ເຊິ່ງຈະຊ່ວຍໃຫ້ເພີ່ມການແບ່ງສ່ວນໃໝ່ໄດ້ທຸກໆມື້. ກົນໄກສໍາລັບການນີ້ຍັງເປັນທີ່ຮູ້ຈັກດີ, Spark ອະນຸຍາດໃຫ້ທ່ານຂຽນພາທິຊັນແຍກຕ່າງຫາກ.

ກ່ອນອື່ນ, ພວກເຮົາເຮັດການໂຫຼດເບື້ອງຕົ້ນ, ບັນທຶກຂໍ້ມູນດັ່ງທີ່ອະທິບາຍຂ້າງເທິງ, ເພີ່ມພຽງແຕ່ການແບ່ງປັນ. ການກະທຳນີ້ເອີ້ນວ່າການເລີ່ມຕົ້ນໜ້າຮ້ານ ແລະເຮັດພຽງຄັ້ງດຽວເທົ່ານັ້ນ:

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

ໃນມື້ຕໍ່ມາ, ພວກເຮົາໂຫຼດພຽງແຕ່ partition ໃຫມ່:

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

ທັງຫມົດທີ່ຍັງເຫຼືອແມ່ນເພື່ອລົງທະບຽນໃຫມ່ໃນ Hive ເພື່ອປັບປຸງ schema.
ຢ່າງໃດກໍຕາມ, ນີ້ແມ່ນບ່ອນທີ່ບັນຫາເກີດຂຶ້ນ.

ບັນຫາທໍາອິດ. ບໍ່ດົນຫຼືຫຼັງຈາກນັ້ນ, parquet ຜົນໄດ້ຮັບຈະບໍ່ສາມາດອ່ານໄດ້. ນີ້ແມ່ນຍ້ອນວິທີການ parquet ແລະ JSON ປະຕິບັດກັບພື້ນທີ່ຫວ່າງເປົ່າແຕກຕ່າງກັນ.

ໃຫ້ພິຈາລະນາສະຖານະການປົກກະຕິ. ຕົວຢ່າງ, ມື້ວານນີ້ JSON ມາຮອດ:

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

ແລະໃນມື້ນີ້ JSON ດຽວກັນເບິ່ງຄືວ່ານີ້:

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

ໃຫ້ເວົ້າວ່າພວກເຮົາມີສອງພາທິຊັນທີ່ແຕກຕ່າງກັນ, ແຕ່ລະຄົນມີເສັ້ນດຽວ.
ເມື່ອພວກເຮົາອ່ານຂໍ້ມູນແຫຼ່ງທັງຫມົດ, Spark ຈະສາມາດກໍານົດປະເພດ, ແລະຈະເຂົ້າໃຈວ່າ "a" ແມ່ນພາກສະຫນາມຂອງປະເພດ "ໂຄງສ້າງ", ໂດຍມີຊ່ອງຂໍ້ມູນ "b" ທີ່ຊ້ອນກັນຂອງປະເພດ INT. ແຕ່, ຖ້າແຕ່ລະການແບ່ງປັນຖືກບັນທຶກໄວ້ແຍກຕ່າງຫາກ, ພວກເຮົາໄດ້ຮັບ parquet ທີ່ມີຮູບແບບການແບ່ງປັນທີ່ບໍ່ເຂົ້າກັນໄດ້:

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

ສະຖານະການນີ້ແມ່ນເປັນທີ່ຮູ້ຈັກດີ, ດັ່ງນັ້ນທາງເລືອກໄດ້ຖືກເພີ່ມເປັນພິເສດ - ເມື່ອວິເຄາະຂໍ້ມູນແຫຼ່ງ, ເອົາຊ່ອງຫວ່າງເປົ່າອອກ:

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

ໃນກໍລະນີນີ້, parquet ຈະປະກອບດ້ວຍການແບ່ງສ່ວນທີ່ສາມາດອ່ານຮ່ວມກັນ.
ເຖິງແມ່ນວ່າຜູ້ທີ່ໄດ້ເຮັດສິ່ງນີ້ໃນການປະຕິບັດຈະຍິ້ມຢ່າງຂົມຂື່ນຢູ່ທີ່ນີ້. ເປັນຫຍັງ? ແມ່ນແລ້ວ, ເນື່ອງຈາກວ່າມີແນວໂນ້ມທີ່ຈະມີສອງສະຖານະການເພີ່ມເຕີມ. ຫຼືສາມ. ຫຼືສີ່. ທໍາອິດ, ເຊິ່ງເກືອບແນ່ນອນຈະເກີດຂຶ້ນ, ແມ່ນວ່າປະເພດຕົວເລກຈະມີລັກສະນະແຕກຕ່າງກັນໃນໄຟລ໌ JSON ທີ່ແຕກຕ່າງກັນ. ຕົວຢ່າງ, {intField: 1} ແລະ {intField: 1.1}. ຖ້າຊ່ອງຂໍ້ມູນດັ່ງກ່າວຖືກພົບເຫັນຢູ່ໃນການແບ່ງປັນຫນຶ່ງ, ຫຼັງຈາກນັ້ນ schema merge ຈະອ່ານທຸກຢ່າງຢ່າງຖືກຕ້ອງ, ນໍາໄປສູ່ປະເພດທີ່ຖືກຕ້ອງທີ່ສຸດ. ແຕ່ຖ້າຢູ່ໃນຕົວທີ່ແຕກຕ່າງກັນ, ຫຼັງຈາກນັ້ນຫນຶ່ງຈະມີ intField: int, ແລະອີກອັນຫນຶ່ງຈະມີ intField: double.

ມີທຸງຕໍ່ໄປນີ້ເພື່ອຈັດການກັບສະຖານະການນີ້:

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

ໃນປັດຈຸບັນພວກເຮົາມີໂຟນເດີທີ່ມີການແບ່ງປັນທີ່ສາມາດອ່ານເຂົ້າໄປໃນກອບຂໍ້ມູນດຽວແລະ parquet ທີ່ຖືກຕ້ອງຂອງ showcase ທັງຫມົດ. ແມ່ນບໍ? ບໍ່.

ພວກເຮົາຕ້ອງຈື່ໄວ້ວ່າພວກເຮົາໄດ້ລົງທະບຽນຕາຕະລາງໃນ Hive. Hive ບໍ່ແມ່ນຕົວພິມນ້ອຍໃຫຍ່ໃນຊື່ພາກສະຫນາມ, ໃນຂະນະທີ່ parquet ແມ່ນຕົວພິມນ້ອຍ. ດັ່ງນັ້ນ, ການແບ່ງສ່ວນທີ່ມີ schemas: field1: int, ແລະ Field1: int ແມ່ນຄືກັນສໍາລັບ Hive, ແຕ່ບໍ່ແມ່ນສໍາລັບ Spark. ຢ່າລືມປ່ຽນຊື່ຊ່ອງຂໍ້ມູນເປັນຕົວພິມນ້ອຍ.

ຫຼັງຈາກນັ້ນ, ທຸກສິ່ງທຸກຢ່າງເບິ່ງຄືວ່າດີ.

ຢ່າງໃດກໍຕາມ, ບໍ່ແມ່ນທັງຫມົດງ່າຍດາຍດັ່ງນັ້ນ. ມີບັນຫາທີສອງ, ຍັງເປັນທີ່ຮູ້ຈັກກັນດີ. ເນື່ອງຈາກແຕ່ລະພາທິຊັນໃໝ່ຖືກບັນທຶກແຍກຕ່າງຫາກ, ໂຟນເດີພາທິຊັນຈະມີໄຟລ໌ບໍລິການ Spark, ຕົວຢ່າງ, ທຸງຄວາມສໍາເລັດຂອງການດໍາເນີນງານ _SUCCESS. ນີ້ຈະສົ່ງຜົນໃຫ້ເກີດຄວາມຜິດພາດໃນເວລາທີ່ພະຍາຍາມ parquet. ເພື່ອຫຼີກເວັ້ນການນີ້, ທ່ານຈໍາເປັນຕ້ອງໄດ້ກໍານົດການຕັ້ງຄ່າເພື່ອປ້ອງກັນບໍ່ໃຫ້ Spark ເພີ່ມໄຟລ໌ບໍລິການໃສ່ໂຟນເດີ:

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

ມັນເບິ່ງຄືວ່າໃນປັດຈຸບັນທຸກໆມື້ມີການເພີ່ມພາທິຊັນ parquet ໃຫມ່ເຂົ້າໄປໃນໂຟເດີ showcase ເປົ້າຫມາຍ, ບ່ອນທີ່ຂໍ້ມູນການແຍກວິເຄາະສໍາລັບມື້ແມ່ນຕັ້ງຢູ່. ພວກ​ເຮົາ​ໄດ້​ເອົາ​ໃຈ​ໃສ່​ລ່ວງ​ຫນ້າ​ວ່າ​ບໍ່​ມີ​ການ​ແບ່ງ​ປັນ​ທີ່​ມີ​ຂໍ້​ຂັດ​ແຍ່ງ​ປະ​ເພດ​ຂໍ້​ມູນ​.

ແຕ່, ພວກເຮົາມີບັນຫາທີສາມ. ໃນປັດຈຸບັນ schema ທົ່ວໄປແມ່ນບໍ່ຮູ້ຈັກ, ນອກຈາກນັ້ນ, ຕາຕະລາງໃນ Hive ມີ schema ທີ່ບໍ່ຖືກຕ້ອງ, ເນື່ອງຈາກວ່າແຕ່ລະ partition ໃຫມ່ມັກຈະນໍາສະເຫນີການບິດເບືອນເຂົ້າໄປໃນ schema ໄດ້.

ທ່ານ ຈຳ ເປັນຕ້ອງລົງທະບຽນຕາຕະລາງຄືນ ໃໝ່. ນີ້ສາມາດເຮັດໄດ້ງ່າຍໆ: ອ່ານ parquet ຂອງຫນ້າຮ້ານອີກເທື່ອຫນຶ່ງ, ເອົາ schema ແລະສ້າງ DDL ໂດຍອີງໃສ່ມັນ, ເຊິ່ງການລົງທະບຽນໂຟນເດີໃຫມ່ໃນ Hive ເປັນຕາຕະລາງພາຍນອກ, ປັບປຸງ schema ຂອງຫນ້າຮ້ານເປົ້າຫມາຍ.

ພວກເຮົາມີບັນຫາທີສີ່. ເມື່ອພວກເຮົາລົງທະບຽນຕາຕະລາງຄັ້ງທໍາອິດ, ພວກເຮົາອີງໃສ່ Spark. ໃນປັດຈຸບັນພວກເຮົາເຮັດມັນເອງ, ແລະພວກເຮົາຈໍາເປັນຕ້ອງຈື່ໄວ້ວ່າທົ່ງນາ parquet ສາມາດເລີ່ມຕົ້ນດ້ວຍຕົວອັກສອນທີ່ບໍ່ໄດ້ຮັບອະນຸຍາດໃຫ້ 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(",", ",`").ແທນທີ່("array<`", "array<") ເຮັດໃຫ້ DDL ປອດໄພ, i.e. ແທນທີ່ຈະເປັນ:

create table tname (_field1 string, 1field string)

ດ້ວຍຊື່ຊ່ອງຂໍ້ມູນເຊັ່ນ "_field1, 1field", ຄວາມປອດໄພ DDL ແມ່ນເຮັດໃຫ້ຊື່ພາກສະຫນາມຖືກຫນີ: ສ້າງຕາຕະລາງ `tname` (`_field1` string, `1field` string).

ຄໍາຖາມທີ່ເກີດຂື້ນ: ວິທີການຮັບ dataframe ຢ່າງຖືກຕ້ອງກັບ schema ຄົບຖ້ວນ (ໃນລະຫັດ pf)? ວິທີການໄດ້ຮັບ pf ນີ້? ນີ້ແມ່ນບັນຫາທີຫ້າ. Reread ໂຄງ ການ ຂອງ ການ ແບ່ງ ປັນ ທັງ ຫມົດ ຈາກ ໂຟນ ເດີ ທີ່ ມີ ໄຟລ ໌ parquet ຂອງ showcase ເປົ້າ ຫມາຍ? ວິທີການນີ້ແມ່ນປອດໄພທີ່ສຸດ, ແຕ່ມີຄວາມຫຍຸ້ງຍາກ.

schema ແມ່ນແລ້ວຢູ່ໃນ Hive. ທ່ານສາມາດໄດ້ຮັບ schema ໃຫມ່ໂດຍການສົມທົບ schema ຂອງຕາຕະລາງທັງຫມົດແລະການແບ່ງປັນໃຫມ່. ດັ່ງນັ້ນທ່ານຈໍາເປັນຕ້ອງເອົາຕາຕະລາງຕາຕະລາງຈາກ Hive ແລະສົມທົບກັບ schema ຂອງການແບ່ງປັນໃຫມ່. ນີ້ສາມາດເຮັດໄດ້ໂດຍການອ່ານ metadata ການທົດສອບຈາກ Hive, ບັນທຶກມັນໄວ້ໃນໂຟນເດີຊົ່ວຄາວ, ແລະໃຊ້ Spark ເພື່ອອ່ານທັງສອງພາທິຊັນໃນເວລາດຽວກັນ.

ໃນຄວາມເປັນຈິງ, ມີທຸກສິ່ງທີ່ທ່ານຕ້ອງການ: ຕາຕະລາງຕາຕະລາງຕົ້ນສະບັບໃນ Hive ແລະການແບ່ງປັນໃຫມ່. ພວກເຮົາຍັງມີຂໍ້ມູນ. ມັນຍັງຄົງພຽງແຕ່ໄດ້ຮັບ schema ໃຫມ່ທີ່ປະສົມປະສານ schema ຫນ້າຮ້ານແລະຊ່ອງໃຫມ່ຈາກການແບ່ງປັນທີ່ສ້າງຂຶ້ນ:

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 ແລະການສ້າງຫນ້າຮ້ານໂດຍອີງໃສ່ມັນເຮັດໃຫ້ການເອົາຊະນະຄວາມຫຍຸ້ງຍາກຫຼາຍ, ວິທີແກ້ໄຂທີ່ທ່ານຕ້ອງຊອກຫາແຍກຕ່າງຫາກ. ແລະເຖິງແມ່ນວ່າວິທີແກ້ໄຂເຫຼົ່ານີ້ແມ່ນງ່າຍດາຍ, ມັນໃຊ້ເວລາຫຼາຍເພື່ອຊອກຫາໃຫ້ເຂົາເຈົ້າ.

ເພື່ອປະຕິບັດການກໍ່ສ້າງຂອງ showcase, ຂ້າພະເຈົ້າຕ້ອງ:

• ເພີ່ມພາທິຊັນໃສ່ບ່ອນວາງສະແດງ, ກໍາຈັດໄຟລ໌ການບໍລິການ
• ຈັດການກັບຊ່ອງຫວ່າງເປົ່າໃນຂໍ້ມູນແຫຼ່ງທີ່ Spark ໄດ້ພິມ
• ຄາສທປະເພດງ່າຍໆໃສ່ສະຕຣິງ
• ປ່ຽນຊື່ຊ່ອງຂໍ້ມູນເປັນຕົວພິມນ້ອຍ
• ການອັບໂຫລດຂໍ້ມູນແຍກຕ່າງຫາກແລະການລົງທະບຽນຕາຕະລາງໃນ Hive (ການຜະລິດ DDL)
• ຢ່າລືມອອກຈາກຊື່ພາກສະຫນາມທີ່ອາດຈະບໍ່ເຂົ້າກັນໄດ້ກັບ Hive
• ຮຽນຮູ້ວິທີການປັບປຸງການລົງທະບຽນຕາຕະລາງໃນ Hive

ສະຫລຸບລວມແລ້ວ, ພວກເຮົາສັງເກດວ່າການຕັດສິນໃຈທີ່ຈະສ້າງປ່ອງຢ້ຽມຂອງຮ້ານແມ່ນ fraught ກັບ pitfalls ຫຼາຍ. ດັ່ງນັ້ນ, ໃນກໍລະນີທີ່ມີຄວາມຫຍຸ້ງຍາກໃນການປະຕິບັດ, ມັນດີກວ່າທີ່ຈະຕິດຕໍ່ກັບຄູ່ຮ່ວມງານທີ່ມີປະສົບການທີ່ມີຄວາມຊໍານານທີ່ປະສົບຜົນສໍາເລັດ.

ຂອບໃຈສໍາລັບການອ່ານບົດຄວາມນີ້, ພວກເຮົາຫວັງວ່າທ່ານຈະຊອກຫາຂໍ້ມູນທີ່ເປັນປະໂຫຍດ.

ແຫຼ່ງຂໍ້ມູນ: www.habr.com