عملية نقل البيانات الكاملة: من MySQL مجزّأ محليًا إلى Cloud Spanner (GoogleSQL)

1. قبل البدء

يرشدك هذا الدرس التطبيقي حول الترميز إلى كيفية نقل قاعدة بيانات MySQL مجزّأة محلية إلى قاعدة بيانات Cloud Spanner باستخدام لغة GoogleSQL. ستستخدم خدمات Google Cloud، بما في ذلك "أداة نقل البيانات في Spanner" (SMT) وDataflow وDatastream وPubSub وGoogle Cloud Storage.

ما الذي ستتعلمه:

• ما هي البيئة المقسّمة وكيفية إعدادها؟
• كيفية استخدام واجهة مستخدم الويب الخاصة بأداة نقل البيانات إلى Spanner (SMT) لتحويل مخطط MySQL إلى مخطط متوافق مع Spanner وإجراء تعديلات متقدّمة على المخطط
• كيفية إجراء عملية نقل بيانات مجمّعة من مثيل MySQL مقسَّم إلى Cloud Spanner باستخدام Dataflow
• كيفية إعداد عملية النسخ المتماثل المستمر (CDC) من مثيل MySQL مقسَّم إلى Cloud Spanner باستخدام Datastream وDataflow
• كيفية إعداد عملية النسخ المتماثل العكسي من Spanner إلى مثيلات MySQL المقسّمة
• كيفية استخدام "عمليات التحويل المخصّصة" لتعبئة أعمدة إضافية أثناء عمليات النقل المجمّعة والمباشرة والعكسية
• كيفية ضبط عمليات تحويل التقسيم باستخدام المفاتيح الأساسية

لا يتناول هذا الدرس التطبيقي حول الترميز ما يلي:

• إعدادات الشبكة المخصّصة المتقدّمة
• إنشاء نماذج Dataflow مخصّصة من البداية
• تحسين أداء عملية نقل البيانات
• نقل التطبيقات: يركّز هذا الدرس العملي على طبقة قاعدة البيانات (المخطط والبيانات). ولا يشمل العملية التشغيلية لإعادة نشر خدمات التطبيق أو نقلها.

ملاحظة: على الرغم من أنّ هذا الدرس التطبيقي حول الترميز يتعامل مع البيانات والمخطط، عليك تعديل الرمز البرمجي للتطبيق يدويًا، بما في ذلك تحديث برامج تشغيل قاعدة البيانات ومنطق الربط وما إلى ذلك، لإكمال عملية نقل البيانات الشاملة.

المتطلبات

• مشروع Google Cloud تم تفعيل الفوترة فيه
• أذونات IAM كافية لتفعيل واجهات برمجة التطبيقات وإنشاء موارد Spanner وDataflow وDatastream وGCS وإدارتها على الرغم من أنّ دور "المشروع" Owner هو الأبسط في codelab، سيتم تناول أدوار أكثر تحديدًا في "إعداد البيئة".
• سنوفّر جهازًا افتراضيًا صغيرًا من Compute Engine خلال مرحلة الإعداد لمحاكاة الخادم المحلي. تأكَّد من أنّ حصة مشروعك تسمح بإنشاء آلات افتراضية.
• متصفّح ويب، مثل Google Chrome
• معرفة أساسية بوحدة تحكّم Google Cloud وأدوات سطر الأوامر، مثل gcloud
• الوصول إلى بيئة shell ننصح باستخدام Cloud Shell لأنّه يتضمّن gcloud.

يمكنك الاطّلاع على مزيد من التفاصيل حول عملية الإعداد أعلاه في قسم إعداد البيئة.

2. فهم عملية نقل البيانات

تتضمّن عملية نقل قاعدة بيانات مقسّمة دمج عدة مثيلات فعلية ومنطقية من MySQL في قاعدة بيانات Spanner واحدة قابلة للتوسّع أفقيًا. يوضّح هذا القسم البنية والأدوات الرئيسية المستخدَمة في عملية نقل البيانات.

بنية مسار نقل البيانات

تتضمّن عملية نقل البيانات المراحل التالية:

1. تحويل المخطط:

• الغرض: تحويل مخطط قاعدة البيانات المصدر إلى مخطط متوافق مع Cloud Spanner
• الأداة: أداة نقل البيانات إلى Spanner (SMT)
• العملية: يحلّل SMT مخطط قاعدة البيانات المصدر وينشئ لغة تعريف البيانات (DDL) المكافئة في Spanner. في مثيل Spanner المستهدف، يتم إنشاء قاعدة بيانات ثم يتم تطبيق DDL تلقائيًا.

2. نقل البيانات المجمّعة:

• الغرض: إجراء عملية تحميل كاملة وأولية للبيانات الحالية من قاعدة البيانات المصدر إلى جداول Spanner التي تم توفيرها
• الأداة: Dataflow، باستخدام نموذج Sourcedb to Spanner الذي توفّره Google
• العملية: تقرأ مهمة Dataflow هذه جميع البيانات من جداول المصدر المحدّدة وتكتبها في جداول Spanner المقابلة. يتم ذلك بعد إنشاء مخطط Spanner.

3- النقل المباشر (CDC):

• الغرض: تسجيل التغييرات المستمرة وتطبيقها من قاعدة البيانات المصدر إلى Cloud Spanner في الوقت الفعلي تقريبًا، ما يقلّل من وقت التوقف عن العمل أثناء عملية نقل البيانات
• الأدوات:
• Datastream: تسجّل التغييرات (عمليات الإدراج والتعديل والحذف) من قاعدة البيانات المصدر وتكتبها في Cloud Storage (GCS).
• Dataflow: تستخدِم نموذج Datastream to Spanner لقراءة أحداث التغيير من GCS وتطبيقها على Cloud Spanner.

4. النسخ المتماثل العكسي:

• الغرض: تكرار تغييرات البيانات من Cloud Spanner إلى قاعدة البيانات المصدر. يمكن أن يكون ذلك مفيدًا لاستراتيجيات الاحتياط أو عمليات النقل المرحلية أو الاحتفاظ بنسخة طبق الأصل في المصدر لحالات استخدام معيّنة.
• الأداة: Dataflow، باستخدام النموذج Spanner to SourceDb
• المعالجة: تستخدم هذه المهمة "قنوات تغيير" Spanner لتسجيل التعديلات في Spanner وإعادة كتابتها إلى مثيل قاعدة البيانات المصدر.

يوضّح المخطط التالي المكوّنات وتدفّق البيانات:

المصطلحات الرئيسية:

• الجزء الفعلي: هو الخادم الأساسي أو مثيل الحوسبة الذي يستضيف قاعدة البيانات (في حالتنا، الجهاز الظاهري لمحاكي GCE على الجهاز).
• التقسيم المنطقي: هو مخطط قاعدة البيانات الفردي داخل خادم فعلي.
• الجهاز الافتراضي (VM) في Compute Engine (GCE): هو جهاز افتراضي مستضاف على البنية التحتية لـ Google Cloud. في هذا الدرس التطبيقي حول الترميز، نستخدم جهازًا افتراضيًا على Google Compute Engine لمحاكاة خادم مستقل "محلي" بدون نظام تشغيل يستضيف قاعدة بيانات MySQL المصدر.
• أداة نقل البيانات في Spanner (SMT): هي أداة تُستخدم لتقييم مخططات MySQL واقتراح مكافئات لمخططات Spanner وإنشاء لغة تعريف البيانات (DDL) في Spanner.
• لغة تعريف البيانات (DDL): عبارات تُستخدم لتحديد بنية قاعدة البيانات وتعديلها، مثل عبارات CREATE TABLE. تنشئ أداة SMT لغة تعريف البيانات (DDL) في Spanner استنادًا إلى مخطط Cloud SQL.
• ‫Dataflow: هي خدمة مُدارة بالكامل لمعالجة البيانات بدون خادم. في هذا الدرس التطبيقي حول الترميز، يتم استخدامها لتشغيل النماذج التي توفّرها Google لنقل البيانات المجمّعة، وتطبيق تغييرات Datastream، وعكس عملية النسخ المتماثل.
• Datastream: هي خدمة حوسبة بدون خادم لتسجيل بيانات التغيير (CDC) وإنشاء نسخ متماثلة. يتم استخدامها لبث التغييرات من مثيل MySQL المستضاف محليًا إلى Cloud Storage في هذا الدرس التطبيقي حول الترميز.
• Spanner Change Streams: هي إحدى ميزات Spanner التي تتيح بث التغييرات على البيانات (عمليات الإدراج والتعديل والحذف) في الوقت الفعلي، وتُستخدَم كمصدر للنسخ المتماثل العكسي.
• ‫Pub/Sub: هي خدمة مراسلة تُستخدَم لفصل الخدمات التي تنشئ الأحداث عن الخدمات التي تعالجها. في هذا الدرس التطبيقي حول الترميز، يتم تشغيل Dataflow لمعالجة التعديلات كلما حمّل Datastream ملفات تغيير جديدة إلى Cloud Storage.

3- إعداد البيئة

قبل بدء عملية نقل البيانات، عليك إعداد مشروعك على السحابة الإلكترونية على Google Cloud وتفعيل الخدمات اللازمة.

1. اختيار مشروع Google Cloud أو إنشاؤه

يجب أن يكون لديك مشروع على Google Cloud مع تفعيل الفوترة لاستخدام الخدمات في هذا الدرس العملي.

1. في Google Cloud Console، انتقِل إلى صفحة اختيار المشروع: الانتقال إلى أداة اختيار المشاريع
2. اختَر مشروعًا على Google Cloud أو أنشِئ مشروعًا.
3. تأكَّد من تفعيل الفوترة لمشروعك. كيفية التأكّد من تفعيل الفوترة لمشروعك

2. فتح Cloud Shell

‫Cloud Shell هي بيئة سطر أوامر تعمل في Google Cloud ومحمّلة مسبقًا بواجهة سطر الأوامر gcloud والأدوات الأخرى التي تحتاج إليها.

• انقر على الزر تفعيل Cloud Shell في أعلى يسار Google Cloud Console.
• يتم فتح جلسة Cloud Shell داخل إطار جديد في أسفل وحدة التحكّم وعرض موجه سطر الأوامر.

تنتهي مهلة جلسات Cloud Shell بعد فترة من عدم النشاط، ما يؤدي إلى محو متغيرات البيئة المحفوظة. في حال انقطاع اتصال جلستك، يمكنك استعادة عملية الإعداد بسرعة من خلال نسخ القائمة الموحّدة للمتغيّرات المتوفّرة في قسم تنظيف الموارد في نهاية هذا الدرس التطبيقي حول الترميز.

3- ضبط متغيرات المشروع والبيئة

في Cloud Shell، اضبط بعض متغيرات البيئة لرقم تعريف مشروعك والمنطقة التي ستستخدمها.

export PROJECT_ID=$(gcloud config get-value project)
export REGION="us-central1" # Or your preferred region
export ZONE="us-central1-a" # Or a zone within your selected region

gcloud config set project $PROJECT_ID
gcloud config set compute/region $REGION
gcloud config set compute/zone $ZONE

echo "Project ID: $PROJECT_ID"
echo "Region: $REGION"
echo "Zone: $ZONE"

ملاحظة مهمة: استخدِم المنطقة نفسها لجميع الموارد التي تنشئها في هذا الدرس التطبيقي حول الترميز (Cloud Spanner وDatastream وDataflow) لتقليل وقت الاستجابة وتعقيد الشبكة.

4. تفعيل واجهات Google Cloud APIs المطلوبة

فعِّل واجهات برمجة التطبيقات اللازمة لخدمات Cloud Spanner وDataflow وDatastream والخدمات الأخرى ذات الصلة.

gcloud services enable \
  spanner.googleapis.com \
  dataflow.googleapis.com \
  datastream.googleapis.com \
  pubsub.googleapis.com \
  storage.googleapis.com \
  compute.googleapis.com \
  sqladmin.googleapis.com \
  servicenetworking.googleapis.com \
  cloudresourcemanager.googleapis.com

قد يستغرق إكمال هذا الأمر بضع دقائق.

4. إعداد قاعدة بيانات MySQL المصدر

في هذا القسم، سنحاكي بنية MySQL مجزّأة محلية من خلال توفير جهازَين افتراضيَين من Compute Engine (الجزءان "الماديان"). بعد ذلك، سنثبِّت MySQL على كليهما وننشئ قاعدتَي بيانات (التقسيم المنطقي) على كل آلة افتراضية.

1. إنشاء أجهزة Compute Engine الافتراضية (الأجزاء المادية)

نفِّذ الأوامر التالية في Cloud Shell لإنشاء آلتين افتراضيتين تعملان بنظام التشغيل Ubuntu. سنخصّص لها علامات شبكة للسماح بحركة بيانات MySQL الواردة لاحقًا.

