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 layanan kesehatan berformat CSV ke BigQuery secara real time. Kita akan menggunakan pipeline Data Real-Time fusion Cloud Data 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, mengubah, dan menyamarkan data layanan kesehatan secara real time.

Apa yang Anda perlukan untuk menjalankan demo ini?

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

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

Data layanan 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 dimuat sebelumnya dapat ditemukan di gs://hcls_testing_data_fhir_10_patients/csv_schemas.

Jika membutuhkan set data baru, Anda selalu dapat membuatnya menggunakan SyntheaTM. Kemudian, upload ke GCS, bukan menyalinnya dari bucket pada langkah Salin data input.

Referensi:

Operasi Dasar di Cloud Data Fusion (Panduan Memulai) menggunakan Google Cloud Platform Console.

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

2. Penyiapan Project GCP

Melakukan inisialisasi variabel shell untuk lingkungan Anda.

Untuk menemukan PROJECT_ID, lihat Identifying projects.

<!-- 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 izin bergabung.
2. Anda akan menerima email berisi petunjuk tentang cara mengonfirmasi tindakan tersebut.
3. Gunakan opsi untuk membalas email agar bergabung dengan grup. JANGAN klik tombol .
4. Setelah menerima email konfirmasi, Anda dapat melanjutkan ke langkah berikutnya di codelab.

Menyalin data input.

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

Membuat 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 lakukan inisialisasi Google Cloud SDK, lalu 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 GCP Console.
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 GCP Console, pilih ProjectID Anda.
2. Pilih Data Fusion dari menu kiri, lalu klik tombol CREATE AN INSTANCE di tengah halaman (pembuatan pertama), atau klik tombol CREATE INSTANCE di menu atas (kreasi tambahan).

3. Berikan nama instance. Pilih Enterprise.

4. Klik tombol CREATE.

Catatan: Diperlukan waktu hingga 30 menit untuk menyelesaikan proses pembuatan instance.

Menyiapkan izin instance.

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

Catatan: Pemberian izin hanya perlu dilakukan satu kali dalam suatu 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 Agen Layanan Cloud Data Fusion API kepada akun layanan dengan mengklik tombol Tambahkan. Tempel "akun layanan" di kolom Anggota baru dan pilih Pengelolaan Layanan -> Peran 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 -> Jaringan VPC -> Aturan firewall untuk memeriksa apakah aturan default-allow-ssh ada atau tidak.

2. Jika tidak, tambahkan aturan firewall yang mengizinkan semua traffic masuk SSH 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 Create Firewall Rule dan isi informasinya:

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 View Instance di kolom Action. Anda akan dialihkan ke halaman lain. Klik url yang diberikan untuk membuka instance Cloud Data Fusion. Pilihan Anda untuk mengklik "Mulai Tur" atau "Lain Kali" pada {i>Welcome pop-up<i}.
2. Luaskan "hamburger" menu, pilih Pipeline -> Daftar

3. Klik tombol + hijau di sudut kanan atas, lalu pilih Create Pipeline. Atau klik "Buat" {i>link pipeline.<i}

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, Analytics, Sink, Pengendali Error, dan Pemberitahuan tempat Anda dapat memilih node atau node untuk pipeline.

Pilih Source node.

1. Pada 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 = {teks apa pun}
• Nama referensi = {any text}
• ID Project = deteksi otomatis
• Langganan = Langganan yang dibuat di bagian Buat Topik Pub/Sub (misalnya, sub-Anda)
• Topik = Topik yang dibuat di bagian Buat Topik Pub/Sub (misalnya, topik-anda)

4. Klik Dokumentasi untuk mendapatkan penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. Hijau "Tidak ditemukan error" menunjukkan keberhasilan.

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

Pilih Transform node.

1. Pada bagian Transform 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 Projection.
2. Arahkan kursor ke node Projection, lalu klik Properties.

3. Isi bidang yang diperlukan. Tetapkan kolom berikut:

• Konversi = konversi pesan dari jenis byte ke jenis string.
• Kolom yang akan dihapus = {kolom mana pun}
• Kolom yang perlu disimpan = {pesan, stempel waktu, dan atribut} (misalnya, atribut: key=‘filename':value=‘pasien' yang dikirim dari Pub/Sub)
• Kolom untuk diganti namanya = {message, timestamp}

4. Klik Dokumentasi untuk mendapatkan penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. Hijau "Tidak ditemukan error" menunjukkan keberhasilan.

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

