בנו פתרון תחזוקה חזוי עם Amazon Kinesis, AWS Glue ו-Amazon SageMaker

ארגונים בונים ומשתמשים יותר ויותר בפתרונות המונעים על ידי למידת מכונה (ML) למגוון מקרי שימוש ובעיות, כולל תחזוקה חזויה של חלקי מכונה, המלצות מוצר המבוססות על העדפות לקוחות, פרופיל אשראי, ניהול תוכן, זיהוי הונאה ועוד. ברבים מהתרחישים הללו, ניתן לשפר עוד יותר את האפקטיביות והיתרונות המופקים מפתרונות אלה המופעלים על ידי ML ​​כאשר הם יכולים לעבד ולהפיק תובנות מאירועי נתונים בזמן כמעט אמת.

למרות שהערך העסקי והיתרונות של פתרונות המופעלים על ידי ML ​​כמעט בזמן אמת מבוססים היטב, הארכיטקטורה הנדרשת ליישום פתרונות אלה בקנה מידה עם אמינות וביצועים מיטביים היא מסובכת. פוסט זה מתאר כיצד ניתן לשלב אמזון קינסי, דבק AWS, ו אמזון SageMaker לבנות פתרון הנדסה והסקת תכונות כמעט בזמן אמת לתחזוקה חזויה.

השתמש בסקירת המקרים

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

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

כשל במכונה מורכב מחמישה מצבי כשל עצמאיים:

• כשל בלבוש כלי (TWF)
• כשל בפיזור חום (HDF)
• הפסקת חשמל (PWF)
• כשל במתח יתר (OSF)
• כשל אקראי (RNF)

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

סקירת פתרונות

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

התרשים הבא מציג את הארכיטקטורה של הפתרון הכולל שלנו.

הפתרון מורכב באופן כללי מהסעיפים הבאים, אשר מוסברים בפירוט בהמשך הפוסט הזה:

• הזרמת מקור נתונים והטמעה - אנו משתמשים זרמי נתונים של אמזון קינסי לאסוף נתונים זורמים מחיישני השטח בקנה מידה ולהפוך אותם לזמינים לעיבוד נוסף.
• הנדסת תכונות כמעט בזמן אמת - אנו משתמשים בעבודות הזרמת AWS Glue כדי לקרוא נתונים מזרם נתונים של Kinesis ולבצע עיבוד נתונים והנדסת תכונות, לפני אחסון התכונות הנגזרות ב שירות אחסון פשוט של אמזון (אמזון S3). Amazon S3 מספק אפשרות אמינה וחסכונית לאחסון כמויות גדולות של נתונים.
• הדרכה ופריסה מודל - אנו משתמשים במערך הנתונים של תחזוקה חזוי AI4I ממאגר הנתונים של UCI כדי להכשיר מודל ML המבוסס על אלגוריתם XGBoost באמצעות SageMaker. לאחר מכן אנו פורסים את המודל המאומן לנקודת קצה אסינכרונית של SageMaker.
• הסקת ML כמעט בזמן אמת - לאחר שהתכונות זמינות באמזון S3, עלינו להפיק מסקנות מהמודל שנפרס בזמן כמעט אמת. נקודות קצה אסינכרוניות של SageMaker מתאימות היטב לדרישה זו מכיוון שהן תומכות בגדלים גדולים יותר של מטען (עד 1 GB) ויכולות ליצור מסקנות תוך דקות (עד לכל היותר 15 דקות). אנו משתמשים בהתראות אירוע S3 כדי להפעיל AWS למבדה פונקציה להפעיל נקודת קצה אסינכרונית של SageMaker. נקודות קצה אסינכרוניות של SageMaker מקבלים מיקומי S3 כקלט, מייצרים מסקנות מהמודל שנפרס, וכותבים את ההסקות הללו בחזרה לאמזון S3 בזמן כמעט אמת.

קוד המקור של פתרון זה נמצא ב- GitHub. הפתרון נבדק ויש להפעיל אותו ב-us-east-1.

אנו משתמשים ב- AWS CloudFormation תבנית, פרוסה באמצעות מודל יישום ללא שרת AWS (AWS SAM), ומחברות SageMaker כדי לפרוס את הפתרון.

תנאים מוקדמים

כדי להתחיל, כתנאי מוקדם, אתה חייב לקבל את SAM CLI, פייתון 3, ו PIP מוּתקָן. אתה חייב גם את ממשק שורת הפקודה של AWS (AWS CLI) מוגדר כהלכה.