# Create Physical Shard 1
gcloud compute instances create mysql-physical-1 \
    --zone=$ZONE \
    --machine-type=e2-small \
    --image-family=ubuntu-2204-lts \
    --image-project=ubuntu-os-cloud \
    --tags=mysql-server

# Create Physical Shard 2
gcloud compute instances create mysql-physical-2 \
    --zone=$ZONE \
    --machine-type=e2-small \
    --image-family=ubuntu-2204-lts \
    --image-project=ubuntu-os-cloud \
    --tags=mysql-server

2. ضبط قواعد جدار الحماية

للسماح بالوصول الآمن عبر بروتوكول النقل الآمن (SSH) بدون إتاحة الوصول للجميع ولتفعيل إمكانية الاتصال بخدمة Datastream، اتّبِع الخطوات التالية:

إنشاء قاعدة جدار حماية لبروتوكول SSH من خلال IAP:

تسمح هذه القاعدة لخدمة Identity-Aware Proxy بالوصول إلى الأجهزة الافتراضية على منفذ SSH (22).

gcloud compute firewall-rules create allow-ssh-iap \
    --direction=INGRESS \
    --priority=1000 \
    --network=default \
    --action=ALLOW \
    --rules=tcp:22 \
    --source-ranges=35.235.240.0/20 \
    --target-tags=mysql-server

إنشاء قاعدة جدار حماية لخدمة Datastream (منفذ MySQL):

يجب أن يتمكّن Datastream من الوصول إلى هذه الأجهزة الافتراضية على منفذ MySQL العادي (3306).

gcloud compute firewall-rules create allow-mysql-datastream \
    --direction=INGRESS \
    --priority=1000 \
    --network=default \
    --action=ALLOW \
    --rules=tcp:3306 \
    --source-ranges=0.0.0.0/0 \
    --target-tags=mysql-server

لا يُنصح باستخدام 0.0.0.0/0 في بيئات الإنتاج لأنّه يعرّض قاعدة البيانات للإنترنت المتاح للجميع. يتم ذلك هنا فقط لتبسيط الدرس العملي. في سيناريوهات العالم الحقيقي، يجب تقييد الوصول إلى قاعدة البيانات:

• مصدر البيانات: بدلاً من 0.0.0.0/0، عليك السماح فقط بعناوين IP العامة لمصدر البيانات في منطقتك.
• Dataflow: لتجنُّب عرض Cloud SQL على 0.0.0.0/0 لعاملين في Dataflow، استخدِم عنوان IP خاص في Cloud SQL واضبط Dataflow لاستخدام شبكة VPC نفسها.
• لإجراء اتصالات آمنة ومفوَّضة من خلال "إدارة الهوية والوصول" من أي بيئة، استخدِم خادم Cloud SQL الوكيل للمصادقة.

3- تثبيت MySQL وضبطه على الجزء 1 من قاعدة البيانات

يمكنك استخدام SSH للوصول إلى جهازك الافتراضي الأول من أجل تثبيت MySQL وإعداد تسجيل البيانات الثنائية (الذي تتطلّبه خدمة Datastream لإجراء عملية النسخ المتماثل المباشر).

1. استخدِم بروتوكول SSH للوصول إلى الجهاز الافتراضي الأول:

gcloud compute ssh mysql-physical-1 --zone=$ZONE --tunnel-through-iap

توجّه العلامة --tunnel-through-iap اتصال SSH من خلال خدمة Identity-Aware Proxy (IAP) من Google. هذه إحدى أفضل ممارسات الأمان لأنّها تتيح لك الاتصال بالجهاز الظاهري بشكل آمن بدون إتاحة منفذ SSH (المنفذ 22) على الإنترنت المتاح للجميع.

2. ثبِّت MySQL:

sudo apt-get update
sudo apt-get install mysql-server-8.0 -y

# Verify the installation and version
sudo mysql --version

3. اضبط ملف mysqld.cnf لتفعيل تسجيل الدخول الثنائي والسماح بالاتصالات الخارجية:

sudo sed -i 's/bind-address.*/bind-address = 0.0.0.0/' /etc/mysql/mysql.conf.d/mysqld.cnf
echo -e "[mysqld]\nserver-id=1\nlog_bin=/var/log/mysql/mysql-bin.log\nbinlog_format=ROW" | sudo tee -a /etc/mysql/mysql.conf.d/mysqld.cnf

4. أعِد تشغيل MySQL لتطبيق التغييرات:

sudo systemctl restart mysql

4. إنشاء أقسام منطقية وإدراج البيانات وإنشاء مستخدم مصدر بيانات (القسم 1)

أثناء تسجيل الدخول إلى mysql-physical-1 باستخدام بروتوكول SSH، سجِّل الدخول إلى موجه MySQL:

sudo mysql

نفِّذ أوامر SQL التالية. ينشئ هذا النص البرمجي جزأين منطقيين منفصلين (shard0_db وshard1_db)، ويضبط المخطط المتطابق في كليهما، ويدرج بيانات يمكن التعرّف عليها بدقة في كل جزء (لتوضيح التقسيم إلى أجزاء)، وينشئ مستخدم النسخ المتماثل لخدمة Datastream.

نفِّذ أوامر SQL التالية لإنشاء أول شريحتَين منطقيتَين وجدول ومستخدم النسخ المتماثل لخدمة Datastream:

CREATE DATABASE shard0_db;
CREATE DATABASE shard1_db;

USE shard0_db;
CREATE TABLE Customers (
    CustomerId INT NOT NULL,
    CustomerName VARCHAR(255),
    CreditLimit DECIMAL(10, 2) NOT NULL,
    LegacyRegion VARCHAR(50), -- Renamed to LoyaltyTier in Spanner

    PRIMARY KEY (CustomerId),
    CONSTRAINT CHK_CreditLimit CHECK (CreditLimit > 1000) -- Relaxed in Spanner to > 0 
);

CREATE TABLE Orders (
    CustomerId INT NOT NULL,  -- Logically references Customers.CustomerId in Spanner
    OrderId INT NOT NULL,
    OrderValue DECIMAL(10, 2),
    LegacyOrderSystem VARCHAR(50), -- Extra column in Source, to be dropped in Spanner

    PRIMARY KEY (CustomerId, OrderId) -- Spanner PK will have one additional column in PK
);

INSERT INTO Customers (CustomerId, CustomerName, CreditLimit, LegacyRegion) VALUES
(4, 'David E.', 2000.00, 'EAST'),
(8, 'Eleanor F.', 8100.00, 'WEST'),
(12, 'Frank G.', 12000.00, 'NORTH'),
(16, 'Grace H.', 6500.00, 'SOUTH');

INSERT INTO Orders (CustomerId, OrderId, OrderValue, LegacyOrderSystem) VALUES
(4, 101, 150.00, 'WebStore_v1'),
(4, 102, 25.50, 'InStore_POS'),
(8, 103, 75.00, 'MobileApp_Legacy'),
(12, 104, 3000.00, 'WebStore_v1'),
(16, 105, 120.00, 'Partner_API');

USE shard1_db;
CREATE TABLE Customers (
    CustomerId INT NOT NULL,
    CustomerName VARCHAR(255),
    CreditLimit DECIMAL(10, 2) NOT NULL,
    LegacyRegion VARCHAR(50), -- Renamed to LoyaltyTier in Spanner
    PRIMARY KEY (CustomerId),
    CONSTRAINT CHK_CreditLimit CHECK (CreditLimit > 1000) -- Relaxed in Spanner to > 0 
);

CREATE TABLE Orders (
    CustomerId INT NOT NULL,  -- Logically references Customers.CustomerId in Spanner
    OrderId INT NOT NULL,
    OrderValue DECIMAL(10, 2),
    LegacyOrderSystem VARCHAR(50), -- Extra column in Source, to be dropped in Spanner
    PRIMARY KEY (CustomerId, OrderId) -- Spanner PK will have one additional column in PK
);

INSERT INTO Customers (CustomerId, CustomerName, CreditLimit, LegacyRegion) VALUES
(1, 'Agnes N.', 5100.00, 'NORTHEAST'),(5, 'Alice I.', 15000.00, 'EAST'),
(9, 'Bob J.', 7500.00, 'WEST'),
(13, 'Charlie K.', 2200.00, 'CENTRAL');

INSERT INTO Orders (CustomerId, OrderId, OrderValue, LegacyOrderSystem) VALUES
(1, 201, 50.00, 'MobileApp_Legacy'),
(5, 202, 1250.00, 'WebStore_v1'),
(5, 203, 80.00, 'Partner_API'),
(9, 204, 600.00, 'InStore_POS'),
(13, 205, 199.99, 'WebStore_v1');


-- Create Datastream Replication User
CREATE USER 'datastream_user'@'%' IDENTIFIED BY 'complex_password_123';
GRANT REPLICATION SLAVE, REPLICATION CLIENT, SELECT, INSERT, UPDATE, DELETE ON *.* TO 'datastream_user'@'%'; 
FLUSH PRIVILEGES;

يمكن العثور على ملف التفريغ الخاص بالمخطط أعلاه هنا. من المهم إنشاء مستخدم لنسخ البيانات من مصدرها بشكل منفصل لأنّ ذلك غير مضمّن في ملف التفريغ.

5- التحقّق من البيانات

للتحقّق بسرعة من توفّر البيانات، اتّبِع الخطوات التالية:

SELECT 'Customers shard0_db' AS tbl, COUNT(*) FROM shard0_db.Customers
UNION ALL
SELECT 'Orders shard0_db', COUNT(*) FROM shard0_db.Orders
UNION ALL
SELECT 'Customers shard1_db', COUNT(*) FROM shard1_db.Customers
UNION ALL
SELECT 'Orders shard1_db', COUNT(*) FROM shard1_db.Orders;
EXIT;

الناتج المتوقّع:

+---------------------+----------+
| tbl                 | COUNT(*) |
+---------------------+----------+
| Customers shard0_db |        4 |
| Orders shard0_db    |        5 |
| Customers shard1_db |        4 |
| Orders shard1_db    |        5 |
+---------------------+----------+

أدخِل exit للخروج من الاتصال بالجهاز الظاهري 1 الخاص بالجزء الفعلي.

6. كرِّر الخطوات نفسها مع Physical Shard 2.

عليك الآن تكرار العملية نفسها تمامًا للجهاز الظاهري الثاني، ولكن ستنشئ shard2_db وshard3_db، وتغيّر server-id.

1. استخدِم بروتوكول النقل الآمن (SSH) في الجهاز الظاهري الثاني:

gcloud compute ssh mysql-physical-2 --zone=$ZONE --tunnel-through-iap

2. ثبِّت MySQL:

sudo apt-get update
sudo apt-get install mysql-server-8.0 -y

3. اضبط ملف mysqld.cnf لتفعيل تسجيل البيانات الثنائية والسماح بالاتصالات الخارجية [يجب أن يكون رقم تعريف الخادم مختلفًا (مثلاً، 2)]

sudo sed -i 's/bind-address.*/bind-address = 0.0.0.0/' /etc/mysql/mysql.conf.d/mysqld.cnf

echo -e "[mysqld]\nserver-id=2\nlog_bin=/var/log/mysql/mysql-bin.log\nbinlog_format=ROW" | sudo tee -a /etc/mysql/mysql.conf.d/mysqld.cnf

4. أعِد تشغيل MySQL لتطبيق التغييرات:

sudo systemctl restart mysql

5. أدخِل MySQL (sudo mysql) ونفِّذ نسخة معدَّلة قليلاً من SQL من الخطوة 4:

CREATE DATABASE shard2_db;
CREATE DATABASE shard3_db;

USE shard2_db;
CREATE TABLE Customers (
    CustomerId INT NOT NULL,
    CustomerName VARCHAR(255),
    CreditLimit DECIMAL(10, 2) NOT NULL,
    LegacyRegion VARCHAR(50), -- Renamed to LoyaltyTier in Spanner
    PRIMARY KEY (CustomerId),
    CONSTRAINT CHK_CreditLimit CHECK (CreditLimit > 1000) -- Relaxed in Spanner to > 0 
);

CREATE TABLE Orders (
    CustomerId INT NOT NULL,  -- Logically references Customers.CustomerId in Spanner
    OrderId INT NOT NULL,
    OrderValue DECIMAL(10, 2),
    LegacyOrderSystem VARCHAR(50), -- Extra column in Source, to be dropped in Spanner
    PRIMARY KEY (CustomerId, OrderId) -- Spanner PK will have one additional column in PK
);

INSERT INTO Customers (CustomerId, CustomerName, CreditLimit, LegacyRegion) VALUES
(2, 'Brian K.', 2500.00, 'SOUTHWEST'),
(6, 'Diana L.', 1999.00, 'NORTH'),
(10, 'Edward M.', 11000.00, 'EAST'),
(14, 'Fiona N.', 3000.00, 'WEST');