6. Klik drop-down Tindakan dan pilih Impor untuk mengimpor skema yang tersimpan (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 mencentang 'Null' .
8. Isi bidang yang diperlukan. Tetapkan kolom berikut:

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

9. Klik Dokumentasi untuk mendapatkan penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. Hijau "Tidak ditemukan error" menunjukkan keberhasilan.

10. Tetapkan nama kolom dalam urutan yang lebih disukai, dan lepaskan kolom yang tidak diperlukan. Salin cuplikan kode berikut dan tempelkan di kotak Recipe.

drop attributes
parse-as-csv :body ',' false
drop body
set columns TIMESTAMP,Id,BIRTHDATE,DEATHDATE,SSN,DRIVERS,PASSPORT,PREFIX,FIRST,LAST,SUFFIX,MAIDEN,MARITAL,RACE,ETHNICITY,GENDER,BIRTHPLACE,ADDRESS,CITY,STATE,ZIP
mask-number SSN xxxxxxx####

Catatan: Salah berarti menguraikan kolom isi dengan "Tetapkan baris pertama sebagai header" tidak dicentang.

11. Lihat Batch-Codelab - CSV ke BigQuery via CDF untuk penyamaran dan de-identifikasi data. Atau tambahkan cuplikan kode ini mask-number SSN xxxxxxx#### di kotak Recipe
12. Untuk menutup jendela Transform Properties, klik tombol X.

Pilih node Sink.

1. Di bagian Sink pada 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. Lengkapi kolom yang wajib diisi:

• Label = {teks apa pun}
• Nama referensi = {any text}
• ID Project = deteksi otomatis
• Set data = Set data BigQuery yang digunakan dalam project saat ini (misalnya, DATASET_ID)
• Tabel = {nama tabel}

4. Klik Dokumentasi untuk mendapatkan penjelasan mendetail. Klik tombol Validasi untuk memvalidasi semua informasi input. Hijau "Tidak ditemukan error" menunjukkan keberhasilan.

5. Untuk menutup Properti BigQuery, klik tombol X.

5. Membuat pipeline data Real-time

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

Menghubungkan semua node dalam pipeline

1. Tarik panah koneksi > di tepi kanan {i>node<i} sumber, lalu letakkan di tepi kiri {i>node<i} tujuan.
2. Pipeline dapat memiliki beberapa cabang yang menerima pesan yang dipublikasikan dari node Sumber PubSub yang sama.

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

3. Beri nama pipeline tersebut.

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

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

Mengirim pesan melalui Cloud Pub/Sub

Menggunakan UI Pub/Sub:

1. Buka konsol GCP -> Pub/Sub -> Topik, pilih topik Anda, lalu klik PUBLIKASIKAN PESAN di menu atas.

2. Tempatkan hanya satu baris data dalam satu waktu di kolom Message. Klik tombol +ADD AN ATTRIBUTE. Berikan Kunci = filename, Nilai = <type of record> (misalnya, pasien, penyedia layanan kesehatan, alergi, dll.).
3. Klik tombol Publikasikan untuk mengirimkan 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. Semi-otomatis menyediakan pesan 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

&lt;key&gt; - namafile

cat - membaca konten file

sed - menyembunyikan output (-n) dan secara eksplisit mencetak (p) baris spesifik #(1-?). 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. Gunakan opsi default Configure.
2. Klik Preview untuk melihat pratinjau data**.** Klik **Preview** lagi untuk beralih kembali ke jendela sebelumnya. Anda juga dapat menjalankan pipeline dalam mode Pratinjau dengan mengklik **RUN**.

3. Klik Log untuk melihat log.
4. Klik Save untuk menyimpan semua perubahan.
5. Klik Import untuk mengimpor konfigurasi pipeline yang tersimpan saat membangun pipeline baru.
6. Klik Export untuk mengekspor konfigurasi pipeline.
7. Klik Deploy untuk men-deploy pipeline.
8. Setelah di-deploy, klik Run dan tunggu hingga pipeline berjalan hingga selesai.

Catatan: Saat Anda menjalankan pipeline, Cloud Data Fusion akan menyediakan cluster Cloud Dataproc sementara, menjalankan pipeline, lalu membongkar cluster. Tindakan ini memerlukan waktu beberapa menit. Anda dapat mengamati status transisi pipeline dari Provisioning ke Starting ke Running ke Mencabut Akses ke Succeeded selama jangka waktu ini.

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

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

7. Validasi

Di bagian ini, kita akan memvalidasi eksekusi pipeline data.

1. Memvalidasi bahwa pipeline berhasil dieksekusi dan berjalan secara berkelanjutan.

2. Validasi bahwa tabel BigQuery dimuat dengan kumpulan data yang diperbarui berdasarkan TIMESTAMP. Dalam contoh ini, dua catatan atau pesan pasien dan satu catatan atau pesan 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 <your-sub> pelanggan.

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

Melihat hasil

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

1. Buat kueri tabel di UI BigQuery. BUKA UI BIGQUERY
2. Perbarui kueri di bawah ini dengan memasukkan nama project, set data, dan tabel Anda.

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 agar tidak menghabiskan kuota dan Anda tidak akan ditagih di masa mendatang. Bagian berikut menjelaskan cara menghapus atau menonaktifkan resource tersebut.

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 memiliki efek berikut:

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

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

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

9. Selamat

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

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

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

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