פרוס את הפתרון

אתה יכול להשתמש AWS CloudShell כדי להפעיל את השלבים האלה. CloudShell היא מעטפת מבוססת דפדפן אשר מאומתת מראש עם אישורי הקונסולה שלך וכוללת כלי פיתוח ותפעול נפוצים מותקנים מראש (כגון AWS SAM, AWS CLI ו-Python). לכן, אין צורך בהתקנה או תצורה מקומית.

• אנו מתחילים ביצירת דלי S3 שבו אנו מאחסנים את הסקריפט עבור עבודת הסטרימינג של AWS Glue שלנו. הפעל את הפקודה הבאה בטרמינל שלך כדי ליצור דלי חדש:

aws s3api create-bucket --bucket sample-script-bucket-$RANDOM --region us-east-1

• רשום את שם הדלי שנוצר.

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

git clone https://github.com/aws-samples/amazon-sagemaker-predictive-maintenance

• נווט אל ספריית sam-template:

cd amazon-sagemaker-predictive-maintenance/sam-template

• הפעל את הפקודה הבאה כדי להעתיק את סקריפט העבודה של AWS Glue (מתוך glue_streaming/app.py) לדלי S3 שיצרת:

aws s3 cp glue_streaming/app.py s3://sample-script-bucket-30232/glue_streaming/app.py

• כעת אתה יכול להמשיך עם הבנייה והפריסה של הפתרון, דרך תבנית CloudFormation דרך AWS SAM. הפעל את הפקודה הבאה:

sam deploy --guided

• ספק ארגומנטים עבור הפריסה כגון שם המחסנית, אזור AWS מועדף (us-east-1,) ו GlueScriptsBucket.

ודא שאתה מספק את אותו דלי S3 שיצרת קודם לכן עבור דלי S3 של AWS Glue script (פרמטר GlueScriptsBucket בצילום המסך הבא).

לאחר שתספק את הארגומנטים הנדרשים, AWS SAM מתחיל את פריסת המחסנית. צילום המסך הבא מציג את המשאבים שנוצרו.

לאחר פריסת הערימה בהצלחה, אתה אמור לראות את ההודעה הבאה.

• במסוף AWS CloudFormation, פתח את המחסנית (עבור פוסט זה, nrt-streaming-inference) שסופק בעת פריסת תבנית CloudFormation.
• על משאבים לשונית, שים לב למזהה מופע המחברת של SageMaker.

• במסוף SageMaker, פתח את המופע הזה.

מופע המחברת של SageMaker כבר מכיל את המחברות הנדרשות נטענות מראש.

נווט אל תיקיית המחברות ופתח ופעל לפי ההוראות בתוך המחברות (Data_Pre-Processing.ipynb ו ModelTraining-Evaluation-and-Deployment.ipynb) כדי לחקור את מערך הנתונים, לבצע עיבוד מקדים והנדסת תכונות, ולאמן ולפרוס את המודל לנקודת קצה אסינכרונית של SageMaker.

הזרמת מקור נתונים והטמעה

Kinesis Data Streams הוא שירות הזרמת נתונים בזמן אמת ללא שרת, ניתן להרחבה ועמיד בו אתה יכול להשתמש כדי לאסוף ולעבד זרמים גדולים של רשומות נתונים בזמן אמת. Kinesis Data Streams מאפשר לכידה, עיבוד ואחסון של זרמי נתונים ממגוון מקורות, כגון נתוני יומן תשתית IT, יומני יישומים, מדיה חברתית, הזנות נתוני שוק, נתוני זרם קליקים באינטרנט, התקני IoT וחיישנים ועוד. אתה יכול לספק זרם נתונים של Kinesis במצב לפי דרישה או במצב אספקה ​​בהתאם לתפוקה ודרישות קנה המידה. למידע נוסף, ראה בחירת מצב קיבולת זרם הנתונים.

במקרה השימוש שלנו, אנו מניחים שחיישנים שונים שולחים מדידות כמו טמפרטורה, מהירות סיבוב, מומנט ובלאי הכלים לזרם נתונים. Kinesis Data Streams פועל כמשפך לאיסוף והטמעת זרמי נתונים.

אנו משתמשים מחולל הנתונים של אמזון קינסיס (KDG) בהמשך הפוסט הזה כדי ליצור ולשלוח נתונים לזרם נתונים של Kinesis, המדמה נתונים שנוצרים על ידי חיישנים. הנתונים מזרם הנתונים חיישן-נתונים-זרם נקלטים ומעובדים באמצעות עבודת הזרמת AWS Glue, עליה נדון בהמשך.

