1. مقدمه
آخرین به روز رسانی: 2020-02-28
این آزمایشگاه کد یک الگوی دریافت داده را برای ورود داده های مراقبت های بهداشتی با فرمت CSV به BigQuery در زمان واقعی نشان می دهد. ما از Cloud Data fusion Real Time Pipeline برای این آزمایشگاه استفاده خواهیم کرد. دادههای واقعی آزمایش مراقبتهای بهداشتی در سطل فضای ذخیرهسازی Google Cloud (gs://hcls_testing_data_fhir_10_patients/csv/) برای شما ایجاد و در دسترس قرار گرفته است.
در این آزمایشگاه کد یاد خواهید گرفت:
- نحوه دریافت دادههای CSV (بارگیری در زمان واقعی) از Pub/Sub به BigQuery با استفاده از Cloud Data Fusion .
- نحوه ایجاد بصری خط لوله یکپارچه سازی داده در Cloud Data Fusion برای بارگیری، تبدیل و پوشاندن داده های مراقبت های بهداشتی در زمان واقعی .
برای اجرای این دمو به چه چیزی نیاز دارید؟
- شما نیاز به دسترسی به یک پروژه GCP دارید.
- شما باید نقش مالک را به پروژه GCP اختصاص دهید.
- داده های مراقبت های بهداشتی در قالب CSV، از جمله هدر.
اگر پروژه GCP ندارید، این مراحل را برای ایجاد یک پروژه GCP جدید دنبال کنید.
داده های مراقبت های بهداشتی در قالب CSV از قبل در سطل GCS در gs://hcls_testing_data_fhir_10_patients/csv/ بارگذاری شده است. هر فایل منبع CSV یک ساختار طرحواره منحصر به فرد دارد. برای مثال Patients.csv طرحی متفاوت از Providers.csv دارد. فایل های طرحواره از پیش بارگذاری شده را می توانید در gs://hcls_testing_data_fhir_10_patients/csv_schemas پیدا کنید.
اگر به یک مجموعه داده جدید نیاز دارید، همیشه می توانید آن را با استفاده از SyntheaTM ایجاد کنید. سپس، به جای کپی کردن آن از سطل در مرحله Copy input data، آن را در GCS آپلود کنید.
2. راه اندازی پروژه GCP
متغیرهای پوسته را برای محیط خود راه اندازی کنید.
برای یافتن PROJECT_ID ، به شناسایی پروژه ها مراجعه کنید.
<!-- CODELAB: Initialize shell variables -> <!-- Your current GCP Project ID -> export PROJECT_ID=<PROJECT_ID> <!-- A new GCS Bucket in your current Project - INPUT -> export BUCKET_NAME=<BUCKET_NAME> <!-- A new BQ Dataset ID - OUTPUT -> export DATASET_ID=<DATASET_ID>
با استفاده از ابزار gsutil یک سطل GCS برای ذخیره داده های ورودی و گزارش خطا ایجاد کنید .
gsutil mb -l us gs://$BUCKET_NAME
به مجموعه داده مصنوعی دسترسی پیدا کنید.
- از آدرس ایمیلی که برای ورود به کنسول Cloud استفاده میکنید، یک ایمیل به hcls-solutions-external+subscribe@google.com ارسال کنید تا بپیوندید.
- شما یک ایمیل حاوی دستورالعمل هایی در مورد نحوه تایید عمل دریافت خواهید کرد.
- برای عضویت در گروه از گزینه پاسخ به ایمیل استفاده کنید. روی آن کلیک نکنید دکمه
- پس از دریافت ایمیل تأیید، میتوانید به مرحله بعدی در لبه کد بروید.
کپی کردن داده های ورودی
gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME
یک مجموعه داده BigQuery ایجاد کنید.
bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID
Google Cloud SDK را نصب و مقداردهی اولیه کنید و Pub یا Sub Topic and Subscriptions ایجاد کنید.
gcloud init gcloud pubsub topics create your-topic gcloud pubsub subscriptions create --topic your-topic your-sub
3. Cloud Data Fusion Environment Setup
برای فعال کردن Cloud Data Fusion API و دادن مجوزهای لازم، مراحل زیر را دنبال کنید:
API ها را فعال کنید
- به کتابخانه GCP Console API بروید.
- از لیست پروژه ها، پروژه خود را انتخاب کنید.
- در کتابخانه API، API مورد نظر برای فعال کردن را انتخاب کنید ( Cloud Data Fusion API، Cloud Pub/Sub API). اگر برای یافتن API به کمک نیاز دارید، از قسمت جستجو و فیلترها استفاده کنید.
- در صفحه API، روی ENABLE کلیک کنید.
یک نمونه Cloud Data Fusion ایجاد کنید .
- در کنسول GCP، ProjectID خود را انتخاب کنید.
- Data Fusion را از منوی سمت چپ انتخاب کنید، سپس روی دکمه CREATE AN INSTANCE در وسط صفحه کلیک کنید (اولین ایجاد)، یا روی دکمه CREATE INSTANCE در منوی بالا کلیک کنید (ایجاد اضافی).
- نام نمونه را وارد کنید. Enterprise را انتخاب کنید.
- روی دکمه CREATE کلیک کنید.
مجوزهای نمونه را تنظیم کنید.
پس از ایجاد یک نمونه، از مراحل زیر برای اعطای مجوز به حساب سرویس مرتبط با مجوزهای نمونه در پروژه خود استفاده کنید:
- با کلیک کردن روی نام نمونه، به صفحه جزئیات نمونه بروید.
- حساب سرویس را کپی کنید.
- به صفحه IAM پروژه خود بروید.
- در صفحه مجوزهای IAM، با کلیک بر روی دکمه افزودن ، نقش Cloud Data Fusion API Service Agent را به حساب سرویس بدهید. «حساب سرویس» را در قسمت اعضای جدید جایگذاری کنید و Service Management -> Cloud Data Fusion API Server Agent را انتخاب کنید.
- برای افزودن نقش مشترک Pub/Sub، روی + افزودن نقش دیگری (یا Edit Cloud Data Fusion API Service Agent) کلیک کنید.
- روی ذخیره کلیک کنید.
پس از انجام این مراحل، میتوانید با کلیک کردن روی پیوند View Instance در صفحه نمونههای Cloud Data Fusion یا صفحه جزئیات یک نمونه، استفاده از Cloud Data Fusion را شروع کنید.
قانون فایروال را تنظیم کنید.
- برای بررسی اینکه آیا قانون پیش فرض-allow-ssh وجود دارد یا خیر، به کنسول GCP -> VPC Network -> قوانین فایروال بروید.
- اگر نه، یک قانون فایروال اضافه کنید که به تمام ترافیک SSH ورودی به شبکه پیش فرض اجازه می دهد.
با استفاده از خط فرمان:
gcloud beta compute --project={PROJECT_ID} firewall-rules create default-allow-ssh --direction=INGRESS --priority=1000 --network=default --action=ALLOW --rules=tcp:22 --source-ranges=0.0.0.0/0 --enable-logging
با استفاده از UI: روی ایجاد قانون فایروال کلیک کنید و اطلاعات را پر کنید:
4. گره هایی برای خط لوله بسازید
اکنون که محیط Cloud Data Fusion را در GCP داریم، بیایید با استفاده از مراحل زیر شروع به ساخت خطوط لوله داده در Cloud Data Fusion کنیم:
- در پنجره Cloud Data Fusion، روی پیوند View Instance در ستون Action کلیک کنید. به صفحه دیگری هدایت خواهید شد. برای باز کردن نمونه Cloud Data Fusion، روی URL ارائه شده کلیک کنید. انتخاب شما برای کلیک بر روی دکمه "شروع تور" یا "نه، متشکرم" در پنجره بازشو خوش آمدید.
- منوی "همبرگر" را گسترش دهید، Pipeline -> List را انتخاب کنید
- روی دکمه سبز رنگ + در گوشه سمت راست بالا کلیک کنید و سپس Create Pipeline را انتخاب کنید. یا روی «ایجاد» پیوند خط لوله کلیک کنید.
- هنگامی که استودیوی خط لوله ظاهر شد، در سمت چپ بالا، Data Pipeline - Realtime را از منوی کشویی انتخاب کنید.
- در رابط کاربری Data Pipelines، بخشهای مختلفی را در پانل سمت چپ بهعنوان فیلتر، منبع، تبدیل، تجزیه و تحلیل، سینک، کنترلکننده خطا و هشدارها خواهید دید که میتوانید یک گره یا گرهها را برای خط لوله انتخاب کنید.
یک گره منبع را انتخاب کنید .
- در قسمت Source در پالت Plugin در سمت چپ، روی گره Google Cloud PubSub که در رابط کاربری Data Pipelines ظاهر می شود، دوبار کلیک کنید.
- به گره منبع PubSub اشاره کنید و روی Properties کلیک کنید.
- فیلدهای مورد نیاز را پر کنید. فیلدهای زیر را تنظیم کنید:
- برچسب = {هر متن}
- نام مرجع = {هر متن}
- شناسه پروژه = شناسایی خودکار
- اشتراک = اشتراک ایجاد شده در بخش Create Pub/Sub Topic (به عنوان مثال، your-sub )
- موضوع = موضوع ایجاد شده در بخش Create Pub/Sub Topic (به عنوان مثال، موضوع شما )
- برای توضیح دقیق روی Documentation کلیک کنید. بر روی دکمه Validate کلیک کنید تا تمام اطلاعات ورودی تایید شود. سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.
- برای بستن Pub/Sub Properties، روی دکمه X کلیک کنید.
گره Transform را انتخاب کنید .
- در قسمت Transform در پالت Plugin در سمت چپ، روی گره Projection که در رابط کاربری Data Pipelines ظاهر می شود، دوبار کلیک کنید. گره منبع Pub/Sub را به گره تبدیل Projection متصل کنید.
- به گره Projection اشاره کنید و روی Properties کلیک کنید.
- فیلدهای مورد نیاز را پر کنید. فیلدهای زیر را تنظیم کنید:
- تبدیل = تبدیل پیام از نوع بایت به نوع رشته.
- فیلدهایی که باید حذف شوند = {هر فیلدی}
- فیلدهایی که باید حفظ شوند = { message , timestamp , and features } ( به عنوان مثال، ویژگی ها: key='filename':value='patients' ارسال شده از Pub/Sub)
- فیلدهایی برای تغییر نام = { message , timestamp }
- برای توضیح دقیق روی Documentation کلیک کنید. بر روی دکمه Validate کلیک کنید تا تمام اطلاعات ورودی تایید شود. سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.
- در قسمت Transform در پالت Plugin در سمت چپ، روی گره Wrangler که در رابط کاربری Data Pipelines ظاهر می شود، دوبار کلیک کنید. گره تبدیل Projection را به گره تبدیل Wrangler متصل کنید. به گره Wrangler اشاره کنید و روی Properties کلیک کنید.
- روی گزینه Actions کلیک کنید و برای وارد کردن یک طرح ذخیره شده، Import را انتخاب کنید (به عنوان مثال: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json ).
- فیلد TIMESTAMP را در طرحواره خروجی (اگر وجود ندارد) با کلیک کردن روی دکمه + در کنار آخرین فیلد اضافه کنید و کادر "تهی" را علامت بزنید.
- فیلدهای مورد نیاز را پر کنید. فیلدهای زیر را تنظیم کنید:
- برچسب = {هر متن}
- نام فیلد ورودی = {*}
- پیش شرط = { attributes.get("filename") != "patients" } برای تشخیص هر نوع رکورد یا پیام ( به عنوان مثال، بیماران، ارائه دهندگان، آلرژی ها و غیره ) ارسال شده از گره منبع PubSub.
- برای توضیح دقیق روی Documentation کلیک کنید. بر روی دکمه Validate کلیک کنید تا تمام اطلاعات ورودی تایید شود. سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.
- نام ستون ها را به ترتیب ترجیحی تنظیم کنید و فیلدهایی را که نیاز ندارید رها کنید. قطعه کد زیر را کپی کرده و در کادر Recipe قرار دهید.
drop attributes parse-as-csv :body ',' false drop body set columns TIMESTAMP,Id,BIRTHDATE,DEATHDATE,SSN,DRIVERS,PASSPORT,PREFIX,FIRST,LAST,SUFFIX,MAIDEN,MARITAL,RACE,ETHNICITY,GENDER,BIRTHPLACE,ADDRESS,CITY,STATE,ZIP mask-number SSN xxxxxxx####
- برای پوشاندن داده ها و حذف هویت به Batch-Codelab - CSV به BigQuery از طریق CDF مراجعه کنید. یا این قطعه کد ماسک-شماره SSN xxxxxxx#### را در کادر دستور غذا اضافه کنید
- برای بستن پنجره Transform Properties روی دکمه X کلیک کنید.
گره سینک را انتخاب کنید.
- در قسمت Sink در پالت Plugin در سمت چپ، روی گره BigQuery که در UI Data Pipeline ظاهر می شود، دوبار کلیک کنید. گره تبدیل رانگلر را به گره سینک BigQuery متصل کنید.
- به گره سینک BigQuery اشاره کنید و روی Properties کلیک کنید.
- فیلدهای مورد نیاز را پر کنید:
- برچسب = {هر متن}
- نام مرجع = {هر متن}
- شناسه پروژه = شناسایی خودکار
- مجموعه داده = مجموعه داده BigQuery مورد استفاده در پروژه فعلی (به عنوان مثال، DATASET_ID)
- جدول = {نام جدول}
- برای توضیح دقیق روی Documentation کلیک کنید. بر روی دکمه Validate کلیک کنید تا تمام اطلاعات ورودی تایید شود. سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.
- برای بستن BigQuery Properties، روی دکمه X کلیک کنید.
5. خط لوله داده در زمان واقعی بسازید
در بخش قبل گره هایی ایجاد کردیم که برای ساخت خط لوله داده در Cloud Data Fusion مورد نیاز هستند. در این بخش، گره ها را برای ساختن خط لوله واقعی به هم وصل می کنیم.
اتصال تمام گره ها در یک خط لوله
- یک فلش اتصال > روی لبه سمت راست گره مبدا را بکشید و روی لبه سمت چپ گره مقصد رها کنید.
- یک خط لوله می تواند چندین شاخه داشته باشد که پیام های منتشر شده را از همان گره PubSub Source دریافت می کند.
- خط لوله را نام ببرید.
همین است. شما به تازگی اولین خط لوله داده بلادرنگ خود را برای استقرار و اجرا ایجاد کرده اید.
ارسال پیام از طریق Cloud Pub/Sub
با استفاده از Pub/Sub UI :
- به کنسول GCP -> Pub/Sub -> Topics بروید، موضوع خود را انتخاب کنید، سپس روی PUBLISH MESSAGE در منوی بالا کلیک کنید.
- هر بار فقط یک ردیف رکورد را در قسمت پیام قرار دهید. روی دکمه +ADD AN ATTRIBUTE کلیک کنید. کلید = نام فایل ، مقدار = < نوع رکورد > ( به عنوان مثال، بیماران، ارائه دهندگان، آلرژی ها و غیره ) را ارائه دهید.
- برای ارسال پیام روی دکمه انتشار کلیک کنید.
با استفاده از دستور gcloud :
- پیام را به صورت دستی ارائه دهید.
gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \ "paste one record row here"
- به صورت نیمه خودکار پیام را با استفاده از دستورات cat و sed unix ارائه دهید. این دستور می تواند بارها و بارها با پارامترهای مختلف اجرا شود.
gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \ "$(gsutil cat gs://$BUCKET_NAME/csv/<value>.csv | sed -n '#p')"
6. Pipeline را پیکربندی، استقرار و اجرا کنید
اکنون که خط لوله داده را توسعه داده ایم، می توانیم آن را در Cloud Data Fusion اجرا و اجرا کنیم.
- تنظیمات پیش فرض را حفظ کنید.
- برای پیشنمایش دادهها روی پیشنمایش کلیک کنید**.** دوباره روی **پیشنمایش** کلیک کنید تا به پنجره قبلی برگردید. همچنین می توانید خط لوله را در حالت پیش نمایش با کلیک روی **RUN** اجرا کنید.
- برای مشاهده گزارشها روی Logs کلیک کنید.
- برای ذخیره همه تغییرات روی ذخیره کلیک کنید.
- برای وارد کردن پیکربندی ذخیره شده خط لوله هنگام ساخت خط لوله جدید، روی Import کلیک کنید.
- برای صادر کردن پیکربندی خط لوله، روی صادرات کلیک کنید.
- برای استقرار خط لوله روی Deploy کلیک کنید.
- پس از استقرار، روی Run کلیک کنید و منتظر بمانید تا خط لوله کامل شود.
- برای توقف اجرای خط لوله در هر زمان، روی Stop کلیک کنید.
- با انتخاب Duplicate در زیر دکمه Actions می توانید خط لوله را کپی کنید.
- میتوانید پیکربندی خط لوله را با انتخاب Export در دکمه Actions صادر کنید.
- برای نمایش نمودارهای تاریخچه اجرا، سوابق، گزارش های خطا و هشدارها روی خلاصه کلیک کنید.
7. اعتبار سنجی
در این بخش، اجرای خط لوله داده را تایید می کنیم.
- تأیید کنید که خط لوله با موفقیت اجرا شده و به طور مداوم در حال اجرا است.
- تأیید کنید که جداول BigQuery با رکوردهای به روز شده بر اساس TIMESTAMP بارگیری می شوند. در این مثال، دو پرونده یا پیام بیمار و یک پرونده یا پیام آلرژی در موضوع Pub/Sub در تاریخ 25/06/2019 منتشر شد.
bq query --nouse_legacy_sql 'select (select count(*) from \ '$PROJECT_ID.$DATASET_ID.Patients' where TIMESTAMP > "2019-06-25 \ 01:29:00.0000 UTC" ) as Patients, (select count(*) from \ '$PROJECT_ID.$DATASET_ID.Allergies' where TIMESTAMP > "2019-06-25 \ 01:29:00.0000 UTC") as Allergies;'
Waiting on bqjob_r14c8b94c1c0fe06a_0000016b960df4e1_1 ... (0s) Current status: DONE
+----------+-----------+
| Patients | Allergies |
+----------+-----------+
| 2 | 1 |
+----------+-----------+
- تأیید کنید که پیام های منتشر شده در <your-topic> توسط مشترک <your-sub> دریافت شده است.
gcloud pubsub subscriptions pull --auto-ack <your-sub>
مشاهده نتایج
برای مشاهده نتایج پس از انتشار پیامها در موضوع Pub/Sub در حالی که خط لوله Realtime در حال اجرا است:
- جدول را در رابط کاربری BigQuery جستجو کنید. به UI BIGQUERY بروید
- پرس و جو زیر را به نام پروژه، مجموعه داده و جدول خود به روز کنید.
8. تمیز کردن
برای جلوگیری از تحمیل هزینه به حساب Google Cloud Platform برای منابع استفاده شده در این آموزش:
پس از پایان آموزش، میتوانید منابعی را که در GCP ایجاد کردهاید پاکسازی کنید تا سهمیه را اشغال نکنند و در آینده برای آنها صورتحساب دریافت نخواهید کرد. بخشهای زیر نحوه حذف یا خاموش کردن این منابع را توضیح میدهند.
حذف مجموعه داده BigQuery
برای حذف مجموعه داده BigQuery که به عنوان بخشی از این آموزش ایجاد کرده اید، این دستورالعمل ها را دنبال کنید.
حذف سطل GCS
برای حذف سطل GCS که به عنوان بخشی از این آموزش ایجاد کرده اید، این دستورالعمل ها را دنبال کنید.
حذف نمونه Cloud Data Fusion
این دستورالعمل ها را دنبال کنید تا نمونه Cloud Data Fusion خود را حذف کنید .
حذف پروژه
ساده ترین راه برای حذف صورتحساب، حذف پروژه ای است که برای آموزش ایجاد کرده اید.
برای حذف پروژه:
- در کنسول GCP، به صفحه پروژه ها بروید. به صفحه پروژه ها بروید
- در لیست پروژه، پروژه ای را که می خواهید حذف کنید انتخاب کنید و روی Delete کلیک کنید.
- در محاوره، شناسه پروژه را تایپ کنید و سپس روی Shut down کلیک کنید تا پروژه حذف شود.
9. تبریک می گویم
تبریک میگوییم، شما با موفقیت آزمایشگاه کد را برای دریافت دادههای مراقبتهای بهداشتی در BigQuery با استفاده از Cloud Data Fusion تکمیل کردید.
دادههای CSV را در موضوع Pub/Sub منتشر کردید و سپس در BigQuery بارگیری کردید.
شما به صورت بصری یک خط لوله ادغام داده برای بارگیری، تبدیل و پوشاندن داده های مراقبت های بهداشتی در زمان واقعی ایجاد کردید.
اکنون مراحل کلیدی مورد نیاز برای شروع سفر تجزیه و تحلیل دادههای بهداشت و درمان خود را با BigQuery در Google Cloud Platform میدانید.