INSERT INTO Orders (CustomerId, OrderId, OrderValue, LegacyOrderSystem) VALUES
(2, 301, 100.00, 'CallCenter_System'),
(6, 302, 99.00, 'MobileApp_Legacy'),
(10, 303, 1000.00, 'WebStore_v1'),
(10, 304, 2500.00, 'InStore_POS'),
(14, 305, 130.00, 'MobileApp_Legacy');

USE shard3_db;
CREATE TABLE Customers (
    CustomerId INT NOT NULL,
    CustomerName VARCHAR(255),
    CreditLimit DECIMAL(10, 2) NOT NULL,
    LegacyRegion VARCHAR(50), -- Renamed to LoyaltyTier in Spanner
    PRIMARY KEY (CustomerId),
    CONSTRAINT CHK_CreditLimit CHECK (CreditLimit > 1000) -- Relaxed in Spanner to > 0 
);

CREATE TABLE Orders (
    CustomerId INT NOT NULL,  -- Logically references Customers.CustomerId in Spanner
    OrderId INT NOT NULL,
    OrderValue DECIMAL(10, 2),
    LegacyOrderSystem VARCHAR(50), -- Extra column in Source, to be dropped in Spanner
    PRIMARY KEY (CustomerId, OrderId) -- Spanner PK will have one additional column in PK
);

INSERT INTO Customers (CustomerId, CustomerName, CreditLimit, LegacyRegion) VALUES
(3, 'Cathy Z.', 6000.00, 'CENTRAL'),
(7, 'George O.', 18000.00, 'SOUTH'),
(11, 'Helen P.', 4000.00, 'NORTHEAST'),
(15, 'Ivy Q.', 9500.00, 'SOUTHWEST');

INSERT INTO Orders (CustomerId, OrderId, OrderValue, LegacyOrderSystem) VALUES
(3, 401, 600.00, 'InStore_POS'),
(7, 402, 1200.00, 'CallCenter_System'),
(11, 403, 350.00, 'MobileApp_Legacy'),
(15, 404, 800.00, 'WebStore_v1'),
(99, 999, 25.00, 'CallCenter_System'); -- Failure row during Bulk Migration due to violation of interleaving

-- Create Datastream Replication User
CREATE USER 'datastream_user'@'%' IDENTIFIED BY 'complex_password_123';
GRANT REPLICATION SLAVE, REPLICATION CLIENT, SELECT, INSERT, UPDATE, DELETE ON *.* TO 'datastream_user'@'%'; 
FLUSH PRIVILEGES;

-- Verify Data
SELECT 'Customers shard2_db' AS tbl, COUNT(*) FROM shard2_db.Customers
UNION ALL
SELECT 'Orders shard2_db', COUNT(*) FROM shard2_db.Orders
UNION ALL
SELECT 'Customers shard3_db', COUNT(*) FROM shard3_db.Customers
UNION ALL
SELECT 'Orders shard3_db', COUNT(*) FROM shard3_db.Orders;

EXIT;

الناتج المتوقّع:

+---------------------+----------+
| tbl                 | COUNT(*) |
+---------------------+----------+
| Customers shard2_db |        4 |
| Orders shard2_db    |        5 |
| Customers shard3_db |        4 |
| Orders shard3_db    |        5 |
+---------------------+----------+

يمكن العثور على ملف التفريغ الخاص بالمخطط أعلاه هنا. من المهم إنشاء مستخدم لنسخ البيانات من مصدرها بشكل منفصل لأنّ ذلك غير مضمّن في ملف التفريغ.

أدخِل exit للخروج من الاتصال بالجهاز الافتراضي.

5- إعداد Cloud Spanner

الآن، عليك إعداد مثيل Cloud Spanner المستهدف الذي سيتم نقل البيانات إليه.

1. إنشاء مثيل Cloud Spanner

أنشئ مثيلاً من Cloud Spanner في المنطقة نفسها التي تتضمّن أجهزة Compute Engine الافتراضية لتقليل وقت الاستجابة. ينشئ هذا الأمر مثيلاً صغيرًا مناسبًا لهذا الدرس التطبيقي حول الترميز، باستخدام 100 وحدة معالجة.

export SPANNER_INSTANCE_NAME="target-spanner-instance"
export SPANNER_DATABASE_NAME="sharded-target-db"
export SPANNER_CONFIG="regional-${REGION}"

gcloud spanner instances create $SPANNER_INSTANCE_NAME \
  --config=$SPANNER_CONFIG \
  --description="Target Spanner Instance" \
  --processing-units=100

قد يستغرق إنشاء الجهاز الافتراضي دقيقة أو دقيقتَين.

على الرغم من أنّنا أنشأنا أربع قواعد بيانات منطقية منفصلة (أجزاء) في MySQL، علينا إنشاء مثيل قاعدة بيانات واحد فقط في Cloud Spanner. ستعمل خدمة Spanner كهدف موحّد وقابل للتوسيع أفقيًا لجميع بياناتنا المقسّمة.

6. تحويل المخطط باستخدام "أداة نقل البيانات إلى Spanner" (SMT)

استخدِم واجهة مستخدم الويب لأداة نقل البيانات إلى Spanner (SMT) للاتصال بأحد الأقسام المنطقية (shard0_db) وتحليل المخطط الخاص به وتطبيق العديد من التعديلات المتقدّمة قبل تحويله إلى Cloud Spanner.

1. تثبيت SMT

سنشغّل واجهة مستخدم الويب الخاصة بأداة SMT مباشرةً من Cloud Shell. في نافذة Cloud Shell، نزِّل أحدث إصدار من "أداة نقل البيانات الذكية" واستخرِجه:

sudo apt-get update && sudo apt-get install google-cloud-cli-spanner-migration-tool

# Verify installation 
gcloud alpha spanner migrate web --help

لا تعمل هذه الأوامر إلا على نظام التشغيل Linux. بالنسبة إلى Windows أو Mac، يمكنك إنشاء التطبيق من المصدر باتّباع هذه الخطوات.

2. الربط بقاعدة البيانات المصدر

1. مصادقة جلستك

# Authenticate your Google Cloud account
gcloud auth login

# Set up Application Default Credentials (ADC) for SMT
gcloud auth application-default login

# Ensure your current project is set correctly
gcloud config set project $PROJECT_ID

(ملاحظة: عند مطالبتك بذلك، اتّبِع عنوان URL المقدَّم لتفويض حسابك وأعِد لصق رمز التحقّق في نافذة الأوامر.)

2. أولاً، ابحث عن عنوان IP الخارجي للجزء المادي الأول من خلال تنفيذ ما يلي في علامة تبويب جديدة في Cloud Shell:

gcloud compute instances describe mysql-physical-1 --zone=$ZONE --format='get(networkInterfaces[0].accessConfigs[0].natIP)'

3. اطبع تفاصيل مثيل Spanner المستهدف لاستخدامها أثناء ضبط SMT.

echo "Project ID: $PROJECT_ID"
echo "Instance ID: $SPANNER_INSTANCE_NAME"
echo "Database Name: $SPANNER_DATABASE_NAME"

4. تشغيل واجهة المستخدم على الويب:

gcloud alpha spanner migrate web --port=8080

5. في أعلى يسار نافذة Cloud Shell، انقر على رمز المعاينة على الويب (يبدو كعين) واختَر المعاينة على المنفذ 8080. سيؤدي ذلك إلى فتح واجهة مستخدم "أداة نقل بيانات التسوّق" في علامة تبويب متصفّح جديدة.

6. في واجهة مستخدم الويب الخاصة بأداة SMT، انقر على الربط بقاعدة بيانات.
7. أدخِل تفاصيل الربط:

• نوع قاعدة البيانات: MySQL
• المضيف: (ألصِق عنوان IP من الخطوة 2)
• المنفذ: 3306
• المستخدم: datastream_user
• كلمة المرور: complex_password_123
• اسم قاعدة البيانات: shard0_db

لإنشاء مخططنا الأساسي، ما علينا سوى ربط SMT بأحد الأجزاء المنطقية المتطابقة.

8. انقر على زر التعديل في أعلى يسار الصفحة لإعداد قاعدة بيانات Spanner.
9. أدخِل تفاصيل Target Spanner:

• رقم تعريف المشروع: (ألصِق رقم تعريف المشروع من الخطوة 3)
• مثيل Spanner: (ألصِق رقم تعريف المثيل من الخطوة 3)

10. انقر على اختبار الاتصال.
11. بعد اجتياز هذا الاختبار، انقر على ربط. ستحلّل "أداة نقل البيانات من SQL Server إلى Spanner" قاعدة البيانات المصدر وستعرض مخططًا أساسيًا لـ Spanner.

3- تطبيق تعديلات المخطط

سنعيد الآن تشكيل المخطط لتغطية سيناريوهات النقل المعقّدة.

تهدف تعديلات المخطط التي تم تطبيقها في هذا القسم إلى توضيح حالات استخدام بسيطة لهذا الدرس التطبيقي حول الترميز. بالنسبة إلى عمليات نقل البيانات الفعلية في بيئة الإنتاج، ننصحك بشدة بمراجعة أفضل الممارسات لتصميم المخطط في Cloud Spanner لضمان تحسين قاعدة البيانات من أجل الأداء وقابلية التوسّع.

في "محرّر المخطط" بواجهة مستخدم SMT، نفِّذ الإجراءات التالية:

أ. أعِد تسمية عمود LegacyRegion:

• انقر على جدول Customers في جزء التنقّل الأيمن. سيتم فتح علامة التبويب الأعمدة تلقائيًا.
• انقر على الزر "تعديل" في قسم Spanner.
• ابحث عن العمود LegacyRegion في طريقة عرض مخطط Spanner.
• غيِّر اسم عمود Spanner إلى LoyaltyTier عن طريق الكتابة في مربّع حوار اسم العمود.
• انقر على حفظ وتحويل.

ب. تخفيف قيد التحقّق:

• في جدول Customers، انتقِل إلى علامة التبويب قيود التحقّق.
• ابحث عن القيد CHK_CreditLimit. انقر على رمز التعديل (القلم).
• غيِّر الشرط من CreditLimit > 1000 إلى CreditLimit > 0. (سيؤدي ذلك عمدًا إلى تعذُّر نقل الصفوف التي تتضمّن حدودًا أدنى للرصيد بشكل عكسي، وسيتم نقلها إلى قائمة الرسائل غير المرسَلة).

ج. إزالة عمود LegacyOrderSystem:

• انقر على جدول Orders، وسيتم فتح علامة التبويب الأعمدة تلقائيًا.
• انقر على الزر "تعديل" في قسم Spanner.
• ابحث عن العمود LegacyOrderSystem في طريقة عرض مخطط Spanner.
• انقر على رمز قائمة النقاط الثلاث بجانبه واختَر إزالة العمود.
• انقر على حفظ وتحويل.

د. أضِف عمود OrderSource واجعله مفتاحًا أساسيًا:

• في جدول Orders، انقر على إضافة عمود. أطلِق عليها الاسم OrderSource واضبط النوع على STRING مع الطول 50، بدون إنشاء تلقائي، واضبط IsNullable على No.
• انتقِل إلى علامة التبويب المفتاح الأساسي.
• انقر على تعديل واختَر OrderSource من القائمة المنسدلة "اسم العمود".
• انقر على إضافة عمود، ثمّ على حفظ وتحويل.

E. دمج جدول الطلبات:

• في جدول Orders، ابحث عن علامة التبويب الدمج في عرض الجدول الرئيسي.
• اضبط الجدول الرئيسي على Customers.
• اختَر IN PARENT نوع التداخل وNO ACTION إجراء الحذف.
• انقر على حفظ.

• تتيح واجهة مستخدم "أداة تعديل المخطط" إجراء مجموعة واسعة من تعديلات المخطط. يمكنك الاطّلاع على قائمة شاملة في المستندات الرسمية.
• تعرض لوحة "المشاكل والاقتراحات" في "أداة نقل المخطط" (على يسار مساحة العمل الرئيسية) أي تغييرات حدثت في عملية تحويل المخطط. يمكنك الاطّلاع على مزيد من التفاصيل هنا.

4. تنزيل ملفات عمليات الإلغاء وتطبيق المخطط

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

تُنشئ أداة نقل البيانات إلى Spanner عدة ملفات أثناء تشغيلها. لكل ملف وظيفة خاصة، بدءًا من تخزين حالة نقل البيانات الحالية إلى إنشاء تقرير للمخطط الذي تم تحليله. يمكنك الاطّلاع على مزيد من التفاصيل حول هذه "العناصر" هنا.

2. انقر على إعداد عملية نقل البيانات.

3. اختَر وضع النقل كـ Schema من القائمة المنسدلة.
4. أدخِل قاعدة بيانات Target Spanner: sharded-target-db