הנדסת תכונות כמעט בזמן אמת

משרות הזרמת AWS Glue לספק דרך נוחה לעיבוד נתונים זורמים בקנה מידה, ללא צורך בניהול סביבת המחשוב. AWS Glue מאפשר לך לבצע פעולות חילוץ, טרנספורמציה וטעינה (ETL) על הזרמת נתונים באמצעות משימות הפועלות ברציפות. AWS Glue Streaming ETL בנוי על מנוע ה-Apache Spark Structured Streaming, ויכול להטמיע זרמים מ-Kinesis, Apache Kafka, ו אמזון מנוהל סטרימינג עבור אפאצ'י קפקא (אמזון MSK).

עבודת ה-ETL הזרמת יכולה להשתמש הן בהמרה מובנות ב-AWS Glue והן בטרנספורמציות שהן מקוריות ל-Apache Spark Structured Streaming. אתה יכול גם להשתמש ב-Spark ML ו MLLib ספריות בעבודות דבק של AWS לעיבוד קל יותר של תכונות באמצעות ספריות מסייעות זמינות.

אם הסכימה של מקור הנתונים הזורמים נקבעה מראש, תוכל לציין אותה בטבלה של קטלוג הנתונים של AWS. אם לא ניתן לקבוע את הגדרת הסכימה מראש, תוכל להפעיל את זיהוי הסכימה בעבודת ה-ETL הזורמת. לאחר מכן, התפקיד קובע אוטומטית את הסכימה מהנתונים הנכנסים. בנוסף, אתה יכול להשתמש ב- AWS Glue Schema Registry כדי לאפשר גילוי מרכזי, שליטה והתפתחות של סכימות זרם נתונים. אתה יכול לשלב עוד יותר את מרשם הסכימה עם קטלוג הנתונים כדי להשתמש באופן אופציונלי בסכמות המאוחסנות במרשם הסכימה בעת יצירה או עדכון של טבלאות או מחיצות דבק של AWS בקטלוג הנתונים.

עבור פוסט זה, אנו יוצרים טבלת קטלוג נתוני דבק של AWS (sensor-stream) עם זרם הנתונים של Kinesis שלנו כמקור ולהגדיר את הסכימה עבור נתוני החיישנים שלנו.

אנו יוצרים מסגרת נתונים דינמית של AWS Glue מטבלת קטלוג הנתונים כדי לקרוא את הנתונים הזורמים מ-Kinesis. אנו גם מציינים את האפשרויות הבאות:

• גודל חלון של 60 שניות, כך שעבודת הדבק של AWS קוראת ומעבדת נתונים בחלונות של 60 שניות
• עמדת המוצא TRIM_HORIZON, כדי לאפשר קריאה מהרשומות העתיקות ביותר בזרם הנתונים של Kinesis

אנו משתמשים גם ב-Spark MLlib's StringIndexer תכונה שנאי כדי לקודד את סוג עמודת המחרוזת לאינדקסים של תוויות. טרנספורמציה זו מיושמת באמצעות Spark ML Pipelines. Spark ML Pipelines לספק קבוצה אחידה של ממשקי API ברמה גבוהה עבור אלגוריתמי ML כדי להקל על שילוב אלגוריתמים מרובים לצינור או זרימת עבודה אחת.

אנו משתמשים בממשק API של foreachBatch כדי להפעיל פונקציה בשם processBatch, אשר בתורה מעבדת את הנתונים שאליהם מתייחסים מסגרת הנתונים הזו. ראה את הקוד הבא:

# Read from Kinesis Data Stream
sourceStreamData = glueContext.create_data_frame.from_catalog(database = "sensordb", table_name = "sensor-stream", transformation_ctx = "sourceStreamData", additional_options = {"startingPosition": "TRIM_HORIZON"})
type_indexer = StringIndexer(inputCol="type", outputCol="type_enc", stringOrderType="alphabetAsc")
pipeline = Pipeline(stages=[type_indexer])
glueContext.forEachBatch(frame = sourceStreamData, batch_function = processBatch, options = {"windowSize": "60 seconds", "checkpointLocation": checkpoint_location})

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

אנו גם מחלקים מחדש את מחיצות ה-AWS Glue למחיצה אחת, כדי למנוע יותר מדי קבצים קטנים באמזון S3. שימוש במספר קבצים קטנים עלול להפריע לביצועי הקריאה, מכיוון שהוא מגביר את התקורה הקשורה לחיפוש, פתיחה וקריאה של כל קובץ. לבסוף אנו כותבים את התכונות כדי ליצור מסקנות לקידומת (תכונות) בתוך דלי S3. ראה את הקוד הבא:

