Spark schemaEvolution בפועל

קוראים יקרים, יום טוב!

במאמר זה, היועץ המוביל של תחום העסקים Big Data Solutions של Neoflex מתאר בפירוט את האפשרויות לבניית חלונות ראווה של מבנה משתנה באמצעות Apache Spark.

כחלק מפרויקט ניתוח נתונים, מתעוררת לעתים קרובות המשימה של בניית חלונות ראווה המבוססים על נתונים בעלי מבנה רופף.

בדרך כלל מדובר ביומנים, או תגובות ממערכות שונות, שנשמרו כ-JSON או XML. הנתונים עולים ל-Hadoop, ואז צריך לבנות מהם חלון ראווה. אנחנו יכולים לארגן גישה לחלון הראווה שנוצר, למשל, דרך אימפלה.

במקרה זה, הסכימה של חלון הראווה של היעד אינה ידועה מראש. יתרה מכך, גם את התוכנית לא ניתן לערוך מראש, שכן היא תלויה בנתונים, ואנו עוסקים בנתונים המאוד רופפים הללו.

לדוגמה, היום נרשמה התגובה הבאה:

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

ומחר מאותה מערכת מגיעה התשובה הבאה:

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

כתוצאה מכך צריך להוסיף עוד שדה אחד לוויטרינה - תיאור, ואף אחד לא יודע אם הוא יגיע או לא.

המשימה של יצירת חלון ראווה על נתונים כאלה היא די סטנדרטית, ול-Spark יש מספר כלים לכך. לניתוח נתוני המקור, קיימת תמיכה הן ב-JSON והן ב-XML, ובסכימה שלא הייתה ידועה בעבר, ניתנת תמיכה ב-schemaEvolution.

במבט ראשון, הפתרון נראה פשוט. אתה צריך לקחת תיקיה עם JSON ולקרוא אותה לתוך מסגרת נתונים. Spark יצור סכמה, יהפוך נתונים מקוננים למבנים. יתרה מכך, הכל צריך להישמר בפרקט, הנתמך גם באימפלה, על ידי רישום חלון הראווה ב-Hive metastore.

הכל נראה פשוט.

עם זאת, לא ברור מהדוגמאות הקצרות בתיעוד מה לעשות עם מספר בעיות בפועל.

התיעוד מתאר גישה לא ליצור חזית חנות, אלא לקרוא JSON או XML לתוך מסגרת נתונים.

כלומר, זה פשוט מראה כיצד לקרוא ולנתח JSON:

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

זה מספיק כדי להפוך את הנתונים לזמינים ל-Spark.

בפועל, הסקריפט הרבה יותר מסובך מסתם קריאת קבצי JSON מתיקיה ויצירת Dataframe. המצב נראה כך: יש כבר חלון ראווה מסוים, נתונים חדשים מגיעים כל יום, צריך להוסיף אותם לחלון הראווה, בלי לשכוח שהתכנית עשויה להיות שונה.

התוכנית הרגילה לבניית חלון ראווה היא כדלקמן:

שלב 1. הנתונים נטענים לתוך Hadoop עם טעינה יומית לאחר מכן ומתווספים למחיצה חדשה. מסתבר שתיקיה עם נתונים ראשוניים מחולקים לפי יום.

שלב 2. במהלך הטעינה הראשונית, תיקייה זו נקראת ומנתקת על ידי Spark. מסגרת הנתונים המתקבלת נשמרת בפורמט שניתן לנתח, למשל, בפרקט, שאותו ניתן לייבא לאימפלה. זה יוצר חלון ראווה של יעד עם כל הנתונים שהצטברו עד לנקודה זו.

שלב 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, ואז אנחנו שומרים את ה-dataframe כפרקט, ורושמים אותו ב-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 + "/")

כל מה שנותר הוא להירשם מחדש בכוורת כדי לעדכן את הסכימה.
עם זאת, כאן מתעוררות בעיות.

בעיה ראשונה. במוקדם או במאוחר, הפרקט שיתקבל יהיה בלתי קריא. זה נובע מהאופן שבו פרקט ו-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)

כעת יש לנו תיקייה שבה יש מחיצות שניתן לקרוא ל-dataframe בודד ופרקט תקף של כל הוויטרינה. כן? לא.

עלינו לזכור שרשמנו את הטבלה בכוורת. 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 כטבלה חיצונית, תוך עדכון הסכימה של חלון הראווה היעד.

יש לנו בעיה רביעית. כשרשמנו את הטבלה בפעם הראשונה, סמכנו על ספארק. עכשיו אנחנו עושים את זה בעצמנו, וצריך לזכור ששדות פרקט יכולים להתחיל בדמויות שאסור לכוורת. לדוגמה, Spark זורק קווים שהוא לא הצליח לנתח בשדה "corrupt_record". לא ניתן לרשום שדה כזה בכוורת מבלי להימלט.

בידיעה זו, אנו מקבלים את התוכנית:

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`, מחרוזת `1field`).

נשאלת השאלה: כיצד להשיג כראוי Dataframe עם סכימה מלאה (בקוד pf)? איך משיגים את ה-PF הזה? זו הבעיה החמישית. לקרוא מחדש את הסכימה של כל המחיצות מהתיקייה עם קבצי פרקט של חלון הראווה היעד? שיטה זו היא הבטוחה ביותר, אך קשה.

הסכימה כבר נמצאת בכוורת. ניתן לקבל סכימה חדשה על ידי שילוב של הסכימה של הטבלה כולה והמחיצה החדשה. אז אתה צריך לקחת את סכימת הטבלה מ-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 הקליד
• יצקו טיפוסים פשוטים למחרוזת
• המרת שמות שדות לאותיות קטנות
• העלאת נתונים נפרדת ורישום טבלה בכוורת (דור DDL)
• אל תשכח לברוח משמות שדות שעלולים להיות לא תואמים ל-Hive
• למד כיצד לעדכן את רישום הטבלה בכוורת

לסיכום, נציין כי ההחלטה לבנות חלונות ראווה טומנת בחובה מלכודות רבות. לכן, במקרה של קשיים ביישום, עדיף לפנות לשותף מנוסה עם מומחיות מוצלחת.

תודה שקראת מאמר זה, אנו מקווים שתמצא את המידע שימושי.

מקור: www.habr.com