Menyerap data CSV (Comma-separated values) ke BigQuery menggunakan Cloud Data Fusion - Penyerapan real time

1. Pengantar

Terakhir Diperbarui: 28-02-2020

Codelab ini menunjukkan pola penyerapan data untuk menyerap data kesehatan berformat CSV ke BigQuery secara real time. Kita akan menggunakan pipeline Data Real-time Cloud Data Fusion untuk lab ini. Data pengujian layanan kesehatan yang realistis telah dibuat dan tersedia di bucket Google Cloud Storage (gs://hcls_testing_data_fhir_10_patients/csv/) untuk Anda.

Dalam codelab ini, Anda akan mempelajari:

• Cara menyerap data CSV (pemuatan real time) dari Pub/Sub ke BigQuery menggunakan Cloud Data Fusion.
• Cara membangun pipeline integrasi data secara visual di Cloud Data Fusion untuk memuat, mentransformasi, dan memasking data healthcare secara real time.

Apa yang Anda butuhkan untuk menjalankan demo ini?

• Anda memerlukan akses ke project GCP.
• Anda harus diberi peran Pemilik untuk project GCP.
• Data layanan kesehatan dalam format CSV, termasuk header.

Jika Anda tidak memiliki project GCP, ikuti langkah-langkah ini untuk membuat project GCP baru.

Data kesehatan dalam format CSV telah dimuat sebelumnya ke dalam bucket GCS di gs://hcls_testing_data_fhir_10_patients/csv/. Setiap file resource CSV memiliki struktur skema yang unik. Misalnya, Patients.csv memiliki skema yang berbeda dengan Providers.csv. File skema yang telah dimuat sebelumnya dapat ditemukan di gs://hcls_testing_data_fhir_10_patients/csv_schemas.

Jika memerlukan set data baru, Anda dapat membuatnya kapan saja menggunakan SyntheaTM. Kemudian, upload ke GCS, bukan menyalinnya dari bucket pada langkah Copy input data.

Referensi:

Operasi Dasar di Cloud Data Fusion (Panduan memulai) menggunakan Konsol Google Cloud Platform.

Memublikasikan Pesan Cloud Pub/Sub https://cloud.google.com/pubsub/docs/publisher.

2. Penyiapan Project GCP

Lakukan inisialisasi variabel shell untuk lingkungan Anda.

Untuk menemukan PROJECT_ID, lihat Mengidentifikasi project.

<!-- 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>

Buat bucket GCS untuk menyimpan data input dan log error menggunakan alat gsutil.

gsutil mb -l us gs://$BUCKET_NAME

Dapatkan akses ke set data sintetis.

1. Dari alamat email yang Anda gunakan untuk login ke Konsol Cloud, kirim email ke [e-mail] untuk meminta bergabung.
2. Anda akan menerima email berisi petunjuk tentang cara mengonfirmasi tindakan tersebut.
3. Gunakan opsi untuk membalas email guna bergabung dengan grup. JANGAN klik tombol .
4. Setelah menerima email konfirmasi, Anda dapat melanjutkan ke langkah berikutnya dalam codelab.

Menyalin data input.

gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME

Buat set data BigQuery.

bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID

Ingat: Anda selalu dapat membuat kumpulan data pengujian baru yang realistis menggunakan SyntheaTM untuk lab ini. Untuk memudahkan Anda, kami menyediakan set data di Google Cloud Storage berikut. gs://hcls_testing_data_fhir_10_patients/csv.

Instal dan inisialisasi Google Cloud SDK serta buat Topik dan Langganan Pub atau Sub.

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

3. Penyiapan Lingkungan Cloud Data Fusion

Ikuti langkah-langkah berikut untuk mengaktifkan Cloud Data Fusion API dan memberikan izin yang diperlukan:

Aktifkan API.

1. Buka Library API Konsol GCP.
2. Dari daftar project, pilih project Anda.
3. Di Library API, pilih API yang ingin Anda aktifkan ( Cloud Data Fusion API, Cloud Pub/Sub API). Jika Anda memerlukan bantuan untuk menemukan API, gunakan kolom penelusuran dan filter.
4. Di halaman API, klik ENABLE.

Buat instance Cloud Data Fusion.

1. Di Konsol GCP, pilih ProjectID Anda.
2. Pilih Penggabungan Data dari menu kiri, lalu klik tombol BUAT INSTANS di tengah halaman (pembuatan pertama), atau klik tombol BUAT INSTANS di menu atas (pembuatan tambahan).

3. Berikan nama instance. Pilih Enterprise.

4. Klik tombol BUAT.

Catatan: Proses pembuatan instance dapat memerlukan waktu hingga 30 menit untuk selesai.

Siapkan izin instance.

Setelah membuat instance, gunakan langkah-langkah berikut untuk memberikan izin akun layanan yang terkait dengan instance pada project Anda:

Catatan: Pemberian izin hanya perlu dilakukan satu kali dalam project.

1. Buka halaman detail instance dengan mengklik nama instance.

2. Salin akun layanan.

3. Buka halaman IAM project Anda.
4. Di halaman izin IAM, berikan peran Cloud Data Fusion API Service Agent kepada akun layanan dengan mengklik tombol Add. Tempelkan "akun layanan" di kolom Anggota baru, lalu pilih peran Service Management -> Cloud Data Fusion API Server Agent.

5. Klik + Add another role (atau Edit Cloud Data Fusion API Service Agent) untuk menambahkan peran Pub/Sub Subscriber.

6. Klik Simpan.

Setelah langkah-langkah ini selesai, Anda dapat mulai menggunakan Cloud Data Fusion dengan mengklik link Lihat Instance di halaman instance Cloud Data Fusion, atau halaman detail instance.

Siapkan aturan firewall.

1. Buka GCP Console -> VPC Network -> Firewall rules untuk memeriksa apakah aturan default-allow-ssh ada atau tidak.

2. Jika tidak, tambahkan aturan firewall yang mengizinkan semua traffic SSH masuk ke jaringan default.

Menggunakan command line:

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

Menggunakan UI: Klik Buat Aturan Firewall dan isi informasi:

4. Membangun node untuk pipeline

Setelah memiliki lingkungan Cloud Data Fusion di GCP, mari kita mulai membangun pipeline Data di Cloud Data Fusion menggunakan langkah-langkah berikut:

1. Di jendela Cloud Data Fusion, klik link Lihat Instance di kolom Tindakan. Anda akan dialihkan ke halaman lain. Klik URL yang diberikan untuk membuka instance Cloud Data Fusion. Pilihan Anda untuk mengklik tombol "Mulai Tur" atau "Tidak, Terima Kasih" di jendela pop-up Selamat Datang.
2. Luaskan menu "hamburger", pilih Pipeline -> List

3. Klik tombol hijau + di sudut kanan atas, lalu pilih Buat Pipeline. Atau klik "Buat" link pipeline.

4. Setelah studio pipeline muncul, di kiri atas, pilih Data Pipeline - Realtime dari dropdown.

5. Di UI Data Pipelines, Anda akan melihat berbagai bagian di panel kiri sebagai Filter, Sumber, Transformasi, Analisis, Tujuan, Penanganan Error, dan Pemberitahuan tempat Anda dapat memilih satu atau beberapa node untuk pipeline.

Pilih node Sumber.

1. Di bagian Source pada palet Plugin di sebelah kiri, klik dua kali node Google Cloud PubSub, yang muncul di UI Data Pipelines.
2. Arahkan kursor ke node sumber PubSub, lalu klik Properties.

3. Isi bidang yang diperlukan. Tetapkan kolom berikut:

• Label = {any text}
• Nama referensi = {any text}
• Project ID = deteksi otomatis
• Subscription = Langganan yang dibuat di bagian Membuat Topik Pub/Sub (misalnya, your-sub)
• Topic = Topik yang dibuat di bagian Create Pub/Sub Topic (misalnya, your-topic)

4. Klik Dokumentasi untuk penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. "Tidak ada error" berwarna hijau menunjukkan keberhasilan.

5. Untuk menutup Pub/Sub Properties, klik tombol X.

Pilih node Transform.

1. Di bagian Transformasi pada palet Plugin di sebelah kiri, klik dua kali node Projection, yang muncul di UI Data Pipelines. Hubungkan node sumber Pub/Sub ke node transformasi Proyeksi.
2. Arahkan kursor ke node Projection, lalu klik Properties.

3. Isi bidang yang diperlukan. Tetapkan kolom berikut:

• Convert = mengonversi message dari jenis byte menjadi jenis string.
• Kolom yang akan dilepas = {any field}
• Kolom yang akan dipertahankan = {message, timestamp, dan attributes} (misalnya, attributes: key=‘filename':value=‘patients' yang dikirim dari Pub/Sub)
• Kolom yang akan diganti namanya = {message, timestamp}

4. Klik Dokumentasi untuk penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. "Tidak ada error" berwarna hijau menunjukkan keberhasilan.

5. Di bagian Transformasi pada palet Plugin di sebelah kiri, klik dua kali pada node Wrangler, yang muncul di UI Data Pipelines. Hubungkan node transformasi Proyeksi ke node transformasi Wrangler. Arahkan kursor ke node Wrangler, lalu klik Properties.

6. Klik drop-down Tindakan, lalu pilih Impor untuk mengimpor skema yang disimpan (misalnya: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json).
7. Tambahkan kolom TIMESTAMP di Output Schema (jika tidak ada) dengan mengklik tombol + di samping kolom terakhir dan centang kotak'Null'.
8. Isi bidang yang diperlukan. Tetapkan kolom berikut:

• Label = {any text}
• Nama kolom input = {*}
• Precondition = {attributes.get("filename") != "patients"} untuk membedakan setiap jenis rekaman atau pesan (misalnya, pasien, penyedia, alergi, dll.) yang dikirim dari node sumber PubSub.

9. Klik Dokumentasi untuk penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. "Tidak ada error" berwarna hijau menunjukkan keberhasilan.

10. Tetapkan nama kolom dalam urutan yang diinginkan, dan hilangkan kolom yang tidak Anda perlukan. Salin cuplikan kode berikut dan tempelkan di kotak Resep.

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####

Catatan: Salah (False) berarti mengurai kolom isi dengan opsi "Tetapkan baris pertama sebagai header" tidak dicentang.

11. Lihat Batch-Codelab - CSV to BigQuery via CDF untuk Penyamaran dan penghapusan identitas data. Atau tambahkan cuplikan kode mask-number SSN xxxxxxx#### ini di kotak Resep
12. Untuk menutup jendela Transform Properties, klik tombol X.

Pilih node Sink.

1. Di bagian Sink di palet Plugin di sebelah kiri, klik dua kali node BigQuery, yang muncul di UI Data Pipeline. Hubungkan node transformasi Wrangler ke node sink BigQuery.
2. Arahkan kursor ke node sink BigQuery, lalu klik Properties.

3. Isi kolom yang wajib diisi:

• Label = {any text}
• Nama referensi = {any text}
• Project ID = deteksi otomatis
• Dataset = Set data BigQuery yang digunakan dalam project saat ini (misalnya, DATASET_ID)
• Table = {table name}

4. Klik Dokumentasi untuk penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. "Tidak ada error" berwarna hijau menunjukkan keberhasilan.

5. Untuk menutup Properti BigQuery, klik tombol X.

5. Membangun pipeline data Real time

Di bagian sebelumnya, kita membuat node yang diperlukan untuk membangun pipeline data di Cloud Data Fusion. Di bagian ini, kita akan menghubungkan node untuk membangun pipeline yang sebenarnya.

Menghubungkan semua node dalam pipeline

1. Tarik panah koneksi > di tepi kanan node sumber dan lepas di tepi kiri node tujuan.
2. Pipeline dapat memiliki beberapa cabang yang mendapatkan pesan yang dipublikasikan dari node Sumber PubSub yang sama.

Catatan: Bangun satu cabang dari node Proyeksi di pipeline untuk setiap pesan PubSub

3. Beri nama pipeline.

Catatan: Selama proses ini, skema akan otomatis disebarkan saat Anda membuat pipeline. Anda juga dapat mengimpor skema yang diinginkan.

Selesai. Anda baru saja membuat pipeline data real-time pertama untuk di-deploy dan dijalankan.

Mengirim pesan melalui Cloud Pub/Sub

Menggunakan UI Pub/Sub:

1. Buka konsol GCP -> Pub/Sub -> Topics, pilih your-topic, lalu klik PUBLISH MESSAGE di menu atas.

2. Tempatkan hanya satu baris data dalam satu waktu di kolom Pesan. Klik tombol +ADD AN ATTRIBUTE. Berikan Key = filename, Value = <type of record> (misalnya, pasien, penyedia, alergi, dll.).
3. Klik tombol Publikasikan untuk mengirim pesan.

Menggunakan perintah gcloud:

1. Berikan pesan secara manual.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"paste one record row here"

2. Menyediakan pesan secara semi-otomatis menggunakan perintah unix cat dan sed. Perintah ini dapat dijalankan berulang kali dengan parameter yang berbeda.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"$(gsutil cat gs://$BUCKET_NAME/csv/<value>.csv | sed -n '#p')"

Catatan: <your-topic> - sama dengan topik yang diberikan di properti node sumber Pub/Sub

<key> - filename

cat - membaca isi file

sed - menekan output (-n) dan mencetak (p) baris #(1-?) tertentu secara eksplisit. Jika baris header ada, gunakan '2p' dan seterusnya.

6. Mengonfigurasi, Men-deploy, dan Menjalankan Pipeline

Setelah mengembangkan pipeline data, kita dapat men-deploy dan menjalankannya di Cloud Data Fusion.

1. Pertahankan default Konfigurasi.
2. Klik Pratinjau untuk melihat pratinjau data**.** Klik **Pratinjau** lagi untuk beralih kembali ke jendela sebelumnya. Anda juga dapat menjalankan pipeline dalam mode Pratinjau dengan mengklik **RUN**.

3. Klik Logs untuk melihat log.
4. Klik Simpan untuk menyimpan semua perubahan.
5. Klik Impor untuk mengimpor konfigurasi pipeline yang disimpan saat membuat pipeline baru.
6. Klik Ekspor untuk mengekspor konfigurasi pipeline.
7. Klik Deploy untuk men-deploy pipeline.
8. Setelah di-deploy, klik Run dan tunggu hingga pipeline selesai berjalan.

Catatan: Saat Anda menjalankan pipeline, Cloud Data Fusion akan menyediakan cluster Cloud Dataproc efemeral, menjalankan pipeline, lalu menghentikan cluster. Tindakan ini memerlukan waktu beberapa menit. Anda dapat mengamati status transisi pipeline dari Penyediaan ke Memulai ke Menjalankan ke Membatalkan penyediaan ke Berhasil selama waktu ini.

9. Klik Stop untuk menghentikan operasi pipeline kapan saja.
10. Anda dapat menduplikasi pipeline dengan memilih Duplikasikan di tombol Tindakan.
11. Anda dapat mengekspor konfigurasi pipeline dengan memilih Ekspor di tombol Tindakan.

12. Klik Ringkasan untuk menampilkan diagram Histori operasi, catatan, log error, dan peringatan.

7. Validasi

Di bagian ini, kita akan memvalidasi eksekusi pipeline data.

1. Pastikan pipeline berhasil dieksekusi dan berjalan terus-menerus.

2. Validasi bahwa tabel BigQuery dimuat dengan data yang diperbarui berdasarkan TIMESTAMP. Dalam contoh ini, dua pesan atau catatan pasien dan satu pesan atau catatan alergi dipublikasikan ke topik Pub/Sub pada 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 |
+----------+-----------+

3. Validasi bahwa pesan yang dipublikasikan ke <your-topic> diterima oleh pelanggan <your-sub>.

gcloud pubsub subscriptions pull --auto-ack <your-sub>

Melihat hasil

Untuk melihat hasil setelah pesan dipublikasikan ke topik Pub/Sub saat pipeline Realtime berjalan:

1. Kueri tabel di UI BigQuery. BUKA UI BIGQUERY
2. Perbarui kueri di bawah ke nama project, set data, dan tabel Anda sendiri.

8. Pembersihan

Agar tidak menimbulkan biaya pada akun Google Cloud Platform Anda untuk resource yang digunakan dalam tutorial ini:

Setelah menyelesaikan tutorial, Anda dapat membersihkan resource yang dibuat di GCP sehingga resource tersebut tidak akan menghabiskan kuota dan Anda tidak akan ditagih di masa mendatang. Bagian berikut menjelaskan cara menghapus atau menonaktifkan resource ini.

Menghapus set data BigQuery

Ikuti petunjuk ini untuk menghapus set data BigQuery yang Anda buat sebagai bagian dari tutorial ini.

Menghapus Bucket GCS

Ikuti petunjuk ini untuk menghapus bucket GCS yang Anda buat sebagai bagian dari tutorial ini.

Menghapus instance Cloud Data Fusion

Ikuti petunjuk ini untuk menghapus instance Cloud Data Fusion Anda.

Menghapus project

Cara termudah untuk menghilangkan penagihan adalah dengan menghapus project yang Anda buat untuk tutorial.

Untuk menghapus project:

Perhatian: Menghapus project akan mengakibatkan:

• Semua hal dalam project akan dihapus. Jika menggunakan project yang sudah ada untuk tutorial ini, saat menghapusnya, Anda juga akan menghapus pekerjaan lain yang telah dilakukan dalam project tersebut.
• Project ID kustom akan hilang. Saat membuat project ini, Anda mungkin juga membuat project ID kustom yang masih ingin Anda gunakan pada masa mendatang. Agar tidak kehilangan URL yang menggunakan project ID tersebut, seperti URL appspot.com, hapus resource yang dipilih di dalam project, bukan menghapus seluruh project.

Jika Anda berencana mempelajari beberapa tutorial dan panduan memulai, menggunakan kembali project dapat membantu Anda agar tidak melampaui batas kuota project.

1. Di Konsol GCP, buka halaman Project. BUKA HALAMAN PROJECT
2. Dalam daftar project, pilih project yang ingin dihapus, lalu klik Delete.
3. Pada dialog, ketik project ID, lalu klik Shut down untuk menghapus project.

9. Selamat

Selamat, Anda telah berhasil menyelesaikan codelab untuk menyerap data kesehatan di BigQuery menggunakan Cloud Data Fusion.

Anda telah memublikasikan data CSV ke topik Pub/Sub, lalu memuatnya ke BigQuery.

Anda telah membangun pipeline integrasi data secara visual untuk memuat, mentransformasi, dan menyamarkan data layanan kesehatan secara real time.

Sekarang Anda mengetahui langkah-langkah utama yang diperlukan untuk memulai perjalanan Analisis Data Layanan Kesehatan dengan BigQuery di Google Cloud Platform.