5. انقر على نقل البيانات.
6. ستطبّق أداة SMT تعريف البيانات (DDL) وتنشئ قاعدة بيانات Spanner. يمكنك إيقاف عملية SMT بأمان في Cloud Shell (Ctrl+C) بعد اكتمالها.

5- التحقّق من صحة المخطط في Cloud Spanner

تأكَّد من إنشاء الجداول في قاعدة بيانات Spanner.

gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
  --instance=$SPANNER_INSTANCE_NAME \
  --sql="SELECT table_name FROM information_schema.tables WHERE table_schema = '' ORDER BY table_name"

من المفترض أن يظهر لك الناتج التالي:

table_name: Customers
table_name: Orders

اختياري: إذا أردت التحقّق من صحة قيود التحقّق والتقسيم المتداخل والأعمدة الإضافية من خلال مراجعة Spanner DDL الفعلي، نفِّذ الأمر التالي:

gcloud spanner databases ddl describe $SPANNER_DATABASE_NAME \
  --instance=$SPANNER_INSTANCE_NAME

الناتج المتوقّع:

CREATE TABLE Customers (
  CustomerId INT64 NOT NULL,
  CustomerName STRING(255),
  CreditLimit NUMERIC NOT NULL,
  LoyaltyTier STRING(50),
  CONSTRAINT CHK_CreditLimit CHECK(`CreditLimit` > 0),
) PRIMARY KEY(CustomerId);

CREATE TABLE Orders (
  CustomerId INT64 NOT NULL,
  OrderId INT64 NOT NULL,
  OrderValue NUMERIC,
  OrderSource STRING(50) NOT NULL,
) PRIMARY KEY(CustomerId, OrderId, OrderSource),
  INTERLEAVE IN PARENT Customers ON DELETE NO ACTION;

7. بدء عملية "تسجيل البيانات المتغيرة" (CDC)

في هذا القسم، عليك إعداد "المسجّل" لعملية نقل البيانات. من خلال إعداد Datastream وPub/Sub قبل بدء تحميل البيانات المجمّعة، يمكنك التأكّد من تسجيل كل تغيير يتم إجراؤه على قواعد البيانات المصدر وإضافته إلى قائمة الانتظار، ما يمنع فقدان أي بيانات أثناء عملية النقل. يجب إكمال عملية الإعداد هذه لاستخدام ميزة "نقل البيانات المباشر".

بما أنّ بنيتنا الأساسية تتضمّن خادمَين ماديَين، علينا إنشاء ملفَّي مصدر منفصلَين في Datastream ومصدرَين منفصلَين في Datastream. سيتم إرسال البيانات من كلا المصدرَين إلى حزمة واحدة في Google Cloud Storage (GCS)، وستكون هذه الحزمة هي المصدر الموحّد لبرنامج Dataflow.

1. إنشاء حزمة في Cloud Storage

يتطلّب Datastream وجهة لتخزين أحداث التغيير التي تمّت مراقبتها. لننشئ حزمة GCS.

export BUCKET_NAME="migration-${PROJECT_ID}-bucket"
gcloud storage buckets create gs://${BUCKET_NAME} --location=$REGION

2. إنشاء ملفات شخصية لعمليات ربط مصادر البيانات

نحتاج إلى ملفَي تعريف مختلفَين لاتصال مصدر MySQL (ملف واحد لكل جزء فعلي) وملف تعريف واحد لاتصال الوجهة في Cloud Storage.

الحصول على عناوين IP المصدر

أولاً، احصل على عناوين IP الخارجية لجهازَي Compute Engine الافتراضيَين وخزِّنها كمتغيرات بيئية:

export MYSQL_IP_1=$(gcloud compute instances describe mysql-physical-1 --zone=$ZONE --format='get(networkInterfaces[0].accessConfigs[0].natIP)')

export MYSQL_IP_2=$(gcloud compute instances describe mysql-physical-2 --zone=$ZONE --format='get(networkInterfaces[0].accessConfigs[0].natIP)')

إنشاء ملفات تعريف لعمليات الربط بمصدر البيانات (MySQL على Compute Engine)

أنشئ ملفات ربط Datastream باستخدام datastream_user الذي تم إنشاؤه سابقًا.

# Create Source Profile for Physical Shard 1
export SQL_CP_NAME_1="mysql-src-cp-1"
gcloud datastream connection-profiles create $SQL_CP_NAME_1 \
    --location=$REGION \
    --type=mysql \
    --mysql-hostname=$MYSQL_IP_1 \
    --mysql-port=3306 \
    --mysql-username=datastream_user \
    --mysql-password=complex_password_123 \
    --display-name="MySQL Source 1 (Physical Shard 1)"

# Create Source Profile for Physical Shard 2
export SQL_CP_NAME_2="mysql-src-cp-2"
gcloud datastream connection-profiles create $SQL_CP_NAME_2 \
    --location=$REGION \
    --type=mysql \
    --mysql-hostname=$MYSQL_IP_2 \
    --mysql-port=3306 \
    --mysql-username=datastream_user \
    --mysql-password=complex_password_123 \
    --display-name="MySQL Source 2 (Physical Shard 2)"

ملاحظة: يربط Datastream بهذه الأجهزة الافتراضية من خلال عناوين IP العامة الخاصة بها، وهو أمر مسموح به لأنّنا أضفنا 0.0.0.0/0 إلى قواعد جدار الحماية في وقت سابق. في بيئة التشغيل الفعلي، عليك إدراج نطاقات عناوين IP العامة المحدّدة في Datastream في القائمة المسموح بها.

إنشاء ملف تعريف لربط الوجهة (Cloud Storage):

يشير ذلك إلى جذر الحزمة التي تم إنشاؤها حديثًا.

export GCS_CP_NAME="gcs-dest-cp"
gcloud datastream connection-profiles create $GCS_CP_NAME \
    --location=$REGION \
    --type=google-cloud-storage \
    --bucket=$BUCKET_NAME \
    --root-path=/ \
    --display-name="GCS Destination" --force

3- إنشاء مصادر بيانات Datastream

سننشئ الآن تدفقَين لتغيير البيانات. سيلتقط البث 1 shard0_db وshard1_db. سيلتقط البث 2 shard2_db وshard3_db. تتم كتابة كلا المصدرَين في حزمة GCS نفسها بتنسيق Avro.

# Stream for Physical Shard 1
export STREAM_NAME_1="mysql-to-spanner-stream-1"
export GCS_STREAM_PATH_1="data/${STREAM_NAME_1}"

