نقل بيانات ملف CSV (القيم المفصولة بفواصل) إلى BigQuery باستخدام Cloud Data Fusion - العرض في الوقت الفعلي

1. مقدمة

509db33558ae025.png

تاريخ آخر تعديل: 2020-02-28

يوضّح هذا الدرس التطبيقي حول الترميز نموذجًا لنقل البيانات من أجل نقل بيانات الرعاية الصحية بتنسيق CSV إلى BigQuery في الوقت الفعلي. سنستخدم مسار بيانات في الوقت الفعلي في Cloud Data Fusion لهذا التمرين العملي. تم إنشاء بيانات اختبار واقعية في مجال الرعاية الصحية وإتاحتها لك في حزمة Google Cloud Storage ‏ (gs://hcls_testing_data_fhir_10_patients/csv/).

في هذا الدرس العملي، ستتعرّف على ما يلي:

  • كيفية نقل بيانات CSV (التحميل في الوقت الفعلي) من Pub/Sub إلى BigQuery باستخدام Cloud Data Fusion
  • كيفية إنشاء مسار تكامل بيانات بشكل مرئي في Cloud Data Fusion من أجل تحميل بيانات الرعاية الصحية وتحويلها وإخفاء هويتها في الوقت الفعلي

ما هي المتطلبات اللازمة لتشغيل هذا الإصدار التجريبي؟

  • يجب أن يكون لديك إذن الوصول إلى مشروع على Google Cloud Platform.
  • يجب أن يكون لديك دور "مالك" في مشروع Google Cloud Platform.
  • بيانات الرعاية الصحية بتنسيق CSV، بما في ذلك العنوان

إذا لم يكن لديك مشروع على Google Cloud Platform، اتّبِع هذه الخطوات لإنشاء مشروع جديد على Google Cloud Platform.

تم تحميل بيانات الرعاية الصحية بتنسيق 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. بعد ذلك، حمِّله إلى GCS بدلاً من نسخه من الحزمة في خطوة "نسخ بيانات الإدخال".

2. إعداد مشروع Google Cloud Platform

تهيئة متغيرات shell لبيئتك

للعثور على 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>

أنشئ حزمة GCS لتخزين بيانات الإدخال وسجلات الأخطاء باستخدام أداة gsutil.

gsutil mb -l us gs://$BUCKET_NAME

الحصول على إذن الوصول إلى مجموعة البيانات الاصطناعية

  1. من عنوان البريد الإلكتروني الذي تستخدمه لتسجيل الدخول إلى Cloud Console، أرسِل رسالة إلكترونية إلى hcls-solutions-external+subscribe@google.com تطلب فيها الانضمام.
  2. ستتلقّى رسالة إلكترونية تتضمّن تعليمات حول كيفية تأكيد الإجراء.
  3. استخدِم الخيار للردّ على الرسالة الإلكترونية للانضمام إلى المجموعة. لا تنقر على الزر 525a0fa752e0acae.png.
  4. بعد تلقّي رسالة التأكيد الإلكترونية، يمكنك الانتقال إلى الخطوة التالية في الدرس العملي.

نسخ بيانات الإدخال:

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

ثبِّت حزمة تطوير البرامج (SDK) من Google Cloud وابدأ إعدادها وأنشئ موضوعًا واشتراكات في Pub أو Sub.

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

3- إعداد بيئة Cloud Data Fusion

اتّبِع الخطوات التالية من أجل تفعيل Cloud Data Fusion API ومنح الأذونات المطلوبة:

فعِّل واجهات برمجة التطبيقات.

  1. انتقِل إلى مكتبة واجهات برمجة التطبيقات في "وحدة تحكّم Google Cloud Platform".
  2. من قائمة المشاريع، اختَر مشروعك.
  3. في "مكتبة واجهات برمجة التطبيقات"، اختَر واجهة برمجة التطبيقات التي تريد تفعيلها ( Cloud Data Fusion API أو Cloud Pub/Sub API). إذا كنت بحاجة إلى مساعدة في العثور على واجهة برمجة التطبيقات، استخدِم حقل البحث والفلاتر.
  4. في صفحة واجهة برمجة التطبيقات، انقر على تفعيل.

إنشاء مثيل Cloud Data Fusion

  1. في "وحدة تحكّم Google Cloud"، اختَر ProjectID.
  2. اختَر "دمج البيانات" من القائمة اليمنى، ثمّ انقر على الزرّ "إنشاء مثيل" في منتصف الصفحة (الإنشاء الأول)، أو انقر على الزرّ "إنشاء مثيل" في القائمة العلوية (الإنشاء الإضافي).

a828690ff3bf3c46.png

e8ffacaba8e61be5.png

  1. أدخِل اسم المثيل. اختَر المؤسسة.

5af91e46917260ff.png

  1. انقر على الزر CREATE (إنشاء).

إعداد أذونات الجهاز الظاهري

بعد إنشاء آلة افتراضية، اتّبِع الخطوات التالية لمنح حساب الخدمة المرتبط بالآلة الافتراضية أذونات في مشروعك:

  1. انتقِل إلى صفحة تفاصيل الجهاز الظاهري من خلال النقر على اسم الجهاز الظاهري.

76ad691f795e1ab3.png

  1. انسخ حساب الخدمة.

6c91836afb72209d.png

  1. انتقِل إلى صفحة "إدارة الهوية وإمكانية الوصول" في مشروعك.
  2. في صفحة أذونات IAM، امنح حساب الخدمة دور وكيل خدمة Cloud Data Fusion API من خلال النقر على الزر إضافة. الصِق "حساب الخدمة" في حقل "أعضاء جدد" (New members) واختَر "إدارة الخدمة" (Service Management) -> دور "وكيل خادم واجهة برمجة التطبيقات Cloud Data Fusion" (Cloud Data Fusion API Server Agent role).

36f03d11c2a4ce0.png

  1. انقر على + إضافة دور آخر (أو تعديل "وكيل خدمة واجهة برمجة التطبيقات في Cloud Data Fusion") لإضافة دور "مشترك في Pub/Sub".

b4bf5500b8cbe5f9.png

  1. انقر على حفظ.

بعد إكمال هذه الخطوات، يمكنك البدء في استخدام Cloud Data Fusion من خلال النقر على الرابط عرض المثيل في صفحة مثيلات Cloud Data Fusion أو صفحة تفاصيل أحد المثيلات.

إعداد قاعدة جدار الحماية:

  1. انتقِل إلى "وحدة تحكّم Google Cloud" -> "شبكة السحابة الافتراضية الخاصة" -> "قواعد جدار الحماية" للتحقّق مما إذا كانت قاعدة default-allow-ssh موجودة أم لا.

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

باستخدام واجهة المستخدم: انقر على "إنشاء قاعدة جدار الحماية" (Create Firewall Rule) واملأ المعلومات:

d9c69ac10496b3d9.png

2dc4971594b82a1f.png

4. إنشاء عُقد للمسار

بعد أن أصبح لدينا بيئة Cloud Data Fusion في GCP، لنبدأ بإنشاء خطوط نقل البيانات في Cloud Data Fusion باتّباع الخطوات التالية:

  1. في نافذة Cloud Data Fusion، انقر على الرابط "عرض المثيل" في عمود "الإجراء". ستتم إعادة توجيهك إلى صفحة أخرى. انقر على عنوان URL المقدَّم لفتح مثيل Cloud Data Fusion. اختيارك النقر على زر "بدء الجولة" أو "لا، شكرًا" في النافذة المنبثقة "مرحبًا"
  2. وسِّع قائمة "النقاط الثلاث"، ثم انقر على "خطوات التنفيذ" -> "القائمة".

317820def934a00a.png

  1. انقر على الزر الأخضر + في أعلى يسار الصفحة، ثم اختَر إنشاء مسار معالجة. يمكنك أيضًا النقر على "إنشاء" رابط مسار.

711975bb2c2416d7.png

3ec0a71409657fb8.png

  1. بعد ظهور "استوديو خطوط نقل البيانات"، انقر على خط نقل البيانات - في الوقت الفعلي من القائمة المنسدلة في أعلى يمين الصفحة.

372a889a81da5e66.png

  1. في واجهة مستخدم "خطوط نقل البيانات"، ستظهر أقسام مختلفة في اللوحة اليمنى، مثل "الفلتر" و"المصدر" و"التحويل" و"الإحصاءات" و"المستودع" و"معالجات الأخطاء" و"التنبيهات"، حيث يمكنك اختيار عقدة أو عقد متعددة لخط النقل.

c63de071d4580f2f.png

اختَر عقدة المصدر.

  1. ضمن قسم "المصدر" (Source) في لوحة "المكوّن الإضافي" (Plugin) على يمين الصفحة، انقر نقرًا مزدوجًا على عقدة Google Cloud PubSub التي تظهر في واجهة مستخدم "مسارات البيانات" (Data Pipelines).
  2. أشِر إلى عقدة مصدر PubSub وانقر على الخصائص.

ed857a5134148d7b.png

  1. املأ الحقول المطلوبة. اضبط الحقول التالية:
  • Label = {any text}
  • اسم المرجع = {أي نص}
  • رقم تعريف المشروع = اكتشاف تلقائي
  • الاشتراك = الاشتراك الذي تم إنشاؤه في قسم "إنشاء موضوع Pub/Sub" (على سبيل المثال، your-sub)
  • الموضوع = الموضوع الذي تم إنشاؤه في قسم "إنشاء موضوع Pub/Sub" (على سبيل المثال، your-topic)
  1. انقر على المستندات للحصول على شرح مفصّل. انقر على الزر "التحقّق" للتحقّق من صحة جميع المعلومات المُدخَلة. يشير اللون الأخضر "لم يتم العثور على أي أخطاء" إلى النجاح.

5c2774338b66bebe.png

  1. لإغلاق "خصائص Pub/Sub"، انقر على الزر X.

اختَر عُقدة التحويل.

  1. ضمن قسم "التحويل" (Transform) في لوحة "المكوّن الإضافي" (Plugin) على اليمين، انقر نقرًا مزدوجًا على عقدة الإسقاط (Projection) التي تظهر في واجهة مستخدم "خطوط نقل البيانات" (Data Pipelines). ربط عُقدة مصدر Pub/Sub بعُقدة تحويل العرض
  2. أشِر إلى عقدة العرض وانقر على الخصائص.

b3a9a3878879bfd7.png

  1. املأ الحقول المطلوبة. اضبط الحقول التالية:
  • Convert = تحويل message من نوع بايت إلى نوع سلسلة
  • الحقول التي سيتم تجاهلها = {أي حقل}
  • الحقول المطلوب الاحتفاظ بها = {message وtimestamp وattributes} (على سبيل المثال، السمات: key=‎'filename':value=‎'patients' المُرسَلة من Pub/Sub)
  • الحقول المطلوب إعادة تسميتها = {message, timestamp}
  1. انقر على المستندات للحصول على شرح مفصّل. انقر على الزر "التحقّق" للتحقّق من صحة جميع المعلومات المُدخَلة. يشير اللون الأخضر "لم يتم العثور على أي أخطاء" إلى النجاح.

b8c2f8efe18234ff.png

  1. ضمن قسم "التحويل" (Transform) في لوحة "المكوّنات الإضافية" (Plugin) على يمين الصفحة، انقر مرّتين على عقدة Wrangler التي تظهر في واجهة مستخدم "خطوط نقل البيانات" (Data Pipelines). وصِّل عقدة تحويل Projection بعقدة تحويل Wrangler. أشِر إلى عقدة Wrangler وانقر على الخصائص.

aa44a4db5fe6623a.png

  1. انقر على القائمة المنسدلة الإجراءات واختَر استيراد لاستيراد مخطط محفوظ (على سبيل المثال: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json).
  2. أضِف حقل TIMESTAMP في "مخطط الإخراج" (Output Schema) (إذا لم يكن متوفّرًا) من خلال النقر على الزر + بجانب الحقل الأخير، وضَع علامة في المربّع "قيمة فارغة" (Null).
  3. املأ الحقول المطلوبة. اضبط الحقول التالية:
  • Label = {any text}
  • اسم حقل الإدخال = {*}
  • Precondition = {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 via CDF بشأن إخفاء البيانات وإخفاء معلومات تحديد الهوية. أو أضِف مقتطف الرمز mask-number SSN xxxxxxx#### في مربّع الوصفة
  2. لإغلاق نافذة "خصائص التحويل"، انقر على الزر X.

اختَر عقدة Sink.

  1. ضمن قسم "المستودع" (Sink) في لوحة "المكوّن الإضافي" (Plugin) على يمين الصفحة، انقر نقرًا مزدوجًا على عقدة BigQuery التي تظهر في واجهة مستخدم "مسار نقل البيانات" (Data Pipeline). اربط عقدة تحويل Wrangler بعقدة مخزَن BigQuery.
  2. أشِر إلى عقدة مخزَن BigQuery وانقر على "السمات".

1be711152c92c692.png

  1. املأ الحقول المطلوبة:
  • Label = {any text}
  • اسم المرجع = {أي نص}
  • رقم تعريف المشروع = اكتشاف تلقائي
  • مجموعة البيانات = مجموعة بيانات BigQuery المستخدَمة في المشروع الحالي (على سبيل المثال، DATASET_ID)
  • الجدول = {اسم الجدول}
  1. انقر على المستندات للحصول على شرح مفصّل. انقر على الزر "التحقّق" للتحقّق من صحة جميع المعلومات المُدخَلة. يشير اللون الأخضر "لم يتم العثور على أي أخطاء" إلى النجاح.

bba71de9f31e842a.png

  1. لإغلاق "خصائص BigQuery"، انقر على الزر X.

5- إنشاء مسار بيانات في الوقت الفعلي

في القسم السابق، أنشأنا العُقد المطلوبة لإنشاء مسار نقل بيانات في Cloud Data Fusion. في هذا القسم، نربط العُقد لإنشاء مسار البيانات الفعلي.

ربط جميع العُقد في مسار

  1. اسحب سهم اتصال > على الحافة اليسرى لعقدة المصدر وأفلِته على الحافة اليمنى لعقدة الوجهة.
  2. يمكن أن تتضمّن سلسلة المعالجة فروعًا متعدّدة تتلقّى الرسائل المنشورة من عقدة مصدر PubSub نفسها.

b22908cc35364cdd.png

  1. أدخِل اسمًا لعملية النقل.

انتهيت! لقد أنشأت للتو أول مسار لتدفّق البيانات في الوقت الفعلي ليتم نشره وتشغيله.

إرسال الرسائل من خلال Cloud Pub/Sub

استخدام واجهة مستخدم Pub/Sub:

  1. انتقِل إلى وحدة تحكّم GCP -> النشر/الاشتراك -> المواضيع، واختَر your-topic، ثم انقر على "نشر رسالة" (PUBLISH MESSAGE) في القائمة العلوية.

d65b2a6af1668ecd.png

  1. ضَع صف سجلّ واحدًا فقط في كل مرة في حقل "الرسالة". انقر على الزر +إضافة سمة. أدخِل المفتاح = اسم الملف، والقيمة = <نوع السجلّ> (على سبيل المثال، المرضى، ومقدّمو الرعاية الصحية، والحساسية، وما إلى ذلك).
  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')"

6. ضبط مسار العرض والنشر والتشغيل

بعد أن أنشأنا مسار نقل البيانات، يمكننا نشره وتشغيله في Cloud Data Fusion.

1bb5b0b8e2953ffa.png

  1. احتفِظ بالإعدادات التلقائية في ضبط.
  2. انقر على معاينة لمعاينة البيانات**.** انقر على **معاينة** مرة أخرى للرجوع إلى النافذة السابقة. يمكنك أيضًا تشغيل خط الأنابيب في "وضع المعاينة" من خلال النقر على **تشغيل**.

b3c891e5e1aa20ae.png

  1. انقر على السجلات لعرض السجلات.
  2. انقر على حفظ لحفظ جميع التغييرات.
  3. انقر على استيراد لاستيراد إعدادات المسار المحفوظة عند إنشاء مسار جديد.
  4. انقر على تصدير لتصدير إعدادات مسار.
  5. انقر على نشر لنشر خط الأنابيب.
  6. بعد نشرها، انقر على تشغيل وانتظِر إلى أن يكتمل تشغيل خط الأنابيب.

f01ba6b746ba53a.png

  1. انقر على إيقاف لإيقاف تنفيذ خط الأنابيب في أي وقت.
  2. يمكنك تكرار خطة التنفيذ من خلال النقر على "تكرار" ضمن زر الإجراءات.
  3. يمكنك تصدير إعدادات خطة المبيعات من خلال النقر على "تصدير" ضمن زر الإجراءات.

28ea4fc79445fad2.png

  1. انقر على الملخّص لعرض رسوم بيانية لسجلّ عمليات التشغيل والسجلات وسجلات الأخطاء والتحذيرات.

7. التحقّق من الصحة

في هذا القسم، نتحقّق من صحة تنفيذ مسار نقل البيانات.

  1. تأكَّد من أنّ عملية التنفيذ تمت بنجاح وأنّ خط الأنابيب يعمل باستمرار.

1644dfac4a2d819d.png

  1. تأكَّد من تحميل جداول BigQuery بسجلات معدَّلة استنادًا إلى الطابع الزمني. في هذا المثال، تم نشر سجلّين أو رسالتين خاصّتَين بمريض وسجلّ أو رسالة واحدة خاصّة بالحساسية في موضوع 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 |
+----------+-----------+
  1. تأكَّد من أنّ المشترك <your-sub> قد تلقّى الرسائل المنشورة في <your-topic>.
gcloud pubsub subscriptions pull --auto-ack <your-sub>

4cae99a9e4f2ec9f.png

عرض النتائج

لعرض النتائج بعد نشر الرسائل في موضوع Pub/Sub أثناء تشغيل مسار البيانات في الوقت الفعلي، اتّبِع الخطوات التالية:

  1. إجراء طلب بحث في الجدول من خلال واجهة مستخدم BigQuery الانتقال إلى واجهة مستخدم BigQuery
  2. عدِّل الاستعلام أدناه ليناسب اسم مشروعك ومجموعة البيانات والجدول.

6a1fb85bd868abc9.png

8. تنظيف

لتجنُّب تحمّل رسوم في حسابك على Google Cloud Platform مقابل الموارد المستخدَمة في هذا البرنامج التعليمي، اتّبِع الخطوات التالية:

بعد الانتهاء من البرنامج التعليمي، يمكنك تنظيف الموارد التي أنشأتها على Google Cloud Platform حتى لا تشغل حصة ولن يتم تحصيل رسوم منك مقابلها في المستقبل. توضّح الأقسام التالية كيفية حذف هذه الموارد أو إيقافها.

حذف مجموعة بيانات BigQuery

اتّبِع هذه التعليمات من أجل حذف مجموعة بيانات BigQuery التي أنشأتها كجزء من هذا البرنامج التعليمي.

حذف حزمة GCS

اتّبِع هذه التعليمات من أجل حذف حزمة GCS التي أنشأتها كجزء من هذا البرنامج التعليمي.

حذف مثيل Cloud Data Fusion

اتّبِع هذه التعليمات لحذف مثيل Cloud Data Fusion.

حذف المشروع

أسهل طريقة لإيقاف الفوترة هي حذف المشروع الذي أنشأته لتنفيذ البرنامج التعليمي.

لحذف المشروع، اتّبِع الخطوات التالية:

  1. في "وحدة تحكّم Google Cloud Platform"، انتقِل إلى صفحة المشاريع. الانتقال إلى صفحة المشاريع
  2. في قائمة المشاريع، اختَر المشروع الذي تريد حذفه وانقر على حذف.
  3. في مربّع الحوار، اكتب رقم تعريف المشروع، ثم انقر على إيقاف لحذف المشروع.

9- تهانينا

تهانينا، لقد أكملت بنجاح تجربة الترميز لنقل بيانات الرعاية الصحية إلى BigQuery باستخدام Cloud Data Fusion.

لقد نشرت بيانات CSV في موضوع Pub/Sub ثم حمّلتها إلى BigQuery.

أنشأت بشكل مرئي مسارًا لدمج البيانات من أجل تحميل بيانات الرعاية الصحية وتحويلها وإخفاء هويتها في الوقت الفعلي.

أصبحت الآن على دراية بالخطوات الأساسية المطلوبة لبدء رحلة تحليل بيانات الرعاية الصحية باستخدام BigQuery على Google Cloud Platform.