دریافت داده‌های CSV (مقادیر جدا شده با کاما) در BigQuery با استفاده از Cloud Data Fusion - مصرف بی‌درنگ

۱. مقدمه

509db33558ae025.png

آخرین به‌روزرسانی: 2020-02-28

این آزمایشگاه کد، الگوی دریافت داده را برای دریافت داده‌های مراقبت‌های بهداشتی با فرمت CSV در BigQuery به صورت بلادرنگ نشان می‌دهد. ما برای این آزمایشگاه از Cloud Data fusion Real time Data pipeline استفاده خواهیم کرد. داده‌های تست مراقبت‌های بهداشتی واقع‌گرایانه تولید شده و در مخزن ذخیره‌سازی ابری گوگل (gs://hcls_testing_data_fhir_10_patients/csv/) برای شما در دسترس قرار گرفته است.

در این آزمایشگاه کد شما یاد خواهید گرفت:

  • نحوه دریافت داده‌های CSV (بارگذاری بلادرنگ) از Pub/Sub به BigQuery با استفاده از Cloud Data Fusion .
  • نحوه ساخت بصری یک خط لوله ادغام داده‌ها در Cloud Data Fusion برای بارگیری، تبدیل و پوشش داده‌های مراقبت‌های بهداشتی به صورت بلادرنگ .

برای اجرای این دمو به چه چیزهایی نیاز دارید؟

  • شما نیاز به دسترسی به یک پروژه GCP دارید.
  • شما باید نقش مالک (Owner) را در پروژه 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 بیابید.

اگر به یک مجموعه داده جدید نیاز دارید، همیشه می‌توانید آن را با استفاده از Synthea™ تولید کنید. سپس، به جای کپی کردن آن از سطل داده در مرحله کپی کردن داده‌های ورودی، آن را در GCS آپلود کنید.

۲. راه‌اندازی پروژه 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

به مجموعه داده‌های مصنوعی دسترسی پیدا کنید.

  1. از آدرس ایمیلی که برای ورود به Cloud Console استفاده می‌کنید، یک ایمیل به hcls-solutions-external+subscribe@google.com ارسال کنید و درخواست عضویت دهید.
  2. شما یک ایمیل با دستورالعمل‌هایی در مورد نحوه تأیید اقدام دریافت خواهید کرد.
  3. برای پیوستن به گروه، از گزینه پاسخ به ایمیل استفاده کنید. روی دکمه کلیک نکنید. 525a0fa752e0acae.png دکمه
  4. پس از دریافت ایمیل تأیید، می‌توانید به مرحله بعدی در codelab بروید.

کپی کردن داده‌های ورودی

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 یا Subtopic و Subscriptions ایجاد کنید.

gcloud init
gcloud pubsub topics create your-topic
gcloud pubsub subscriptions create --topic your-topic your-sub

۳. تنظیمات محیط ادغام داده‌های ابری

برای فعال کردن API ادغام داده‌های ابری و اعطای مجوزهای لازم، این مراحل را دنبال کنید:

فعال کردن APIها

  1. به کتابخانه API کنسول GCP بروید.
  2. از لیست پروژه‌ها، پروژه خود را انتخاب کنید.
  3. در کتابخانه API، API مورد نظر برای فعال‌سازی را انتخاب کنید ( Cloud Data Fusion API، Cloud Pub/Sub API). اگر برای یافتن API به کمک نیاز دارید، از فیلد جستجو و فیلترها استفاده کنید.
  4. در صفحه API، روی فعال کردن (ENABLE) کلیک کنید.

یک نمونه Cloud Data Fusion ایجاد کنید .

  1. در کنسول GCP، شناسه پروژه خود را انتخاب کنید.
  2. از منوی سمت چپ، Data Fusion را انتخاب کنید، سپس روی دکمه CREATE AN INSTANCE در وسط صفحه (اولین ایجاد) کلیک کنید، یا روی دکمه CREATE INSTANCE در منوی بالا (ایجاد اضافی) کلیک کنید.

a828690ff3bf3c46.png

e8ffacaba8e61be5.png

  1. نام نمونه را وارد کنید. Enterprise را انتخاب کنید.

5af91e46917260ff.png

  1. روی دکمه‌ی CREATE کلیک کنید.

مجوزهای نمونه را تنظیم کنید.

پس از ایجاد یک نمونه، از مراحل زیر برای اعطای مجوزهای مربوط به حساب سرویس در پروژه خود استفاده کنید:

  1. با کلیک روی نام نمونه، به صفحه جزئیات نمونه بروید.

76ad691f795e1ab3.png

  1. حساب سرویس را کپی کنید.

6c91836afb72209d.png

  1. به صفحه IAM پروژه خود بروید.
  2. در صفحه مجوزهای IAM، با کلیک بر روی دکمه افزودن، به حساب سرویس، نقش Cloud Data Fusion API Service Agent را اعطا کنید. عبارت "service account" را در فیلد اعضای جدید جایگذاری کنید و Service Management -> Cloud Data Fusion API Server Agent role را انتخاب کنید.

36f03d11c2a4ce0.png

  1. برای افزودن نقش مشترک Pub/Sub، روی + افزودن یک نقش دیگر (یا ویرایش عامل سرویس API ادغام داده‌های ابری) کلیک کنید.

b4bf5500b8cbe5f9.png

  1. روی ذخیره کلیک کنید.

پس از انجام این مراحل، می‌توانید با کلیک روی پیوند مشاهده نمونه در صفحه نمونه‌های Cloud Data Fusion یا صفحه جزئیات یک نمونه، استفاده از Cloud Data Fusion را شروع کنید.

قانون فایروال را تنظیم کنید.

  1. برای بررسی وجود یا عدم وجود قانون default-allow-ssh، به کنسول GCP -> شبکه VPC -> قوانین فایروال بروید.

102adef44bbe3a45.png

  1. اگر اینطور نیست، یک قانون فایروال اضافه کنید که به همه ترافیک 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

استفاده از رابط کاربری: روی ایجاد قانون فایروال کلیک کنید و اطلاعات را پر کنید:

d9c69ac10496b3d9.png

2dc4971594b82a1f.png

۴. گره‌هایی برای خط لوله بسازید

اکنون که محیط Cloud Data Fusion را در GCP داریم، بیایید با استفاده از مراحل زیر، ساخت خطوط لوله داده در Cloud Data Fusion را شروع کنیم:

  1. در پنجره Cloud Data Fusion، روی لینک View Instance در ستون Action کلیک کنید. به صفحه دیگری هدایت خواهید شد. برای باز کردن نمونه Cloud Data Fusion، روی آدرس اینترنتی ارائه شده کلیک کنید. در پنجره خوشامدگویی، می‌توانید روی دکمه "شروع تور" یا "خیر، متشکرم" کلیک کنید.
  2. منوی «همبرگر» را باز کنید، گزینه‌ی Pipeline -> List را انتخاب کنید.

۳۱۷۸۲۰def۹۳۴a۰۰a.png

  1. روی دکمه سبز + در گوشه بالا سمت راست کلیک کنید، سپس گزینه ایجاد خط لوله را انتخاب کنید. یا روی «ایجاد» یک لینک خط لوله کلیک کنید.

۷۱۱۹۷۵bb2c2416d7.png

3ec0a71409657fb8.png

  1. وقتی pipeline studio ظاهر شد، در بالا سمت چپ، از منوی کشویی Data Pipeline - Realtime را انتخاب کنید.

۳۷۲a۸۸۹a۸۱da۵e۶۶.png

  1. در رابط کاربری Data Pipelines، بخش‌های مختلفی را در پنل سمت چپ مانند Filter، Source، Transform، Analytics، Sink، Error Handlers و Alerts مشاهده خواهید کرد که می‌توانید در آن‌ها یک یا چند گره را برای pipeline انتخاب کنید.

c63de071d4580f2f.png

یک گره منبع انتخاب کنید .

  1. در قسمت Source در پنل Plugin در سمت چپ، روی گره Google Cloud PubSub که در رابط کاربری Data Pipelines ظاهر می‌شود، دوبار کلیک کنید.
  2. به گره منبع PubSub اشاره کنید و روی Properties کلیک کنید.

ed857a5134148d7b.png

  1. فیلدهای مورد نیاز را پر کنید. فیلدهای زیر را تنظیم کنید:
  • برچسب = {هر متنی}
  • نام مرجع = {هر متنی}
  • شناسه پروژه = تشخیص خودکار
  • Subscription = اشتراکی که در بخش ایجاد Pub/Sub Topic ایجاد شده است (برای مثال، your-sub )
  • موضوع = موضوع ایجاد شده در بخش ایجاد میخانه/زیرموضوع (برای مثال، موضوع شما )
  1. برای توضیحات بیشتر روی مستندات کلیک کنید. برای تأیید تمام اطلاعات ورودی، روی دکمه اعتبارسنجی کلیک کنید. رنگ سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.

5c2774338b66bebe.png

  1. برای بستن ویژگی‌های Pub/Sub، روی دکمه X کلیک کنید.

گره Transform را انتخاب کنید .

  1. در قسمت Transform در پنل Plugin در سمت چپ، روی گره Projection که در رابط کاربری Data Pipelines ظاهر می‌شود، دوبار کلیک کنید. گره Pub/Sub source را به گره Transform Projection متصل کنید.
  2. به گره Projection اشاره کنید و روی Properties کلیک کنید.

b3a9a3878879bfd7.png

  1. فیلدهای مورد نیاز را پر کنید. فیلدهای زیر را تنظیم کنید:
  • تبدیل = تبدیل پیام از نوع بایت به نوع رشته.
  • فیلدهایی که باید حذف شوند = {هر فیلدی}
  • فیلدهایی که باید نگه داشته شوند = { پیام ، مهر زمان و ویژگی‌ها } ( برای مثال، ویژگی‌ها: key='filename':value='patients' sent from Pub/Sub)
  • فیلدهایی که باید تغییر نام داده شوند = { message , timestamp }
  1. برای توضیحات بیشتر روی مستندات کلیک کنید. برای تأیید تمام اطلاعات ورودی، روی دکمه اعتبارسنجی کلیک کنید. رنگ سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.

b8c2f8efe18234ff.png

  1. در قسمت Transform در پالت Plugin در سمت چپ، روی گره Wrangler که در رابط کاربری Data Pipelines ظاهر می‌شود، دوبار کلیک کنید. گره تبدیل Projection را به گره تبدیل Wrangler متصل کنید. به گره Wrangler اشاره کنید و روی Properties کلیک کنید.

aa44a4db5fe6623a.png

  1. روی منوی کشویی Actions کلیک کنید و برای وارد کردن یک طرحواره ذخیره شده، Import را انتخاب کنید (برای مثال: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json ).
  2. فیلد TIMESTAMP را در Output Schema (اگر وجود ندارد) با کلیک روی دکمه + در کنار آخرین فیلد اضافه کنید و کادر «Null» را علامت بزنید.
  3. فیلدهای مورد نیاز را پر کنید. فیلدهای زیر را تنظیم کنید:
  • برچسب = {هر متنی}
  • نام فیلد ورودی = {*}
  • پیش‌شرط = { attributes.get("filename") != "patients" } برای تشخیص هر نوع رکورد یا پیام ( به عنوان مثال، بیماران، ارائه دهندگان خدمات، آلرژی‌ها و غیره ) ارسال شده از گره منبع PubSub.
  1. برای توضیحات بیشتر روی مستندات کلیک کنید. برای تأیید تمام اطلاعات ورودی، روی دکمه اعتبارسنجی کلیک کنید. رنگ سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.

3b8e552cd2e3442c.png

  1. نام ستون‌ها را به ترتیب دلخواه تنظیم کنید و فیلدهایی را که نیازی ندارید حذف کنید. قطعه کد زیر را کپی کرده و در کادر دستور پخت قرار دهید.
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####

b93cb9952ca2de73.png

  1. برای پوشش داده‌ها و حذف هویت، به Batch-Codelab - CSV to BigQuery از طریق CDF مراجعه کنید. یا این قطعه کد mask-number SSN xxxxxxx#### را در کادر Recipe اضافه کنید.
  2. برای بستن پنجره Transform Properties، روی دکمه X کلیک کنید.

گره Sink را انتخاب کنید.

  1. در قسمت Sink در پنل Plugin در سمت چپ، روی گره BigQuery که در رابط کاربری Data Pipeline ظاهر می‌شود، دوبار کلیک کنید. گره تبدیل Wrangler را به گره sink BigQuery متصل کنید.
  2. به گره سینک BigQuery اشاره کنید و روی Properties کلیک کنید.

1be711152c92c692.png

  1. فیلدهای مورد نیاز را پر کنید:
  • برچسب = {هر متنی}
  • نام مرجع = {هر متنی}
  • شناسه پروژه = تشخیص خودکار
  • مجموعه داده = مجموعه داده BigQuery که در پروژه فعلی استفاده شده است (برای مثال، DATASET_ID)
  • جدول = {نام جدول}
  1. برای توضیحات بیشتر روی مستندات کلیک کنید. برای تأیید تمام اطلاعات ورودی، روی دکمه اعتبارسنجی کلیک کنید. رنگ سبز "هیچ خطایی یافت نشد" نشان دهنده موفقیت است.

bba71de9f31e842a.png

  1. برای بستن BigQuery Properties، روی دکمه X کلیک کنید.

۵. ساخت خط لوله داده بلادرنگ

در بخش قبلی، گره‌هایی را ایجاد کردیم که برای ساخت یک خط لوله داده در Cloud Data Fusion مورد نیاز بودند. در این بخش، گره‌ها را برای ساخت خط لوله واقعی به هم متصل می‌کنیم.

اتصال همه گره‌ها در یک خط لوله

  1. یک فلش اتصال > را در لبه سمت راست گره منبع بکشید و در لبه سمت چپ گره مقصد رها کنید.
  2. یک خط لوله می‌تواند چندین شاخه داشته باشد که پیام‌های منتشر شده را از یک گره PubSub Source دریافت می‌کنند.

b22908cc35364cdd.png

  1. خط لوله را نام ببرید.

همین. شما اولین خط لوله داده بلادرنگ خود را برای استقرار و اجرا ایجاد کرده‌اید.

ارسال پیام از طریق Cloud Pub/Sub

استفاده از رابط کاربری Pub/Sub :

  1. به کنسول GCP -> Pub/Sub -> Topics بروید، موضوع خود را انتخاب کنید، سپس روی انتشار پیام در منوی بالا کلیک کنید.

d65b2a6af1668ecd.png

  1. فقط یک ردیف رکورد را در هر زمان در فیلد پیام قرار دهید. روی دکمه +ADD AN ATTRIBUTE کلیک کنید. Key = filename , Value = < type of record > ( برای مثال، بیماران، ارائه دهندگان خدمات، آلرژی‌ها و غیره ) را وارد کنید.
  2. برای ارسال پیام، روی دکمه انتشار کلیک کنید.

با استفاده از دستور gcloud :

  1. پیام را به صورت دستی ارائه دهید.
gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"paste one record row here"
  1. با استفاده از دستورات یونیکس cat و sed ، پیام را به صورت نیمه خودکار ارائه دهید. این دستور می‌تواند بارها و بارها با پارامترهای مختلف اجرا شود.
gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"$(gsutil cat gs://$BUCKET_NAME/csv/<value>.csv | sed -n '#p')"

۶. پیکربندی، استقرار و اجرای پایپ‌لاین

اکنون که خط لوله داده را توسعه داده‌ایم، می‌توانیم آن را در Cloud Data Fusion مستقر و اجرا کنیم.

۱bb5b0b8e2953ffa.png

  1. تنظیمات پیش‌فرض Configure را حفظ کنید.
  2. برای پیش‌نمایش داده‌ها، روی پیش‌نمایش کلیک کنید**.** برای بازگشت به پنجره قبلی، دوباره روی **پیش‌نمایش** کلیک کنید. همچنین می‌توانید با کلیک روی **اجرا**، خط لوله را در حالت پیش‌نمایش اجرا کنید.

b3c891e5e1aa20ae.png

  1. برای مشاهده گزارش‌ها، روی گزارش‌ها کلیک کنید.
  2. برای ذخیره همه تغییرات، روی ذخیره کلیک کنید.
  3. برای وارد کردن پیکربندی خط لوله ذخیره شده هنگام ساخت خط لوله جدید، روی «وارد کردن» کلیک کنید.
  4. برای خروجی گرفتن از پیکربندی خط لوله، روی Export کلیک کنید.
  5. برای استقرار خط لوله، روی «استقرار» کلیک کنید.
  6. پس از استقرار، روی Run کلیک کنید و منتظر بمانید تا خط لوله به طور کامل اجرا شود.

f01ba6b746ba53a.png

  1. برای متوقف کردن اجرای خط لوله در هر زمان، روی توقف کلیک کنید.
  2. می‌توانید با انتخاب گزینه‌ی «تکثیر» در زیر دکمه‌ی «اقدامات» ، خط لوله را کپی کنید.
  3. شما می‌توانید با انتخاب گزینه‌ی «صادرات» در زیر دکمه‌ی «اقدامات» ، پیکربندی خط لوله را صادر کنید.

28ea4fc79445fad2.png

  1. برای نمایش نمودارهای تاریخچه اجرا، رکوردها، گزارش‌های خطا و هشدارها، روی خلاصه کلیک کنید.

۷. اعتبارسنجی

در این بخش، اجرای خط لوله داده را اعتبارسنجی می‌کنیم.

  1. تأیید کنید که خط لوله با موفقیت اجرا شده و به طور مداوم در حال اجرا است.

۱۶۴۴dfac4a2d819d.png

  1. اعتبارسنجی کنید که جداول BigQuery با رکوردهای به‌روزرسانی‌شده بر اساس TIMESTAMP بارگذاری شوند. در این مثال، دو رکورد یا پیام بیمار و یک رکورد یا پیام آلرژی در تاریخ 2019-06-25 در موضوع Pub/Sub منتشر شده‌اند.
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 |
+----------+-----------+
  1. تأیید کنید که پیام‌های منتشر شده در <your-topic> توسط مشترک <your-sub> دریافت شده‌اند.
gcloud pubsub subscriptions pull --auto-ack <your-sub>

4cae99a9e4f2ec9f.png

مشاهده نتایج

برای مشاهده نتایج پس از انتشار پیام‌ها در تاپیک Pub/Sub در حالی که خط لوله Realtime در حال اجرا است:

  1. جدول را در رابط کاربری BigQuery جستجو کنید. به رابط کاربری BigQuery بروید
  2. کوئری زیر را با نام پروژه، مجموعه داده و جدول خودتان به‌روزرسانی کنید.

6a1fb85bd868abc9.png

۸. تمیز کردن

برای جلوگیری از تحمیل هزینه به حساب پلتفرم گوگل کلود خود برای منابع استفاده شده در این آموزش:

بعد از اتمام آموزش، می‌توانید منابعی را که در GCP ایجاد کرده‌اید، پاک کنید تا سهمیه‌ای را اشغال نکنند و در آینده برای آنها هزینه‌ای از شما دریافت نشود. بخش‌های بعدی نحوه حذف یا غیرفعال کردن این منابع را شرح می‌دهند.

حذف مجموعه داده BigQuery

برای حذف مجموعه داده BigQuery که به عنوان بخشی از این آموزش ایجاد کرده‌اید، این دستورالعمل‌ها را دنبال کنید.

حذف سطل GCS

برای حذف سطل GCS که به عنوان بخشی از این آموزش ایجاد کرده‌اید، این دستورالعمل‌ها را دنبال کنید.

حذف نمونه Cloud Data Fusion

برای حذف نمونه Cloud Data Fusion خود، این دستورالعمل‌ها را دنبال کنید.

حذف پروژه

ساده‌ترین راه برای حذف هزینه‌ها، حذف پروژه‌ای است که برای آموزش ایجاد کرده‌اید.

برای حذف پروژه:

  1. در کنسول GCP، به صفحه پروژه‌ها بروید. به صفحه پروژه‌ها بروید
  2. در لیست پروژه‌ها، پروژه‌ای را که می‌خواهید حذف کنید انتخاب کنید و روی «حذف» کلیک کنید.
  3. در کادر محاوره‌ای، شناسه پروژه را تایپ کنید و سپس برای حذف پروژه، روی خاموش کردن کلیک کنید.

۹. تبریک

تبریک می‌گویم، شما با موفقیت آزمایشگاه کد برای دریافت داده‌های مراقبت‌های بهداشتی در BigQuery با استفاده از Cloud Data Fusion را به پایان رساندید.

شما داده‌های CSV را در Pub/Sub topic منتشر کردید و سپس آنها را در BigQuery بارگذاری کردید.

شما به صورت بصری یک خط لوله یکپارچه‌سازی داده‌ها برای بارگیری، تبدیل و پوشش داده‌های مراقبت‌های بهداشتی به صورت بلادرنگ ایجاد کردید.

اکنون مراحل کلیدی مورد نیاز برای شروع سفر تجزیه و تحلیل داده‌های مراقبت‌های بهداشتی خود با BigQuery در پلتفرم Google Cloud را می‌دانید.