gcloud datastream streams create $STREAM_NAME_1 \
    --location=$REGION \
    --display-name="MySQL Source 1 CDC Stream" \
    --source=$SQL_CP_NAME_1 \
    --destination=$GCS_CP_NAME \
    --mysql-source-config=<(echo "includeObjects:
  mysqlDatabases:
  - database: 'shard0_db'
  - database: 'shard1_db'") \
    --gcs-destination-config=<(echo "path: ${GCS_STREAM_PATH_1}/
fileRotationMb: 5
fileRotationInterval: 15s
avroFileFormat: {}") \
    --backfill-none

# Stream for Physical Shard 2
export STREAM_NAME_2="mysql-to-spanner-stream-2"
export GCS_STREAM_PATH_2="data/${STREAM_NAME_2}"

gcloud datastream streams create $STREAM_NAME_2 \
    --location=$REGION \
    --display-name="MySQL Source 2 CDC Stream" \
    --source=$SQL_CP_NAME_2 \
    --destination=$GCS_CP_NAME \
    --mysql-source-config=<(echo "includeObjects:
  mysqlDatabases:
  - database: 'shard2_db'
  - database: 'shard3_db'") \
    --gcs-destination-config=<(echo "path: ${GCS_STREAM_PATH_2}/
fileRotationMb: 5
fileRotationInterval: 15s
avroFileFormat: {}") \
    --backfill-none

يساعدنا استخدام إعدادات أصغر لتدوير الملفات (5 ميغابايت أو 15 ثانية) في رصد التغييرات المكرّرة بشكل أسرع أثناء درس تطبيقي حول الترميز.

قد يستغرق إكمال هذا الأمر بعض الوقت. التحقّق من الحالة: gcloud datastream streams describe $STREAM_NAME_1 --location=$REGION

4. بدء مصادر بيانات Datastream

فعِّل كلا المصدرَين لبدء تسجيل التغييرات.

gcloud datastream streams update $STREAM_NAME_1 \
    --location=$REGION \
    --state=RUNNING

gcloud datastream streams update $STREAM_NAME_2 \
    --location=$REGION \
    --state=RUNNING

التحقّق من الحالة: يمكنك تنفيذ gcloud datastream streams describe $STREAM_NAME_1 --location=$REGION. ستكون الحالة في البداية STARTING وستتغيّر إلى RUNNING بعد بضع لحظات. انتظِر إلى أن يتم تشغيلهما بالكامل قبل بدء عملية النقل المباشر.

5- إعداد Pub/Sub لإشعارات GCS

يجب إعلام Dataflow على الفور عندما يكتب أي من مصادر بيانات Datastream ملفًا جديدًا في حزمة GCS. سنضبط GCS لإرسال الإشعارات إلى موضوع واحد في Pub/Sub.

إنشاء موضوع Pub/Sub:

export PUBSUB_TOPIC="datastream-gcs-updates"
gcloud pubsub topics create $PUBSUB_TOPIC

إنشاء إشعار GCS

إرسال إشعار إلى الموضوع عند إنشاء أي عنصر ضمن البادئة data/ (التي تغطي كلا المصدرين)

gcloud storage buckets notifications create gs://${BUCKET_NAME} --topic=projects/$PROJECT_ID/topics/$PUBSUB_TOPIC --payload-format=json --object-prefix=data/

إنشاء اشتراك في Pub/Sub

أنشئ الاشتراك مع الموعد النهائي المقترَح للإقرار باستلام البيانات في Dataflow.

export PUBSUB_SUBSCRIPTION="datastream-gcs-sub"
gcloud pubsub subscriptions create $PUBSUB_SUBSCRIPTION \
  --topic=$PUBSUB_TOPIC \
  --ack-deadline=600

8. التحويل المخصّص

بما أنّ مخطط Spanner يختلف عن مخطط MySQL (بسبب الأعمدة التي أضفناها وأزلناها من خلال واجهة مستخدم الويب الخاصة بأداة SMT)، ستتعذّر عملية نقل البيانات الجاهزة في Dataflow. تحتاج Dataflow إلى تعليمات حول كيفية ربط هذه الاختلافات أثناء خطوط النقل الأمامية (من MySQL إلى Spanner) والعكسية (من Spanner إلى MySQL).

بالإضافة إلى ذلك، بما أنّنا نُجري عملية نقل عكسي مجزّأ، تحتاج Dataflow إلى آلية توجيه لمعرفة الجزء المنطقي (shard0_db أو shard1_db أو غير ذلك) الذي ينتمي إليه صف Spanner المعدَّل أثناء عملية النسخ المتماثل العكسي.

سنحقّق ذلك من خلال كتابة ملف Custom Transformation JAR باستخدام نموذج Spanner Custom Shard الذي توفّره Google.

1. تنزيل نموذج التقسيم المخصّص

في Cloud Shell، نزِّل مستودع نماذج Google Cloud Dataflow وانتقِل إلى مجلد الأجزاء المخصّصة:

git clone https://github.com/GoogleCloudPlatform/DataflowTemplates.git
cd DataflowTemplates/v2/spanner-custom-shard

2. ضبط منطق تحويل البيانات

علينا تعديل ملف CustomTransformationFetcher.java.

• نقل البيانات إلى إصدار أحدث (toSpannerRow): يتم ملء العمود OrderSource الذي تمت إضافته حديثًا باستخدام العمود LegacyOrderSystem من MySQL.
• عملية النقل العكسي (toSourceRow): تعيد ملء العمود LegacyOrderSystem الذي تم إسقاطه والذي يتطلبه MySQL، وذلك من خلال استخراجه من OrderSource في Spanner.

عدِّل الملف CustomTransformationFetcher.java. بدلاً من فتح محرّر نصوص يدويًا، نفِّذ الأمر التالي للكتابة تلقائيًا فوق ملف النموذج باستخدام منطقنا المخصّص:

cat << 'EOF' > src/main/java/com/custom/CustomTransformationFetcher.java 
package com.custom;

import com.google.cloud.teleport.v2.spanner.exceptions.InvalidTransformationException;
import com.google.cloud.teleport.v2.spanner.utils.ISpannerMigrationTransformer;
import com.google.cloud.teleport.v2.spanner.utils.MigrationTransformationRequest;
import com.google.cloud.teleport.v2.spanner.utils.MigrationTransformationResponse;
import java.util.HashMap;
import java.util.Map;

public class CustomTransformationFetcher implements ISpannerMigrationTransformer {

 @Override
 public void init(String customParameters) {}

 @Override
 public MigrationTransformationResponse toSpannerRow(MigrationTransformationRequest request)
     throws InvalidTransformationException {
   if (request.getTableName().equals("Orders")) {
     Map<String, Object> requestRow = request.getRequestRow();
     Map<String, Object> responseRow = new HashMap<>();

     Object legacySysObj = requestRow.get("LegacyOrderSystem");
     String legacySys = (legacySysObj != null) ? (String) legacySysObj : "UNKNOWN_SYSTEM";

     // Transform: Trim the string to remove everything after the first underscore
     String orderSource = legacySys;
     if (legacySys.contains("_")) {
       orderSource = legacySys.substring(0, legacySys.indexOf('_'));
     }

     // Populate the new Spanner column (e.g., "WebStore_v1" becomes "WebStore")
     responseRow.put("OrderSource", orderSource);

     return new MigrationTransformationResponse(responseRow, false);
   }

   return new MigrationTransformationResponse(new HashMap<>(), false);
 }

 @Override
 public MigrationTransformationResponse toSourceRow(MigrationTransformationRequest request)
     throws InvalidTransformationException {
   if (request.getTableName().equals("Orders")) {
     Map<String, Object> requestRow = request.getRequestRow();
     Map<String, Object> responseRow = new HashMap<>();

     // Safely fetch the Spanner OrderSource
     Object sourceObj = requestRow.get("OrderSource");
     String source = (sourceObj != null) ? (String) sourceObj : "UNKNOWN_SYSTEM";
     String legacySys = "'" + source + "_v1'";

     // Transform: Append a suffix to visibly prove the reverse transformation worked
     // e.g., "WebStore" becomes "WebStore_v1"
     responseRow.put("LegacyOrderSystem", legacySys);

     return new MigrationTransformationResponse(responseRow, false);
   }

   return new MigrationTransformationResponse(new HashMap<>(), false);
 }

 @Override
 public MigrationTransformationResponse transformFailedSpannerMutation(
     MigrationTransformationRequest request) throws InvalidTransformationException {
   return new MigrationTransformationResponse(new HashMap<>(), false);
 }
}
EOF

يُستخدَم التحويل المخصّص بشكل أساسي لتحويل القيم للأعمدة المتوفّرة مسبقًا أو لتعبئة الأعمدة الجديدة. يتم التعامل مع منطق الأعمدة التي تمت إعادة تسميتها أو إزالتها ضمنيًا، ولا تحتاج إلى تحويل مخصّص. لذلك، في الدالة أعلاه، لا نتعامل بشكل صريح مع العمود LegacyRegion الذي تمت إعادة تسميته في الجدول Customers.

3- ضبط منطق التقسيم العكسي

تستخدم Dataflow CustomShardIdFetcher.java أثناء النسخ المتماثل العكسي لتحديد المكان الذي يجب توجيه تغيير Spanner إليه. سنستخدم المفتاح الأساسي CustomerId ومنطق القسمة (%4) لتوجيه السجلات ديناميكيًا إلى الجزء المنطقي الصحيح.

عدِّل الملف CustomShardIdFetcher.java باستخدام cat واستبدِل محتواه بالكامل بالرمز التالي:

cat << 'EOF' > src/main/java/com/custom/CustomShardIdFetcher.java 
package com.custom;

import com.google.cloud.teleport.v2.spanner.utils.IShardIdFetcher;
import com.google.cloud.teleport.v2.spanner.utils.ShardIdRequest;
import com.google.cloud.teleport.v2.spanner.utils.ShardIdResponse;
import java.util.Map;

public class CustomShardIdFetcher implements IShardIdFetcher {
    
    @Override
    public void init(String parameters) {}

    @Override
    public ShardIdResponse getShardId(ShardIdRequest shardIdRequest) {
        Map<String, Object> keys = shardIdRequest.getSpannerRecord
();
        
        // Use the Primary Key to identify the correct logical shard
        if (keys != null && keys.containsKey("CustomerId")) {
            long customerId = Long.parseLong(keys.get("CustomerId").toString());
            long shardIdx = customerId % 4;
            
            ShardIdResponse response = new ShardIdResponse();
            response.setLogicalShardId("shard" + shardIdx + "_db");
            return response;
        }
        
        return new ShardIdResponse();
    }
}
EOF

4. إنشاء ملف JAR وتحميله

بعد كتابة منطق Java المخصّص، علينا تجميعه في ملف JAR وتحميله إلى حزمة Google Cloud Storage التي أنشأناها سابقًا ليتمكّن Dataflow من الوصول إليه.

نفِّذ الأوامر التالية في Cloud Shell:

# Return to DataflowTemplates directory 
cd ../..

# Build the JAR using Maven
mvn clean install -DskipTests -Dcheckstyle.skip -Dspotless.check.skip=true -Djib.skip -pl v2/spanner-custom-shard -am

# Upload the JAR to GCS
export CUSTOM_JAR_PATH="gs://${BUCKET_NAME}/custom-logic/spanner-custom-shard-1.0.jar"

gcloud storage cp v2/spanner-custom-shard/target/spanner-custom-shard-1.0-SNAPSHOT.jar $CUSTOM_JAR_PATH

# Return to home directory
cd ~

9. نقل البيانات بشكل مجمّع من MySQL إلى Spanner

بعد إعداد مخطط Spanner وإنشاء ملف JAR الخاص بعملية التحويل المخصّصة، يمكننا الآن نسخ البيانات الحالية من قاعدة بيانات MySQL إلى Cloud Spanner. ستستخدم Sourcedb to Spanner Dataflow Flex Template، وهو مصمّم لنسخ البيانات بشكل مجمّع من قواعد البيانات التي يمكن الوصول إليها من خلال JDBC إلى Spanner.

1. تحميل ملف عمليات إلغاء المخطط

في القسم 6، نزّلت ملف JSON الخاص بتجاوزات Spanner باستخدام واجهة مستخدم SMT على الويب. علينا تحميل هذا الملف إلى حزمة GCS لكي تتمكّن Dataflow من استخدامه لتحديد الاختلافات في المخطط (مثل الأعمدة التي تمت إعادة تسميتها).

1. في Cloud Shell، انقر على قائمة الخيارات الإضافية (المزيد) واختَر تحميل.

2. اختَر ملف JSON الخاص بعمليات الإلغاء الذي نزّلته سابقًا (مثل spanner_overrides.json).
3. انقل الملف إلى حزمة GCS باتّباع الخطوات التالية:

export OVERRIDES_FILE="spanner_overrides.json" # Change this if your downloaded file has a different name

export GCS_OVERRIDES_PATH="gs://${BUCKET_NAME}/config/${OVERRIDES_FILE}"

gcloud storage cp ~/${OVERRIDES_FILE} $GCS_OVERRIDES_PATH

2. إنشاء ملف إعداد التقسيم وتحميله

يجب أن تعرف خدمة Dataflow كيفية الاتصال بجميع الأقسام المنطقية الأربعة على مستوى الجهازَين الظاهريَين الفعليَين. سننشئ ملف sharding.json لهذا الغرض.

نفِّذ ما يلي في Cloud Shell لإنشاء الإعدادات وتحميلها:

cat <<EOF > sharding.json
{
  "configType": "dataflow",
  "shardConfigurationBulk": {
    "schemaSource": {
      "dataShardId": "mysql-physical-1",
      "host": "${MYSQL_IP_1}",
      "user": "datastream_user",
      "password": "complex_password_123",
      "port": "3306",
      "dbName": "shard0_db"
    },
    "dataShards": [
      {
        "dataShardId": "mysql-physical-1",
        "host": "${MYSQL_IP_1}",
        "user": "datastream_user",
        "password": "complex_password_123",
        "port": "3306",
        "dbName": "",
        "namespace": "namespace-mysql-1",
        "databases": [
          {
            "dbName": "shard0_db",
            "databaseId": "shard0_db",
            "refDataShardId": "mysql-physical-1"
          },
          {
            "dbName": "shard1_db",
            "databaseId": "shard1_db",
            "refDataShardId": "mysql-physical-1"
          }
        ]
      },
      {
        "dataShardId": "mysql-physical-2",
        "host": "${MYSQL_IP_2}",
        "user": "datastream_user",
        "password": "complex_password_123",
        "port": "3306",
        "dbName": "",
        "namespace": "namespace-mysql-2",
        "databases": [
          {
            "dbName": "shard2_db",
            "databaseId": "shard2_db",
            "refDataShardId": "mysql-physical-2"
          },
          {
            "dbName": "shard3_db",
            "databaseId": "shard3_db",
            "refDataShardId": "mysql-physical-2"
          }
        ]
      }
    ]
  }
}
EOF


export GCS_SHARDING_PATH="gs://${BUCKET_NAME}/config/sharding.json"
gcloud storage cp sharding.json $GCS_SHARDING_PATH

3- تشغيل مهمة Bulk Migration Dataflow

سنستخدم نموذج Flex Sourcedb إلى Spanner. بما أنّ عملية النقل هذه مجزّأة وتتضمّن عمليات تحويل مخصّصة، فإنّنا نمرّر ملف Overrides وإعدادات التجزئة وملف Java JAR المخصّص.

export JOB_NAME="mysql-sharded-bulk-to-spanner-$(date +%Y%m%d-%H%M%S)"
export OUTPUT_DIR="gs://${BUCKET_NAME}/bulk-migration"

gcloud dataflow flex-template run $JOB_NAME \
  --project=$PROJECT_ID \
  --region=$REGION \
--template-file-gcs-location="gs://dataflow-templates-${REGION}/latest/flex/Sourcedb_to_Spanner_Flex" \
--max-workers=2 \
--num-workers=1 \
--worker-machine-type=n2-highmem-8 \
  --parameters \
sourceConfigURL=$GCS_SHARDING_PATH,\
instanceId=$SPANNER_INSTANCE_NAME,\
databaseId=$SPANNER_DATABASE_NAME,\
projectId=$PROJECT_ID,\
outputDirectory=$OUTPUT_DIR,\
username=datastream_user,\
password=complex_password_123,\
schemaOverridesFilePath=$GCS_OVERRIDES_PATH,\
transformationJarPath=$CUSTOM_JAR_PATH,\
transformationClassName=com.custom.CustomTransformationFetcher

شرح المَعلمات الرئيسية:

• sourceConfigURL: مسار ملف sharding.json الذي أنشأناه يخبر هذا الإعداد Dataflow بكيفية الاتصال بجميع أقسام MySQL المنطقية الأربعة على مستوى الآلتين الافتراضيتين الماديتين.
• schemaOverridesFilePath: مسار ملف JSON الذي نزّلناه من واجهة مستخدم الويب الخاصة بأداة SMT يُعلِم هذا الأمر خدمة Dataflow بكيفية التعامل مع تعديلات المخطط التي أجريناها (مثل عمود LegacyRegion الذي تم إسقاطه وقيود التحقّق الأكثر صرامة).
• transformationJarPath: مسار GCS إلى ملف Java JAR المجمَّع الذي أنشأناه في القسم السابق. يحتوي هذا الملف على الرمز البرمجي الفعلي لتنفيذ عمليات التحويل المخصّصة.
• ‫transformationClassName: الاسم المؤهَّل بالكامل لفئة Java داخل ملف JAR الذي ينفّذ منطق النقل إلى الإصدار الأحدث (com.custom.CustomTransformationFetcher).
• استبدِل outputDirectory بموقع GCS الذي ستكتب فيه خدمة Dataflow ملفاتها المؤقتة، والأهم من ذلك، ملفات قائمة الرسائل غير المسلَّمة (DLQ).
• استبدِل maxWorkers بـ numWorkers: يتحكّم في توسيع نطاق مهمة Dataflow. تم إبقاء القيمة منخفضة لمجموعة البيانات الصغيرة هذه.
• instanceId وdatabaseId وprojectId: تحدّد مثيل Cloud Spanner وقاعدة البيانات المستهدَفين.

ملاحظة بشأن الشبكة: ترتبط هذه المهمة بمثيل Cloud SQL عبر عنوان IP العلني. يمكن إجراء ذلك لأنّك أضفت 0.0.0.0/0 سابقًا إلى "الشبكات المسموح بها" في الجهاز الظاهري. يتيح ذلك لآلات Dataflow الافتراضية العاملة التي تتضمّن عناوين IP خارجية الوصول إلى قاعدة البيانات.

4. مراقبة مهمة Dataflow

يمكنك تتبُّع مدى تقدّم المهمة في Google Cloud Console:

1. انتقِل إلى صفحة "مهام Dataflow": الانتقال إلى "مهام Dataflow"
2. ابحث عن الوظيفة المسماة mysql-sharded-bulk-to-spanner-... وانقر عليها.
3. مراقبة الرسم البياني للوظائف والمقاييس انتظِر إلى أن تتغيّر حالة المهمة إلى تم بنجاح. من المفترض أن يستغرق ذلك بين 5 و15 دقيقة تقريبًا.

• إذا واجهت المهمة مشاكل، راجِع علامة التبويب السجلات ضِمن صفحة تفاصيل مهمة Dataflow بحثًا عن رسائل الخطأ.
• تقدّم مقاييس المهام المزيد من المعلومات حول تقدّم المهمة واستهلاك الموارد، مثل سرعة معالجة البيانات واستخدام وحدة المعالجة المركزية.

يمكنك الاطّلاع على دليل رصد أكثر شمولاً هنا.

5- التحقّق من البيانات في Cloud Spanner وفحص قائمة انتظار الرسائل غير القابلة للتسليم (DLQ)

بعد اكتمال مهمة Dataflow بنجاح، علينا التأكّد من وصول بياناتنا بأمان وفحص السجلات التي صمّمناها عمدًا لتعطيلها.

ما هي قائمة انتظار الرسائل غير الصالحة؟ قائمة انتظار الرسائل غير الصالحة (DLQ) هي دليل تخزين توجّه فيه خدمة Dataflow بأمان السجلات التي يتعذّر كتابتها في Spanner. بدلاً من إيقاف مسار نقل البيانات بأكمله، ينقل Dataflow الصفوف غير الصالحة إلى قائمة انتظار الرسائل غير المعالَجة (كملفات JSON في outputDirectory) لتتمكّن من فحصها وإصلاحها وإعادة محاولة معالجتها لاحقًا.

أ. التحقّق من الصحة العامة للبيانات التي تم نقلها:

استخدِم واجهة سطر الأوامر gcloud لتنفيذ بعض عمليات التحقّق السريعة من سلامة قاعدة بيانات Spanner الموحّدة للتأكّد من نقل السجلات الصالحة بشكل صحيح وملء ملف JAR المخصّص للعمود الإضافي.

# 1. Verify total Customer count
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT COUNT(*) as TotalCustomers FROM Customers"

# 2. Verify total Orders count (Total minus the orphan record)
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT COUNT(*) as TotalOrders FROM Orders"

# 3. Verify the Custom Transformation on OrderSource worked
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT CustomerId, OrderId, OrderSource FROM Orders LIMIT 3"

# 4. Verify that renamed column LoyaltyTier has the correct data
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT CustomerId, CustomerName, LoyaltyTier FROM Customers LIMIT 3"

الناتج المتوقّع:

TotalCustomers: 16
TotalOrders: 19

CustomerId: 1
OrderId: 201
OrderSource: MobileApp

CustomerId: 2
OrderId: 301
OrderSource: CallCenter

CustomerId: 3
OrderId: 401
OrderSource: InStore

CustomerId: 1
CustomerName: Agnes N.
LoyaltyTier: NORTHEAST

CustomerId: 2
CustomerName: Brian K.
LoyaltyTier: SOUTHWEST

CustomerId: 3
CustomerName: Cathy Z.
LoyaltyTier: CENTRAL

• تم نقل جميع الصفوف في جدول "العملاء" بنجاح.
• نلاحظ تعذُّر عرض صف واحد في جدول Orders بسبب INTERLEAVE IN PARENT على Spanner - CustomerId 99 هو عنصر ثانوي غير مرتبط بسبب عدم توفّر صف مطابق في جدول Customers.

ب. التحقّق من حالات الفشل المقصودة في قائمة انتظار الرسائل غير الصالحة:

تم توثيق الخطأ أعلاه في مجلد "قائمة الرسائل غير المرسَلة" (DLQ) الذي أنشأته عملية نقل البيانات المجمّعة.

1. انتقِل إلى Cloud Storage في Google Cloud Console.
2. انتقِل إلى الحزمة وافتح المجلد bulk-migration/dlq/severe.
3. افحص ملفات JSON بداخلها. ستجد الصف Orders الذي يحتوي على CustomerId اليتيم.
4. يمكن إعادة محاولة إصلاح أخطاء قائمة انتظار الرسائل غير الصالحة لعملية النقل المجمّع باتّباع الخطوات المذكورة هنا.

اكتملت الآن عملية التحميل المجمّع الأوّلي للبيانات من Cloud SQL إلى Cloud Spanner. الخطوة التالية هي إعداد عملية النسخ المتماثل المباشر لتسجيل التغييرات الجارية.

10. بدء عملية النقل المباشر (تغيير البيانات)

بعد اكتمال عملية تحميل البيانات المجمّعة، ستطلق مهمة بث مستمر في Dataflow. ستقرأ هذه المهمة أحداث "تسجيل بيانات التغيير" (CDC) التي يكتبها Datastream في حزمة GCS، وستطبّق هذه التغييرات على Cloud Spanner في الوقت الفعلي تقريبًا.

سنجرّب أيضًا مسار النقل هذا من خلال إدخال بيانات صالحة وغير صالحة عمدًا لمراقبة طريقة تعامل Dataflow مع النسخ المتماثل المباشر وتوجيه الأخطاء إلى قائمة الرسائل غير المرسَلة (DLQ).

1. إنشاء ملف إعداد تقسيم البيانات في عملية النقل المباشر

على عكس عملية النقل المجمّع (التي تستخدم سلاسل اتصال JDBC)، يقرأ مسار النقل المباشر أحداث Datastream من GCS. يتطلّب ذلك إعدادًا مختلفًا تمامًا بتنسيق JSON يربط بين أسماء مصادر البيانات وقواعد البيانات في Datastream وبين أقسام Spanner المنطقية.

نفِّذ ما يلي في Cloud Shell لإنشاء إعدادات التقسيم المباشر وتحميلها:

cat <<EOF > live-sharding.json
{
  "StreamToDbAndShardMap": {
    "${STREAM_NAME_1}": {
      "shard0_db": "shard0_db",
      "shard1_db": "shard1_db"
    },
    "${STREAM_NAME_2}": {
      "shard2_db": "shard2_db",
      "shard3_db": "shard3_db"
    }
  }
}
EOF

export GCS_LIVE_SHARDING_PATH="gs://${BUCKET_NAME}/config/live-sharding.json"
gcloud storage cp live-sharding.json $GCS_LIVE_SHARDING_PATH

2. تشغيل مهمة Dataflow لنقل البيانات المباشر

ابدأ مهمة Dataflow لبث البيانات من أجل القراءة من GCS والكتابة إلى Spanner. سيستخدم هذا النموذج إشعارات GCS Pub/Sub لمعالجة الملفات الجديدة على الفور.

export JOB_NAME_CDC="mysql-sharded-cdc-to-spanner-$(date +%Y%m%d-%H%M%S)"
export DLQ_DIR_CDC="gs://${BUCKET_NAME}/live-migration"

gcloud dataflow flex-template run $JOB_NAME_CDC \
  --project=$PROJECT_ID \
  --region=$REGION \
--worker-machine-type=n2-highmem-8 \
--template-file-gcs-location="gs://dataflow-templates-${REGION}/latest/flex/Cloud_Datastream_to_Spanner" \
  --parameters \
instanceId="$SPANNER_INSTANCE_NAME",\
databaseId="$SPANNER_DATABASE_NAME",\
projectId="$PROJECT_ID",\
inputFileFormat="avro",\
gcsPubSubSubscription="projects/${PROJECT_ID}/subscriptions/${PUBSUB_SUBSCRIPTION}",\
shardingContextFilePath=$GCS_LIVE_SHARDING_PATH,\
deadLetterQueueDirectory="$DLQ_DIR_CDC",\
transformationJarPath=$CUSTOM_JAR_PATH,\
transformationClassName="com.custom.CustomTransformationFetcher",\
schemaOverridesFilePath=$GCS_OVERRIDES_PATH,\
datastreamSourceType="mysql",\
dlqRetryMinutes=1,\
dlqMaxRetryCount=2

المَعلمات الرئيسية

• ‫gcsPubSubSubscription: اشتراك Pub/Sub الذي يستمع إلى إشعارات الملفات الجديدة من GCS. يتيح ذلك للمهمة معالجة التغييرات على الفور أثناء كتابة Datastream لها.
• inputFileFormat="avro": يطلب من Dataflow توقّع ملفات Avro من Datastream. يجب أن يتطابق هذا مع إعدادات "الوجهة" في مصدر البيانات (مثل avroFileFormat مقابل jsonFileFormat).
• shardingContextFilePath: ملف JSON يربط بين مصادر بيانات Datastream وأجزاء منطقية.
• dlqRetryMinutes: عدد الدقائق بين عمليات إعادة المحاولة في قائمة انتظار الرسائل غير القابلة للتسليم القيمة التلقائية هي 10.
• ‫dlqMaxRetryCount: الحد الأقصى لعدد المرات التي يمكن فيها إعادة محاولة الأخطاء المؤقتة من خلال قائمة انتظار الرسائل غير المعالَجة. القيمة التلقائية هي 500.

يمكنك مراقبة بدء تشغيل المهمة في وحدة تحكّم مهام Dataflow.

3- إدخال بيانات مباشرة وتفعيل حالات تعذُّر مقصودة

أثناء بدء تشغيل مهمة البث في Dataflow (قد يستغرق ذلك من 3 إلى 5 دقائق)، لننتقل إلى أول جهاز افتراضي فعلي يعمل بنظام MySQL عبر SSH ونُدرج بعض السجلات الجديدة. سندرج سجلاً صالحًا وسجلاً غير صالح.

استخدِم بروتوكول SSH للوصول إلى الجزء الأول من قاعدة البيانات:

gcloud compute ssh mysql-physical-1 --zone=$ZONE

سجِّل الدخول إلى MySQL:

sudo mysql

نفِّذ عمليات الإدراج التالية على shard1_db:

USE shard1_db;

-- 1. Valid Insert: 'MobileApp_v2' will be trimmed to 'MobileApp'
INSERT INTO Orders (CustomerId, OrderId, OrderValue, LegacyOrderSystem) 
VALUES (4, 501, 99.99, 'MobileApp_v2');

-- 2. Invalid Insert (DLQ Test): This violates Interleave constraint as CustomerId 99999 doesn't exist in Customers table.
INSERT INTO Orders (CustomerId, OrderId, OrderValue, LegacyOrderSystem) 
VALUES (99999, 502, 50.00, 'WebStore_v1');

-- 3. Valid Update
UPDATE Orders SET OrderValue = '1500' WHERE CustomerId = 5 AND OrderId = 202; 

-- 4. Valid Delete
DELETE FROM Orders WHERE CustomerId = 5 AND OrderId = 203; 

EXIT;

اكتب exit مرة أخرى للرجوع إلى طلب Cloud Shell.

على عكس عملية النقل المُجمَّع، هذه مهمة بث. ستظلّ حالته Running إلى أن يتم إلغاؤه يدويًا لأنّه يستمع باستمرار إلى التغييرات الجديدة. بعد تشغيلها، يمكنك الانتقال فورًا إلى الخطوة التالية.

4. التحقّق من بيانات "نقل البيانات المباشر" وفحص قائمة انتظار الرسائل غير المعالَجة في CDC

بعد إدخال البيانات، ستسجّل خدمة Datastream أحداث التقاط البيانات المتغيرة (CDC)، وستحاول خدمة Dataflow تطبيقها على Spanner.

أ. التحقّق من صحة تغييرات لغة معالجة البيانات (DML) في Spanner

نفِّذ طلبات البحث التالية للتأكّد من أنّ أحداث INSERT وUPDATE وDELETE قد وصلت بنجاح إلى Spanner، وأنّ عملية التحويل المخصّص تم تنفيذها عند كلّ من الإدراج والتعديل.

# 1. Verify INSERT: Should return the new row with transformed OrderSource
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT CustomerId, OrderId, OrderValue, OrderSource FROM Orders WHERE CustomerId = 4 AND OrderId = 501"

# 2. Verify UPDATE: Should show OrderValue changed to 1500
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT CustomerId, OrderId, OrderValue, OrderSource FROM Orders WHERE CustomerId = 5 AND OrderId = 202"

# 3. Verify DELETE: Should return 0, confirming the order was deleted
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT COUNT(*) FROM Orders WHERE CustomerId = 5 AND OrderId = 203"

# 4. Verify DLQ Failure: Should return 0, confirming the row migration failed
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT COUNT(*) FROM Orders WHERE CustomerId = 99999 AND OrderId = 502"

الناتج المتوقّع:

CustomerId: 4
OrderId: 501
OrderValue: 99.99
OrderSource: MobileApp

CustomerId: 5
OrderId: 202
OrderValue: 1500
OrderSource: WebStore

0
0

ملاحظة: إذا لم يعرض أي طلب البحث النتيجة المتوقّعة، انتظِر دقيقة واحدة وأعِد المحاولة، لأنّ العاملين في البث قد يكونون لا يزالون بصدد معالجة قائمة الانتظار.

ب. التحقّق من الخطأ المقصود في قائمة انتظار الرسائل غير الصالحة:

بما أنّ CustomerId = 99999 ليس لديه عنصر رئيسي في جدول Customers، كان من المفترض أن يرفضه Spanner وأن يتم توجيهه بأمان إلى قائمة انتظار الرسائل غير الصالحة من خلال Dataflow.

1. انتقِل إلى Cloud Storage في Google Cloud Console.
2. انتقِل إلى الحزمة وافتح المجلد live-migration/dlq/severe/.
3. من المفترض أن تظهر لك ملفات JSON تم إنشاؤها حديثًا. انقر عليها لفحص المحتوى. ستظهر لك تفاصيل CustomerId = 99999 ورسالة خطأ Spanner المحدّدة: NOT_FOUND: Parent row for row [99999,502,WebStore] in table Orders is missing. Row cannot be written."
4. يمكن إعادة محاولة أخطاء قائمة انتظار الرسائل غير المعالَجة في عملية النقل المباشر من خلال تشغيل نموذج تدفق البيانات مع ضبط runMode=retryDLQ.

5- معالجة أخطاء قائمة انتظار الرسائل غير الصالحة

تتطلّب الأخطاء في دليل severe/ تدخّلاً يدويًا. لنحلّ مشكلة البيانات ونعيد معالجة الحدث الذي تعذّر تنفيذه.

أ. إصلاح البيانات في المصدر

حدث الخطأ لأنّ سجلّ العميل الرئيسي CustomerId = 99999 غير متوفّر. لنُدرِجها في قاعدة بيانات MySQL المصدر.

استخدِم بروتوكول SSH للدخول إلى مثيل MySQL مرة أخرى:

gcloud compute ssh mysql-physical-1 --zone=$ZONE

سجِّل الدخول إلى MySQL باستخدام sudo mysql وأدرِج صف العنصر الرئيسي الناقص في shard1_db:

USE shard1_db;

INSERT INTO Customers (CustomerId, CustomerName, CreditLimit, LegacyRegion) VALUES
(99999, 'DLQ Parent Holder', 5000.00, 'NORTH_AMERICA');

EXIT;

اكتب exit للرجوع إلى Cloud Shell.

ب. تشغيل مهمة Dataflow الخاصة بإعادة المحاولة في قائمة انتظار الرسائل غير الصالحة

لإعادة معالجة الأحداث من severe/ قائمة انتظار الرسائل غير الصالحة، عليك تشغيل نموذج Dataflow نفسه ولكن في وضع retryDLQ. يقرأ هذا الوضع تحديدًا من المسار deadLetterQueueDirectory/severe، ويعيد تشغيلها من خلال عمليات التحويل المخصّصة، ويطبّقها على Spanner.

ابدأ المهمة في وضع retryDLQ:

export JOB_NAME_RETRY="mysql-sharded-cdc-retry-$(date +%Y%m%d-%H%M%S)"

gcloud dataflow flex-template run $JOB_NAME_RETRY \
  --project=$PROJECT_ID \
  --region=$REGION \
  --worker-machine-type=n2-highmem-8 \
--template-file-gcs-location="gs://dataflow-templates-${REGION}/latest/flex/Cloud_Datastream_to_Spanner" \
  --parameters \
instanceId="$SPANNER_INSTANCE_NAME",\
databaseId="$SPANNER_DATABASE_NAME",\
projectId="$PROJECT_ID",\
runMode="retryDLQ",\
deadLetterQueueDirectory="$DLQ_DIR_CDC",\
datastreamSourceType="mysql",\
transformationJarPath=$CUSTOM_JAR_PATH,\
transformationClassName="com.custom.CustomTransformationFetcher",\
schemaOverridesFilePath=$GCS_OVERRIDES_PATH,\
shardingContextFilePath=$GCS_LIVE_SHARDING_PATH

التغييرات الرئيسية في المَعلمات لإعادة المحاولة

• ‫runMode="retryDLQ": يطلب من النموذج القراءة من دليل severe DLQ.
• تمت إزالة gcsPubSubSubscription: لا حاجة إليه لأنّنا لا نقرأ من حزمة GCS المباشرة في Datastream.

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

مثل مسار CDC الرئيسي، retryDLQ هو مسار بث سيبقى RUNNING إلى أن يتم إلغاؤه يدويًا.

1. انتقِل إلى صفحة "مهمة Dataflow" الخاصة بـ $JOB_NAME_RETRY.
2. ضِمن لوحة المقاييس، ابحث عن العدادَين التاليَين:
3. ‫elementsReconsumedFromDeadLetterQueue: يتم تقييمها عند جلب ملفات الخطأ.
4. ‫Successful events: يتم زيادته عند كتابة السجلّ في Spanner.
5. تحقَّق من الدليل severe/ بحثًا عن حالات تعذُّر متكرّرة.
6. بعد أن يزداد عدد أحداث "ناجح" بمقدار عدد العناصر التي أردت إعادة محاولة إرسالها (1 في حالة الاختبار)، انتقِل إلى خطوة التحقّق التالية.

ج. التحقّق من البيانات التي تمت إعادة محاولة إرسالها

بعد إعادة محاولة السجلّ الذي تعذّر نقله (قد يستغرق ذلك بعض الوقت)، تحقَّق من Spanner لمعرفة ما إذا تم نقل صف التابع بنجاح:

gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="SELECT CustomerId, OrderId, OrderValue, OrderSource FROM Orders WHERE CustomerId = 99999 AND OrderId = 502"

من المفترض أن يظهر لك الصف الآن:

CustomerId: 99999
OrderId: 502
OrderValue: 50
OrderSource: WebStore

تحقَّق أيضًا من المجلد $DLQ_DIR_CDC/severe/ في GCS. من المفترض أن تكون الملفات التي تمت معالجتها قد تم نقلها أو حذفها، ما يشير إلى إعادة المعالجة بنجاح.

11. إعداد النسخ المتماثل العكسي (من Spanner إلى MySQL)

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

تستخدم مسار البيانات هذا "قنوات تغيير البيانات في Spanner" لتسجيل التعديلات المباشرة في Spanner. بعد ذلك، تستخدم هذه الخدمة ملف Custom Transformation JAR لعكس عملية الربط بين المخططين، وملف Custom Sharding JAR لتحديد الجهاز الظاهري (VM) الفعلي لقاعدة بيانات MySQL والجزء المنطقي اللذين يجب إعادة كتابة التعديل فيهما.

1. إنشاء "بث تغييرات" في Spanner

عليك أولاً إنشاء مصدر تغيير في قاعدة بيانات Spanner لتتبُّع التغييرات في الجدولَين Customers وOrders.

export CHANGE_STREAM_NAME="CustomersOrdersChangeStream"

gcloud spanner databases ddl update $SPANNER_DATABASE_NAME \
  --instance=$SPANNER_INSTANCE_NAME \
  --ddl="CREATE CHANGE STREAM $CHANGE_STREAM_NAME FOR Customers, Orders"

سيسجّل مصدر تغيير البيانات هذا الآن جميع تعديلات البيانات في الجداول المحدّدة.

2. إنشاء قاعدة بيانات Spanner لبيانات وصفية في Dataflow

يتطلّب نموذج Spanner to SourceDB Dataflow قاعدة بيانات Spanner منفصلة لتخزين بيانات التعريف من أجل إدارة استخدام "دفق التغيير".

export SPANNER_METADATA_DB_NAME="migration-metadata-db"

gcloud spanner databases create $SPANNER_METADATA_DB_NAME \
  --instance=$SPANNER_INSTANCE_NAME

3- إعداد إعدادات اتصال Cloud SQL لخدمة Dataflow

يحتاج نموذج Dataflow إلى ملف JSON في Cloud Storage يحتوي على تفاصيل الاتصال بقاعدة بيانات Cloud SQL المستهدَفة.

أنشئ ملفًا محليًا باسم shard_config.json:

cat <<EOF > reverse-sharding.json
[
  {
    "logicalShardId": "shard0_db",
    "host": "${MYSQL_IP_1}",
    "port": "3306",
    "user": "datastream_user",
    "password": "complex_password_123",
    "dbName": "shard0_db"
  },
  {
    "logicalShardId": "shard1_db",
    "host": "${MYSQL_IP_1}",
    "port": "3306",
    "user": "datastream_user",
    "password": "complex_password_123",
    "dbName": "shard1_db"
  },
  {
    "logicalShardId": "shard2_db",
    "host": "${MYSQL_IP_2}",
    "port": "3306",
    "user": "datastream_user",
    "password": "complex_password_123",
    "dbName": "shard2_db"
  },
  {
    "logicalShardId": "shard3_db",
    "host": "${MYSQL_IP_2}",
    "port": "3306",
    "user": "datastream_user",
    "password": "complex_password_123",
    "dbName": "shard3_db"
  }
]
EOF

حمِّل هذا الملف إلى حزمة GCS:

export GCS_REVERSE_SHARDING_PATH="gs://${BUCKET_NAME}/config/reverse-sharding.json"
gcloud storage cp reverse-sharding.json $GCS_REVERSE_SHARDING_PATH

4. تشغيل مهمة Reverse Replication Dataflow

شغِّل مهمة Dataflow باستخدام Spanner_to_SourceDb Flex Template.

export JOB_NAME_REVERSE="spanner-sharded-reverse-to-mysql-$(date +%Y%m%d-%H%M%S)"
export DLQ_DIR_REVERSE="gs://${BUCKET_NAME}/reverse-replication"

gcloud dataflow flex-template run $JOB_NAME_REVERSE \
  --project=$PROJECT_ID \
  --region=$REGION \
--worker-machine-type=n2-highmem-8 \
--max-workers=2 \
--num-workers=1 \
--additional-experiments=use_runner_v2 \
--template-file-gcs-location="gs://dataflow-templates-${REGION}/latest/flex/Spanner_to_SourceDb" \
  --parameters \
changeStreamName="$CHANGE_STREAM_NAME",\
instanceId="$SPANNER_INSTANCE_NAME",\
databaseId="$SPANNER_DATABASE_NAME",\
spannerProjectId="$PROJECT_ID",\
metadataInstance="$SPANNER_INSTANCE_NAME",\
metadataDatabase="$SPANNER_METADATA_DB_NAME",\
sourceShardsFilePath="$GCS_REVERSE_SHARDING_PATH",\
transformationJarPath=$CUSTOM_JAR_PATH,\
transformationClassName="com.custom.CustomTransformationFetcher",\
shardingCustomJarPath=$CUSTOM_JAR_PATH,\
shardingCustomClassName="com.custom.CustomShardIdFetcher",\
schemaOverridesFilePath=$GCS_OVERRIDES_PATH,\
deadLetterQueueDirectory=$DLQ_DIR_REVERSE

المَعلمات الرئيسية

• changeStreamName: اسم "دفق تغييرات" Spanner الذي سيتم القراءة منه.
• ‫metadataInstance, metadataDatabase: مثيل/قاعدة بيانات Spanner لتخزين البيانات الوصفية التي يستخدمها الموصل للتحكّم في استهلاك بيانات واجهة برمجة التطبيقات الخاصة بدفق البيانات المتغيرة
• sourceShardsFilePath: مسار GCS إلى shard_config.json
• ‫filtrationMode: تحدّد هذه السمة كيفية حذف سجلات معيّنة استنادًا إلى معيار معيّن. القيمة التلقائية هي forward_migration (فلترة السجلات التي تمت كتابتها باستخدام مسار نقل البيانات إلى الأمام)
• ‫shardingCustomJarPath: مسار GCS إلى ملف JAR المجمَّع في Java الذي أنشأناه سابقًا.
• shardingCustomClassName: اسم الفئة المؤهَّل بالكامل (com.custom.CustomShardIdFetcher) الذي ينفّذ عملية حسابية مخصّصة %4 modulo لتحديد الجزء المنطقي الذي يجب أن يتلقّى السجلّ بشكلٍ ديناميكي.

ملاحظة بشأن الشبكة: ستربط العاملات في Dataflow بمثيل Cloud SQL باستخدام عنوان IP العلني المحدّد في shard_config.json. يُسمح بهذا الاتصال بسبب الإدخال 0.0.0.0/0 في "الشبكات المصرّح بها" الخاصة بمثيل Cloud SQL.

يمكنك مراقبة بدء تشغيل المهمة في وحدة تحكّم مهام Dataflow.

5- إدخال بيانات Spanner وتفعيل حالات تعذُّر مقصودة

انتظِر إلى أن تنتقل مهمة Dataflow إلى الحالة Running (قد يستغرق ذلك حوالي 5 دقائق). بعد ذلك، لننفّذ مجموعة كاملة من طلبات البحث (INSERT وUPDATE وDELETE) مباشرةً في Spanner، بالإضافة إلى حدوث خطأ مقصود لاختبار قائمة انتظار الرسائل غير المعالَجة العكسية.

نفِّذ ما يلي في Cloud Shell:

# All these operations are done on rows mapping to shard0_db for convenience

# Valid INSERT: Insert parent row in Customers
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="INSERT INTO Customers (CustomerId, CustomerName, CreditLimit, LoyaltyTier) VALUES (88, 'Reverse Tester', 5000, 'GOLD_TIER')"

# 1. Valid INSERT (Orders): 'WebStore' transformed to 'WebStore_v1'
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="INSERT INTO Orders (CustomerId, OrderId, OrderValue, OrderSource) VALUES (88, 9001, 150.00, 'WebStore')"

# 2. Valid UPDATE (Orders)
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="UPDATE Orders SET OrderValue = 200.00 WHERE CustomerId = 16 AND OrderId = 105 AND OrderSource = 'Partner'"

# 3. Valid DELETE (Orders)
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="DELETE FROM Orders WHERE CustomerId = 12 AND OrderId = 104 AND OrderSource = 'WebStore'"

# 4. INVALID Insert- DLQ Test: CreditLimit=500 will fail check constraint of >1000 at source 
gcloud spanner databases execute-sql $SPANNER_DATABASE_NAME \
    --instance=$SPANNER_INSTANCE_NAME \
    --sql="INSERT INTO Customers (CustomerId, CustomerName, CreditLimit, LoyaltyTier) VALUES (44, 'DLQ Test Customer', 500, 'GOLD_TIER')"

6. التحقّق من بيانات النسخ المتماثل العكسي وفحص قائمة انتظار الرسائل غير الصالحة

لنؤكّد أنّ ملف Custom Sharding JAR قد وجّه CustomerId 88 بنجاح إلى shard0_db على أول جهاز افتراضي فعلي، وأنّ ملف Custom Transformation JAR قد أزال "_TIER" بنجاح من المنطقة.

أ. تحقَّق من السجلّ الصالح في MySQL:

استخدِم بروتوكول SSH للوصول إلى الجزء الأول من قاعدة البيانات:

gcloud compute ssh mysql-physical-1 --zone=$ZONE

سجِّل الدخول إلى MySQL وأدخِل طلب البحث shard0_db:

sudo mysql
USE shard0_db;

-- 1. Verify INSERT: Row migrated with transformed LegacyOrderSystem
SELECT CustomerId, OrderId, OrderValue, LegacyOrderSystem 
FROM Orders 
WHERE CustomerId = 88 AND OrderId = 9001;

-- 2. Verify UPDATE: The OrderValue should now be updated to 200.00.
SELECT CustomerId, OrderId, OrderValue, LegacyOrderSystem 
FROM Orders 
WHERE CustomerId = 16 AND OrderId = 105;

-- 3. Verify DELETE: Returns 0 rows, confirming the order was successfully deleted from MySQL.
SELECT CustomerId, OrderId 
FROM Orders 
WHERE CustomerId = 12 AND OrderId = 104;

-- 4. Verify failed replication - this should be in DLQ as CreditLimit < 1000 and will fail stricter check constraint at source 
SELECT CustomerId, CustomerName, CreditLimit, LegacyRegion
FROM Customers
WHERE CustomerId = 44;

EXIT;

يجب أن يعكس الناتج المتوقّع في Cloud SQL التغييرات التي تم إجراؤها في Spanner.

+------------+---------+------------+-------------------+
| CustomerId | OrderId | OrderValue | LegacyOrderSystem |
+------------+---------+------------+-------------------+
|         88 |    9001 |     150.00 | Webstore_v1       |
+------------+---------+------------+-------------------+

+------------+---------+------------+-------------------+
| CustomerId | OrderId | OrderValue | LegacyOrderSystem |
+------------+---------+------------+-------------------+
|         16 |     105 |     200.00 | Partner_v1        |
+------------+---------+------------+-------------------+

Empty set (0.00 sec)
Empty set (0.00 sec)

النوع

exit

للرجوع إلى Cloud Shell

يؤكّد ذلك أنّ مسار النسخ المتماثل العكسي يعمل، ويتم مزامنة التغييرات من Spanner إلى Cloud SQL.

ب. التحقّق من الخطأ المقصود في قائمة انتظار الرسائل غير الصالحة

بما أنّ سجلّ Customers الجديد يتضمّن قيمة CreditLimit تبلغ 500 (وهو ما يخالف قيد التحقّق الصارم > 1000 الذي حدّدناه في قاعدة بيانات MySQL المصدر)، رصدت خدمة Dataflow الخطأ بأمان.

1. انتقِل إلى Cloud Storage في Google Cloud Console.
2. انتقِل إلى الحزمة وافتح المجلد dlq/severe/.
3. افتح ملف JSON للاطّلاع على السجلّ Customers المرفوض وخطأ انتهاك قيود التحقّق.
4. يمكن إعادة محاولة إصلاح أخطاء قائمة انتظار الرسائل غير الصالحة في عملية النسخ المتماثل العكسي من خلال تشغيل نموذج Dataflow مع ضبط runMode=retryDLQ.

12. تنظيف الموارد

لتجنُّب تكبُّد رسوم إضافية على حسابك على Google Cloud، احذف الموارد التي تم إنشاؤها أثناء استخدام هذا الدرس التطبيقي حول الترميز.

ضبط متغيرات البيئة (إذا لزم الأمر)

إذا انتهت مهلة جلسة Cloud Shell أو فتحت نافذة طرفية جديدة، عليك إعادة تصدير متغيرات البيئة قبل تنفيذ أوامر التنظيف.

export PROJECT_ID=$(gcloud config get-value project)
export REGION="us-central1" # Or your preferred region
export ZONE="us-central1-a" # Or a zone within your selected region
export SPANNER_INSTANCE_NAME="target-spanner-instance"
export SPANNER_DATABASE_NAME="sharded-target-db"
export SPANNER_CONFIG="regional-${REGION}"
export BUCKET_NAME="migration-${PROJECT_ID}-bucket"
export MYSQL_IP_1=$(gcloud compute instances describe mysql-physical-1 --zone=$ZONE --format='get(networkInterfaces[0].accessConfigs[0].natIP)')
export MYSQL_IP_2=$(gcloud compute instances describe mysql-physical-2 --zone=$ZONE --format='get(networkInterfaces[0].accessConfigs[0].natIP)')
export SQL_CP_NAME_1="mysql-src-cp-1"
export SQL_CP_NAME_2="mysql-src-cp-2"
export GCS_CP_NAME="gcs-dest-cp"
export STREAM_NAME_1="mysql-to-spanner-stream-1"
export GCS_STREAM_PATH_1="data/${STREAM_NAME_1}"
export STREAM_NAME_2="mysql-to-spanner-stream-2"
export GCS_STREAM_PATH_2="data/${STREAM_NAME_2}"
export PUBSUB_TOPIC="datastream-gcs-updates"
export PUBSUB_SUBSCRIPTION="datastream-gcs-sub"
export CUSTOM_JAR_PATH="gs://${BUCKET_NAME}/custom-logic/spanner-custom-shard-1.0.jar"
export OVERRIDES_FILE="spanner_overrides.json" 
export GCS_OVERRIDES_PATH="gs://${BUCKET_NAME}/config/${OVERRIDES_FILE}"
export GCS_SHARDING_PATH="gs://${BUCKET_NAME}/config/sharding.json"
export OUTPUT_DIR="gs://${BUCKET_NAME}/bulk-migration"
export GCS_LIVE_SHARDING_PATH="gs://${BUCKET_NAME}/config/live-sharding.json"
export DLQ_DIR_CDC="gs://${BUCKET_NAME}/live-migration"
export CHANGE_STREAM_NAME="CustomersOrdersChangeStream"
export SPANNER_METADATA_DB_NAME="migration-metadata-db"
export GCS_REVERSE_SHARDING_PATH="gs://${BUCKET_NAME}/config/reverse-sharding.json"
export DLQ_DIR_REVERSE="gs://${BUCKET_NAME}/reverse-replication"

إيقاف مهام البث في Dataflow

أدرِج وظائفك للعثور على معرّفات الوظائف الخاصة بوظائف تدفّق البيانات قيد التشغيل. صدِّر JOB_ID_CDC وJOB_ID_REVERSE وفقًا لذلك.

gcloud dataflow jobs list --region=$REGION --filter="state=Running"
export JOB_ID_CDC=<PASTE_JOB_ID_HERE>
export JOB_ID_CDC_RETRY=<PASTE_JOB_ID_HERE>
export JOB_ID_REVERSE=<PASTE_JOB_ID_HERE>

إلغاء مهمة Datastream to Spanner (النقل المباشر) ومهمة إعادة المحاولة:

gcloud dataflow jobs cancel $JOB_ID_CDC --region=$REGION --project=$PROJECT_ID

gcloud dataflow jobs cancel $JOB_ID_CDC_RETRY --region=$REGION --project=$PROJECT_ID

ألغِ مهمة Spanner to Cloud SQL (النسخ المتماثل العكسي) باتّباع الخطوات التالية:

gcloud dataflow jobs cancel $JOB_ID_REVERSE --region=$REGION --project=$PROJECT_ID

حذف موارد Datastream

إيقاف البث وحذفه:

gcloud datastream streams update $STREAM_NAME_1 \
  --location=$REGION --state=PAUSED --project=$PROJECT_ID
# Wait a moment for the stream to pause
gcloud datastream streams delete $STREAM_NAME_1 \
  --location=$REGION --project=$PROJECT_ID --quiet

gcloud datastream streams update $STREAM_NAME_2 \
  --location=$REGION --state=PAUSED --project=$PROJECT_ID
# Wait a moment for the stream to pause
gcloud datastream streams delete $STREAM_NAME_2 \
  --location=$REGION --project=$PROJECT_ID --quiet

# Delete Connection Profiles
gcloud datastream connection-profiles delete $SQL_CP_NAME_1 \
  --location=$REGION --project=$PROJECT_ID --quiet
gcloud datastream connection-profiles delete $SQL_CP_NAME_2 \
  --location=$REGION --project=$PROJECT_ID --quiet
gcloud datastream connection-profiles delete $GCS_CP_NAME \
  --location=$REGION --project=$PROJECT_ID --quiet

حذف الأجهزة الافتراضية المصدر التي تعمل بنظام MySQL (Compute Engine)

احذف مثيلَي Compute Engine اللذين يحاكيان الأقسام المادية لـ MySQL المحلية.

gcloud compute instances delete mysql-physical-1 mysql-physical-2 --zone=$ZONE --quiet

حذف قواعد جدار الحماية

أزِل قواعد جدار الحماية للشبكة التي تم إنشاؤها للسماح بالوصول إلى بروتوكول النقل الآمن (SSH) وإمكانية اتصال Datastream بأجهزتك الافتراضية. (ملاحظة: إذا كنت قد استخدمت أسماء مختلفة لقواعد جدار الحماية في وقت سابق من الدرس العملي، يمكنك تعديلها هنا).

gcloud compute firewall-rules delete allow-ssh-iap --quiet
gcloud compute firewall-rules delete allow-mysql-datastream --quiet

حذف موارد Pub/Sub

حذف الاشتراك:

gcloud pubsub subscriptions delete $PUBSUB_SUBSCRIPTION \
  --project=$PROJECT_ID --quiet

حذف الموضوع:

gcloud pubsub topics delete $PUBSUB_TOPIC \
  --project=$PROJECT_ID --quiet

حذف مثيل Cloud Spanner

احذف مثيل Cloud Spanner (سيؤدي ذلك تلقائيًا إلى حذف كل من قاعدة بيانات sharded-target-db وقاعدة بيانات migration-metadata-db بداخله).

gcloud spanner instances delete $SPANNER_INSTANCE_NAME \
  --project=$PROJECT_ID --quiet

حذف حزمة GCS ومحتواها

أخيرًا، احذف حزمة Cloud Storage التي تحتوي على ملفات Datastream وإعدادات Dataflow وقوائم الرسائل غير المسلَّمة. يحذف الأمر rm -r الحزمة وجميع محتوياتها بشكل متكرر.

ملاحظة: إذا كنت تخطّط لفحص السجلات التي تعذّر إرسالها أو تحديد المشاكل فيها أو إعادة محاولة إرسالها في قوائم انتظار الرسائل غير المرسَلة (DLQ) لاحقًا، لا تحذف هذه الحزمة بعد.

gcloud storage rm --recursive gs://${BUCKET_NAME}

حذف ملفات Cloud Shell المحلية

لتنظيف الملفات والأدلة المحلية التي تم إنشاؤها في Cloud Shell أثناء هذا الدرس التطبيقي حول الترميز، نفِّذ الأوامر التالية:

# Remove the JSON configuration files
rm -f sharding.json live-sharding.json reverse-sharding.json spanner_overrides.json

# Remove the cloned Google Cloud DataflowTemplates repository
rm -rf DataflowTemplates