# Function that gets called to perform processing, feature engineering and writes to S3 for every micro batch of streaming data from Kinesis.
def processBatch(data_frame, batchId):
transformer = pipeline.fit(data_frame)
now = datetime.datetime.now()
year = now.year
month = now.month
day = now.day
hour = now.hour
minute = now.minute
if (data_frame.count() > 0):
data_frame = transformer.transform(data_frame)
data_frame = data_frame.drop("type")
data_frame = DynamicFrame.fromDF(data_frame, glueContext, "from_data_frame")
data_frame.printSchema()
# Write output features to S3
s3prefix = "features" + "/year=" + "{:0>4}".format(str(year)) + "/month=" + "{:0>2}".format(str(month)) + "/day=" + "{:0>2}".format(str(day)) + "/hour=" + "{:0>2}".format(str(hour)) + "/min=" + "{:0>2}".format(str(minute)) + "/batchid=" + str(batchId)
s3path = "s3://" + out_bucket_name + "/" + s3prefix + "/"
print("-------write start time------------")
print(str(datetime.datetime.now()))
data_frame = data_frame.toDF().repartition(1)
data_frame.write.mode("overwrite").option("header",False).csv(s3path)
print("-------write end time------------")
print(str(datetime.datetime.now()))

הדרכה ופריסה של מודל

SageMaker הוא שירות ML מנוהל ומשולב במלואו המאפשר למדעני נתונים ומהנדסי ML לבנות, לאמן ולפרוס מודלים של ML במהירות ובקלות.

בתוך המחברת Data_Pre-Processing.ipynb, ראשית, אנו מייבאים את מערך הנתונים של AI4I Predictive Maintenance ממאגר הנתונים של UCI ומבצעים ניתוח נתונים חקרניים (EDA). אנו גם מבצעים הנדסת תכונות כדי להפוך את התכונות שלנו לשימושיות יותר עבור אימון המודל.

לדוגמה, בתוך מערך הנתונים, יש לנו תכונה בשם סוג, המייצגת את סוג האיכות של המוצר כ-L (נמוך), M (בינוני) או H (גבוה). מכיוון שזוהי תכונה קטגורית, עלינו לקודד אותה לפני אימון המודל שלנו. אנו משתמשים ב-LabelEncoder של Scikit-Learn כדי להשיג זאת:

from sklearn.preprocessing import LabelEncoder
type_encoder = LabelEncoder()
type_encoder.fit(origdf['type'])
type_values = type_encoder.transform(origdf['type'])

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

משרות הדרכה של SageMaker לספק דרך עוצמתית וגמישה לאימון דגמי ML ב- SageMaker. SageMaker מנהלת את תשתית המחשוב הבסיסית ומספקת אפשרויות מרובות לבחירה, עבור דרישות הכשרה מגוונות של מודל, בהתבסס על מקרה השימוש.

xgb = sagemaker.estimator.Estimator(container,
role,
instance_count=1,
instance_type='ml.c4.4xlarge',
output_path=xgb_upload_location,
sagemaker_session=sagemaker_session)
xgb.set_hyperparameters(max_depth=5,
eta=0.2,
gamma=4,
min_child_weight=6,
subsample=0.8,
silent=0,
objective='binary:hinge',
num_round=100)

xgb.fit({'train': s3_train_channel, 'validation': s3_valid_channel})

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

endpoint_config_name = resource_name.format("EndpointConfig")
create_endpoint_config_response = sm_client.create_endpoint_config(
EndpointConfigName=endpoint_config_name,
ProductionVariants=[
{
"VariantName": "variant1",
"ModelName": model_name,
"InstanceType": "ml.m5.xlarge",
"InitialInstanceCount": 1,
}
],
AsyncInferenceConfig={
"OutputConfig": {
"S3OutputPath": f"s3://{bucket}/{prefix}/output",
#Specify Amazon SNS topics
"NotificationConfig": {
"SuccessTopic": "arn:aws:sns:<region>:<account-id>:<success-sns-topic>",
"ErrorTopic": "arn:aws:sns:<region>:<account-id>:<error-sns-topic>",
}},
"ClientConfig": {"MaxConcurrentInvocationsPerInstance": 4},
},)

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

endpoint_name = resource_name.format("Endpoint")
create_endpoint_response = sm_client.create_endpoint(
EndpointName=endpoint_name, EndpointConfigName=endpoint_config_name)

מסקנות כמעט בזמן אמת

SageMaker הסקה אסינכרונית נקודות קצה מספקות את היכולת לעמוד בתור בקשות הסקת מסקנות נכנסות ולעבד אותן באופן אסינכרוני בזמן כמעט אמת. זה אידיאלי עבור יישומים שיש להם בקשות הסקה עם גדלי מטען גדולים יותר (עד 1 GB), עשויים לדרוש זמני עיבוד ארוכים יותר (עד 15 דקות), ויש להם דרישות השהייה כמעט בזמן אמת. הסקה אסינכרונית גם מאפשרת לך לחסוך בעלויות על ידי שינוי קנה מידה אוטומטי של ספירת המופעים לאפס כאשר אין בקשות לעיבוד, כך שאתה משלם רק כאשר נקודת הקצה שלך מעבדת בקשות.

אתה יכול ליצור נקודת קצה אסינכרונית של SageMaker בדומה לאופן שבו אתה יוצר נקודת קצה בזמן אמת ובנוסף לציין את AsyncInferenceConfig אובייקט, תוך יצירת תצורת נקודת הקצה שלך עם השדה EndpointConfig ב-CreateEndpointConfig API. התרשים הבא מציג את זרימת העבודה של הסקת מסקנות וכיצד נקודת קצה של הסקה אסינכרונית יוצרת הסקה.

כדי להפעיל את נקודת הקצה האסינכרונית, יש לאחסן את מטען הבקשה ב-Amazon S3 ויש לספק התייחסות למטען זה כחלק מבקשת InvokeEndpointAsync. עם הפניה, SageMaker מעמיד את הבקשה בתור לעיבוד ומחזיר מזהה ומיקום פלט כתגובה. לאחר העיבוד, SageMaker ממקם את התוצאה במיקום Amazon S3. אתה יכול לבחור לקבל הודעות הצלחה או שגיאה עם שירות התראה פשוט של אמזון (Amazon SNS).

בדוק את הפתרון מקצה לקצה

כדי לבדוק את הפתרון, בצע את השלבים הבאים:

• במסוף AWS CloudFormation, פתח את המחסנית שיצרת קודם לכן (nrt-streaming-inference).
• על יציאות לשונית, העתק את השם של דלי S3 (EventsBucket).

זהו דלי S3 שאליו משימת הסטרימינג של AWS Glue שלנו כותבת תכונות לאחר קריאה ועיבוד מזרם הנתונים של Kinesis.

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

• בקונסולת Amazon S3, נווט אל הדלי EventsBucket.
• על מאפיין הכרטיסייה, ב הודעות על אירוע סעיף, בחר צור התראה על אירוע.

• בעד שם אירוע, להיכנס invoke-endpoint-lambda.
• בעד קידומת, להיכנס features/.
• בעד סִיוֹמֶת, להיכנס .csv.
• בעד סוגי אירועים, בחר כל האובייקטים יוצרים אירועים.

• בעד יַעַד, בחר פונקציית למדה.
• בעד פונקציית למדה, ובחר את הפונקציה invoke-endpoint-asynch.
• בחרו שמור את השינויים.

• במסוף הדבק של AWS, פתח את העבודה GlueStreaming-Kinesis-S3.
• בחרו הפעל עבודה.

לאחר מכן אנו משתמשים במחולל הנתונים של Kinesis (KDG) כדי לדמות חיישנים השולחים נתונים לזרם הנתונים של Kinesis. אם זו הפעם הראשונה שאתה משתמש ב-KDG, עיין ב סקירה כללית עבור ההגדרה הראשונית. ה-KDG מספק תבנית CloudFormation כדי ליצור את המשתמש ולהקצות מספיק הרשאות כדי להשתמש ב-KDG לשליחת אירועים ל-Kinesis. הפעל את תבנית CloudFormation בחשבון AWS שבו אתה משתמש לבניית הפתרון בפוסט זה. לאחר הגדרת ה-KDG, התחבר וגשת ל-KDG כדי לשלוח אירועי בדיקה לזרם הנתונים שלנו ב-Kinesis.

• השתמש באזור שבו יצרת את זרם הנתונים של Kinesis (us-east-1).
• בתפריט הנפתח, בחר את זרם הנתונים sensor-data-stream.
• ב שיאים בשנייה בחר, בחר קָבוּעַ והזן 100.
• בטל את הבחירה דחוס רשומות.
• בעד תבנית שיא, השתמש בתבנית הבאה:

{
"air_temperature": {{random.number({"min":295,"max":305, "precision":0.01})}},
"process_temperature": {{random.number({"min":305,"max":315, "precision":0.01})}},
"rotational_speed": {{random.number({"min":1150,"max":2900})}},
"torque": {{random.number({"min":3,"max":80, "precision":0.01})}},
"tool_wear": {{random.number({"min":0,"max":250})}},
"type": "{{random.arrayElement(["L","M","H"])}}"
}

• נְקִישָׁה לשלוח נתונים כדי להתחיל לשלוח נתונים לזרם הנתונים של Kinesis.

עבודת הזרמת AWS Glue קוראת ומחלצת מיקרו-אצט של נתונים (המייצגים קריאות חיישנים) מזרם הנתונים של Kinesis בהתבסס על גודל החלון שסופק. לאחר מכן, עבודת הסטרימינג מעבדת ומבצעת הנדסת תכונות במיקרו-אצווה זו לפני חלוקתה וכתיבתה לתכונות הקידומת בתוך דלי S3.

מכיוון שתכונות חדשות שנוצרו על ידי משימת הסטרימינג של AWS Glue נכתבות לדלי S3, פונקציית Lambda (invoke-endpoint-asynch) מופעלת, אשר מפעילה נקודת קצה אסינכרונית של SageMaker על ידי שליחת בקשת הזמנה לקבל מסקנות ממודל ה-ML הפרוס שלנו. נקודת הסיום הא-סינכרונית מעמידה בתור את הבקשה להפעלה אסינכרונית. לאחר השלמת העיבוד, SageMaker מאחסן את תוצאות ההסקה במיקום אמזון S3 (S3OutputPath) שצוין במהלך תצורת נקודת הקצה האסינכרונית.

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

SageMaker שולחת גם הודעת הצלחה או שגיאה עם Amazon SNS. לדוגמה, אם אתה מגדיר an מנוי לדוא"ל עבור נושאי ההצלחה והשגיאה של SNS (המצוינים בתצורת נקודת הקצה הא-סינכרונית של SageMaker), ניתן לשלוח דוא"ל בכל פעם שמעובדים בקשת הסקה. צילום המסך הבא מציג דוא"ל לדוגמה מנושא ההצלחה של SNS.

עבור יישומים בעולם האמיתי, אתה יכול לשלב התראות SNS עם שירותים אחרים כגון שירות תורים פשוט של אמזון (Amazon SQS) ו-Lambda עבור עיבוד נוסף של ההסקות שנוצרו או אינטגרציה עם יישומים אחרים במורד הזרם, בהתבסס על הדרישות שלך. לדוגמה, עבור מקרה השימוש החזוי שלנו בתחזוקה, אתה יכול להפעיל פונקציית Lambda המבוססת על הודעת SNS כדי לקרוא את ההסקה שנוצרה מ-Amazon S3, להמשיך לעבד אותה (כגון צבירה או סינון), וליזום זרימות עבודה כגון שליחת הזמנות עבודה עבור תיקון ציוד לטכנאים.

לנקות את

כשתסיים לבדוק את המחסנית, מחק את המשאבים (במיוחד זרם הנתונים של Kinesis, עבודת הזרמת דבק ונושאי SNS) כדי למנוע חיובים בלתי צפויים.

הפעל את הקוד הבא כדי למחוק את הערימה שלך:

sam delete nrt-streaming-inference

מחק גם את המשאבים כגון נקודות הקצה של SageMaker על ידי ביצוע סעיף הניקוי במחברת ModelTraining-Evaluation and Deployment.

סיכום

בפוסט הזה, השתמשנו במקרה שימוש תחזוקה חזוי כדי להדגים כיצד להשתמש בשירותים שונים כגון Kinesis, AWS Glue ו- SageMaker כדי לבנות צינור מסקנות כמעט בזמן אמת. אנו ממליצים לך לנסות את הפתרון הזה ולספר לנו מה אתה חושב.

אם יש לך שאלות, שתף אותן בתגובות.

על המחברים

רחול שרמה הוא ארכיטקט פתרונות במעבדת הנתונים של AWS, שעוזר ללקוחות AWS לתכנן ולבנות פתרונות AI/ML. לפני הצטרפותה ל-AWS, Rahul בילה מספר שנים במגזר הפיננסים והביטוח, וסייעה ללקוחות לבנות נתונים ופלטפורמות אנליטיות.

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