Cloud Data Fusion'ı kullanarak CSV (virgülle ayrılmış değerler) verilerini BigQuery'ye aktarma: Gerçek zamanlı kullanım

1. Giriş

Son Güncelleme: 2020-02-28

Bu codelab, CSV biçimli sağlık verilerini BigQuery'ye gerçek zamanlı olarak almak için bir veri kullanımı kalıbını gösterir. Bu laboratuvarda Cloud Data Fusion'ın gerçek zamanlı veri ardışık düzenini kullanacağız. Gerçekçi sağlık hizmeti test verileri oluşturuldu ve sizin için Google Cloud Storage paketinde (gs://hcls_testing_data_fhir_10_patients/csv/) kullanıma sunuldu.

Bu kod laboratuvarında öğrenecekleriniz:

• Cloud Data Fusion'ı kullanarak Pub/Sub'dan BigQuery'ye CSV verilerini nasıl alacağınızı (gerçek zamanlı yükleme) öğrenin.
• Sağlık hizmeti verilerini gerçek zamanlı olarak yüklemek, dönüştürmek ve maskelemek için Cloud Data Fusion'da veri entegrasyonu ardışık düzenini görsel olarak oluşturma

Bu demoyu çalıştırmak için neye ihtiyacınız var?

• Bir GCP projesine erişiminiz olmalıdır.
• GCP projesinde size Sahip rolü atanmış olmalıdır.
• Başlık dahil olmak üzere CSV biçimindeki sağlık hizmeti verileri

GCP projeniz yoksa yeni bir GCP projesi oluşturmak için bu adımları uygulayın.

CSV biçimindeki sağlık hizmeti verileri, gs://hcls_testing_data_fhir_10_patients/csv/ adresindeki GCS paketine önceden yüklenmiştir. Her CSV kaynak dosyasının benzersiz bir şema yapısı vardır. Örneğin, Hastalar.csv, Sağlayıcılar.csv'den farklı bir şemaya sahiptir. Önceden yüklenmiş şema dosyalarını gs://hcls_testing_data_fhir_10_patients/csv_schemas adresinde bulabilirsiniz.

Yeni bir veri kümesine ihtiyacınız varsa SyntheaTM'i kullanarak istediğiniz zaman oluşturabilirsiniz. Ardından, giriş verilerini kopyalama adımında paketten kopyalamak yerine GCS'ye yükleyin.

Referans:

Google Cloud Platform Console'u kullanarak Cloud Data Fusion'daki (Hızlı Başlangıç) temel işlemleri gerçekleştirin.

Cloud Pub/Sub Mesaj Yayınlama https://cloud.google.com/pubsub/docs/publisher.

2. GCP Projesi Kurulumu

Ortamınız için kabuk değişkenlerini başlatın.

PROJECT_ID'yi bulmak için Projeleri tanımlama başlıklı makaleyi inceleyin.

<!-- CODELAB: Initialize shell variables ->
<!-- Your current GCP Project ID ->
export PROJECT_ID=<PROJECT_ID>
<!-- A new GCS Bucket in your current Project  - INPUT ->
export BUCKET_NAME=<BUCKET_NAME>
<!-- A new BQ Dataset ID - OUTPUT ->
export DATASET_ID=<DATASET_ID>

gsutil aracını kullanarak giriş verilerini ve hata günlüklerini depolamak için GCS paketi oluşturun.

gsutil mb -l us gs://$BUCKET_NAME

Yapay veri kümesine erişin.

1. Cloud Console'da oturum açmak için kullandığınız e-posta adresinden [e-mail] adresine katılma isteği içeren bir e-posta gönderin.
2. İşlemi nasıl onaylayacağınızla ilgili talimatları içeren bir e-posta alırsınız.
3. Gruba katılmak için e-postayı yanıtlama seçeneğini kullanın. düğmesini TIKLAMAYIN.
4. Onay e-postasını aldıktan sonra codelab'deki sonraki adıma geçebilirsiniz.

Giriş verilerini kopyalama

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

BigQuery veri kümesi oluşturun.

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

Unutmayın: Bu laboratuvar için SyntheaTM'yı kullanarak istediğiniz zaman yeni bir gerçekçi test verisi grubu oluşturabilirsiniz. Kolaylık sağlamak için aşağıdaki Google Cloud Storage'da bir veri kümesi sunuyoruz. gs://hcls_testing_data_fhir_10_patients/csv.

Google Cloud SDK'yı yükleyip başlatın ve Pub/Sub konusu ile abonelikleri oluşturun.

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

3. Cloud Data Fusion Ortam Kurulumu

Cloud Data Fusion API'yi etkinleştirmek ve gerekli izinleri vermek için aşağıdaki adımları uygulayın:

API'leri etkinleştirin.

1. GCP Console API Kitaplığı'na gidin.
2. Proje listesinden projenizi seçin.
3. API kitaplığında etkinleştirmek istediğiniz API'yi ( Cloud Data Fusion API, Cloud Pub/Sub API) seçin. API'yi bulma konusunda yardıma ihtiyacınız olursa arama alanını ve filtreleri kullanın.
4. API sayfasında ETKİNLEŞTİR'i tıklayın.

Cloud Data Fusion örneği oluşturun.

1. GCP Console'da ProjectID'nizi seçin.
2. Sol menüden Veri Füzyonu'nu seçin, ardından sayfanın ortasındaki ÖRNEK OLUŞTUR düğmesini (ilk oluşturma) veya üst menüdeki ÖRNEK OLUŞTUR düğmesini (ek oluşturma) tıklayın.

3. Örnek adını girin. Enterprise'ı seçin.

4. OLUŞTUR düğmesini tıklayın.

Not: Örnek oluşturma işleminin tamamlanması 30 dakika sürebilir.

Örnek izinlerini ayarlayın.

Bir örnek oluşturduktan sonra, örnekle ilişkili hizmet hesabına projenizde izin vermek için aşağıdaki adımları uygulayın:

Not: İzin verme işlemi bir projede yalnızca bir kez yapılmalıdır.

1. Örnek adını tıklayarak örnek ayrıntıları sayfasına gidin.

2. Hizmet hesabını kopyalayın.

3. Projenizin IAM sayfasına gidin.
4. IAM izinleri sayfasında Ekle düğmesini tıklayarak hizmet hesabına Cloud Data Fusion API Hizmet Aracısı rolünü verin. "Hizmet hesabı"nı Yeni üyeler alanına yapıştırın ve Hizmet Yönetimi -> Cloud Data Fusion API Sunucusu Aracısı rolünü seçin.

5. Pub/Sub Abonesi rolü eklemek için + Başka bir rol ekle'yi (veya Cloud Data Fusion API Hizmet Aracısı'nı Düzenle) tıklayın.

6. Kaydet'i tıklayın.

Bu adımlar tamamlandıktan sonra Cloud Data Fusion örnekleri sayfasındaki veya bir örneğin ayrıntılar sayfasındaki Örneği Görüntüle bağlantısını tıklayarak Cloud Data Fusion'ı kullanmaya başlayabilirsiniz.

Güvenlik duvarı kuralını ayarlayın.

1. GCP Console -> VPC Network -> Firewall rules'a (Güvenlik duvarı kuralları) giderek default-allow-ssh kuralının mevcut olup olmadığını kontrol edin.

2. Aksi takdirde, varsayılan ağa tüm giriş SSH trafiğine izin veren bir güvenlik duvarı kuralı ekleyin.

Komut satırını kullanma:

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

Kullanıcı arayüzünü kullanarak: Güvenlik Duvarı Kuralı Oluştur'u tıklayın ve bilgileri doldurun:

4. Ardışık düzen için düğümler oluşturma

GCP'de Cloud Data Fusion ortamını oluşturduğumuza göre, aşağıdaki adımları uygulayarak Cloud Data Fusion'da Data Pipelines oluşturmaya başlayalım:

1. Cloud Data Fusion penceresinde, İşlem sütunundaki Örneği Görüntüle bağlantısını tıklayın. Başka bir sayfaya yönlendirilirsiniz. Cloud Data Fusion örneğini açmak için sağlanan URL'yi tıklayın. Karşılama pop-up'ında "Tur başlat" veya "Hayır, teşekkürler" düğmesini tıklama tercihinize bağlıdır.
2. "Hamburger" menüsünü genişletin, Pipeline -> List'i (Liste) seçin.

3. Sağ üst köşedeki yeşil + düğmesini tıklayıp Dönüşüm hunisi oluştur'u seçin. Alternatif olarak, "Oluştur"u tıklayarak bir ardışık düzen bağlantısı oluşturabilirsiniz.

4. İşlem hattı stüdyosu göründüğünde sol üstteki açılır listeden Veri işlem hattı - Anlık'ı seçin.

5. Veri işlem hatları kullanıcı arayüzünde, sol panelde Filtre, Kaynak, Dönüştürme, Analiz, Hedef, Hata İşleyiciler ve Uyarılar gibi farklı bölümler görürsünüz. Bu bölümlerde işlem hattı için bir veya daha fazla düğüm seçebilirsiniz.

Kaynak düğümü seçin.

1. Soldaki Eklenti paletinde Kaynak bölümünün altında, Veri İşleme Hatları kullanıcı arayüzünde görünen Google Cloud PubSub düğümünü çift tıklayın.
2. İmleci PubSub kaynak düğümünün üzerine getirin ve Özellikler'i tıklayın.

3. Zorunlu alanları doldurun. Aşağıdaki alanları ayarlayın:

• Etiket = {herhangi bir metin}
• Referans adı = {herhangi bir metin}
• Project ID (Proje kimliği) = otomatik algılama
• Subscription = Pub/Sub konusu oluşturma bölümünde oluşturulan abonelik (örneğin, your-sub)
• Topic = Pub/Sub Konusu Oluşturma bölümünde oluşturulan konu (örneğin, your-topic)

4. Ayrıntılı açıklama için Belgeler'i tıklayın. Tüm giriş bilgilerini doğrulamak için Doğrula düğmesini tıklayın. Yeşil "Hata bulunamadı" ifadesi, işlemin başarılı olduğunu gösterir.

5. Pub/Sub Özellikleri'ni kapatmak için X düğmesini tıklayın.

Dönüştür düğümünü seçin.

1. Soldaki eklenti paletinde Dönüştür bölümünün altında, Veri İşleme Hatları kullanıcı arayüzünde görünen Projection (Projeksiyon) düğmesini çift tıklayın. Pub/Sub kaynak düğümünü Projection transform düğümüne bağlayın.
2. Fare imlecini Projection (Projeksiyon) düğümünün üzerine getirin ve Properties'i (Özellikler) tıklayın.

3. Zorunlu alanları doldurun. Aşağıdaki alanları ayarlayın:

• Convert = message öğesini bayt türünden dize türüne dönüştürür.
• Bırakılacak alanlar = {any field}
• Tutulacak alanlar = {message, timestamp ve attributes} (örneğin, Pub/Sub'dan gönderilen attributes: key=‘filename':value=‘patients')
• Yeniden adlandırılacak alanlar = {message, timestamp}

4. Ayrıntılı açıklama için Belgeler'i tıklayın. Tüm giriş bilgilerini doğrulamak için Doğrula düğmesini tıklayın. Yeşil "Hata bulunamadı" ifadesi, işlemin başarılı olduğunu gösterir.

5. Soldaki Eklenti paletinde Dönüştür bölümünün altında, Veri İşleme Hatları kullanıcı arayüzünde görünen Wrangler düğmesini çift tıklayın. Projection transform düğümünü Wrangler transform düğümüne bağlayın. Fare imlecini Wrangler düğümünün üzerine getirin ve Properties'i (Özellikler) tıklayın.

6. Kaydedilmiş bir şemayı içe aktarmak için İşlemler açılır menüsünü tıklayın ve İçe aktar'ı seçin (örneğin: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json).
7. Son alanın yanındaki + düğmesini tıklayarak Çıkış Şeması'na TIMESTAMP alanını ekleyin (yoksa) ve "Null" kutusunu işaretleyin.
8. Zorunlu alanları doldurun. Aşağıdaki alanları ayarlayın:

• Etiket = {herhangi bir metin}
• Giriş alanı adı = {*}
• Pub/Sub kaynak düğümünden gönderilen her kayıt veya ileti türünü (örneğin, hastalar, sağlayıcılar, alerjiler vb.) ayırt etmek için ön koşul = {attributes.get("filename") != "patients"}.

9. Ayrıntılı açıklama için Belgeler'i tıklayın. Tüm giriş bilgilerini doğrulamak için Doğrula düğmesini tıklayın. Yeşil "Hata bulunamadı" ifadesi, işlemin başarılı olduğunu gösterir.

10. Sütun adlarını tercih ettiğiniz sıraya göre ayarlayın ve ihtiyacınız olmayan alanları bırakın. Aşağıdaki kod snippet'ini kopyalayıp Yemek Tarifi kutusuna yapıştırın.

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

Not: Yanlış, "İlk satırı başlık olarak ayarla" seçeneği işaretlenmeden gövde sütununun ayrıştırılması anlamına gelir.

11. Veri maskeleme ve kimlik gizleme için Batch-Codelab - CSV to BigQuery via CDF başlıklı makaleyi inceleyin. Alternatif olarak, tarif kutusuna şu kod snippet'ini mask-number SSN xxxxxxx#### ekleyin.
12. Dönüştürme Özellikleri penceresini kapatmak için X düğmesini tıklayın.

Sink düğümünü seçin.

1. Soldaki eklenti paletinde, Sink bölümünde, Veri İşleme Hattı kullanıcı arayüzünde görünen BigQuery düğmesini çift tıklayın. Wrangler dönüştürme düğümünü BigQuery havuzu düğümüne bağlayın.
2. BigQuery havuz düğümünün üzerine gelin ve Özellikler'i tıklayın.

3. Zorunlu alanları doldurun:

• Etiket = {herhangi bir metin}
• Referans adı = {herhangi bir metin}
• Project ID (Proje kimliği) = otomatik algılama
• Veri kümesi = Geçerli projede kullanılan BigQuery veri kümesi (örneğin, DATASET_ID)
• Tablo = {table name}

4. Ayrıntılı açıklama için Belgeler'i tıklayın. Tüm giriş bilgilerini doğrulamak için Doğrula düğmesini tıklayın. Yeşil "Hata bulunamadı" ifadesi, işlemin başarılı olduğunu gösterir.

5. BigQuery Özellikleri'ni kapatmak için X düğmesini tıklayın.

5. Gerçek zamanlı veri ardışık düzeni oluşturma

Önceki bölümde, Cloud Data Fusion'da veri ardışık düzeni oluşturmak için gereken düğümleri oluşturduk. Bu bölümde, gerçek işlem hattını oluşturmak için düğümleri bağlıyoruz.

Bir işlem hattındaki tüm düğümleri bağlama

1. Kaynak düğümün sağ kenarındaki bağlantı okunu (>) sürükleyip hedef düğümün sol kenarına bırakın.
2. Bir boru hattında, aynı PubSub Kaynak düğümünden yayınlanan mesajları alan birden fazla dal olabilir.

Not: Her Pub/Sub mesajı için işlem hattındaki Projection düğümünden bir dal oluşturun.

3. Ardışık düzeni adlandırın.

Not: Bu işlem boyunca, siz işlem hattını oluştururken şema otomatik olarak yayılır. İstediğiniz bir şemayı da içe aktarabilirsiniz.

Bu kadar basit. Dağıtılacak ve çalıştırılacak ilk gerçek zamanlı veri ardışık düzeninizi oluşturdunuz.

Cloud Pub/Sub üzerinden mesaj gönderme

Pub/Sub kullanıcı arayüzünü kullanma:

1. GCP Console -> Pub/Sub -> Konular'a gidin, your-topic'i seçin ve üst menüden MESAJ YAYINLA'yı tıklayın.

2. İleti alanına tek seferde yalnızca bir kayıt satırı yerleştirin. +ÖZELLİK EKLE düğmesini tıklayın. Anahtar = filename, Değer = <type of record> (örneğin, hastalar, sağlayıcılar, alerjiler vb.) değerini girin.
3. Mesajı göndermek için Yayınla düğmesini tıklayın.

gcloud komutunu kullanma:

1. Mesajı manuel olarak girin.

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

2. cat ve sed Unix komutlarını kullanarak mesajı yarı otomatik olarak sağlama. Bu komut, farklı parametrelerle tekrar tekrar çalıştırılabilir.

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

Not: <your-topic>: Pub/Sub kaynak düğümü özelliklerinde sağlanan konuyla aynıdır.

<key> - filename

cat: Dosya içeriklerini okuma

sed: Çıkışı bastırın (-n) ve belirli satır numarasını (1-?) açıkça yazdırın (p). Başlık satırı varsa "2p" vb. kullanın.

6. Ardışık Düzeni Yapılandırma, Dağıtma ve Çalıştırma

Veri ardışık düzenini geliştirdiğimize göre artık Cloud Data Fusion'da dağıtıp çalıştırabiliriz.

1. Yapılandır varsayılanlarını koruyun.
2. Verileri önizlemek için Önizleme'yi tıklayın**.** Önceki pencereye geri dönmek için **Önizleme**'yi tekrar tıklayın. **ÇALIŞTIR**'ı tıklayarak da işlem hattını önizleme modunda çalıştırabilirsiniz.

3. Günlükleri görüntülemek için Günlükler'i tıklayın.
4. Tüm değişiklikleri kaydetmek için Kaydet'i tıklayın.
5. Yeni bir işlem hattı oluştururken kaydedilen işlem hattı yapılandırmasını içe aktarmak için İçe aktar'ı tıklayın.
6. Ardışık düzen yapılandırmasını dışa aktarmak için Dışa aktar'ı tıklayın.
7. İşlem hattını dağıtmak için Dağıt'ı tıklayın.
8. Dağıtım tamamlandıktan sonra Çalıştır'ı tıklayın ve işlem hattının tamamlanmasını bekleyin.

Not: Bir ardışık düzen çalıştırdığınızda Cloud Data Fusion, kısa ömürlü bir Cloud Dataproc kümesi sağlar, ardışık düzeni çalıştırır ve ardından kümeyi kapatır. Bu işlem birkaç dakika sürebilir. Bu süre zarfında, işlem hattı geçişinin Sağlanıyor, Başlatılıyor, Çalışıyor, Sağlama Kaldırılıyor ve Başarılı durumlarını gözlemleyebilirsiniz.

9. Ardışık düzen çalıştırmasını istediğiniz zaman durdurmak için Durdur'u tıklayın.
10. İşlemler düğmesi altından Kopyala'yı seçerek işlem hattını kopyalayabilirsiniz.
11. İşlemler düğmesinin altındaki Dışa aktar'ı seçerek işlem hattı yapılandırmasını dışa aktarabilirsiniz.

12. Çalıştırma geçmişi, kayıtlar, hata günlükleri ve uyarıların grafiklerini göstermek için Özet'i tıklayın.

7. Doğrulama

Bu bölümde, veri ardışık düzeninin yürütülmesini doğruluyoruz.

1. İşlem hattının başarıyla yürütüldüğünü ve sürekli olarak çalıştığını doğrulayın.

2. BigQuery tablolarının, TIMESTAMP'e göre güncellenmiş kayıtlarla yüklendiğini doğrulayın. Bu örnekte, 25.06.2019 tarihinde Pub/Sub konusuna iki hasta kaydı veya mesajı ve bir alerji kaydı ya da mesajı yayınlanmıştır.

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. <your-topic> konusuna yayınlanan iletilerin <your-sub> abonesi tarafından alındığını doğrulayın.

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

Sonuçları görüntüleme

Gerçek zamanlı işlem hattı çalışırken mesajlar Pub/Sub konusuna yayınlandıktan sonra sonuçları görüntülemek için:

1. BigQuery kullanıcı arayüzünde tabloyu sorgulayın. BIGQUERY KULLANICI ARAYÜZÜNE GİT
2. Aşağıdaki sorguyu kendi proje adınız, veri kümeniz ve tablonuzla güncelleyin.

8. Temizleme

Bu eğiticide kullanılan kaynaklar için Google Cloud Platform hesabınızın ücretlendirilmesini istemiyorsanız şunları yapın:

Eğitimi tamamladıktan sonra, GCP'de oluşturduğunuz kaynakları temizleyerek kota kullanımlarını önleyebilir ve gelecekte bu kaynaklar için faturalandırılmamayı sağlayabilirsiniz. Aşağıdaki bölümlerde bu kaynakların nasıl silineceği veya devre dışı bırakılacağı açıklanmaktadır.

BigQuery veri kümesini silme

Bu eğitimin bir parçası olarak oluşturduğunuz BigQuery veri kümesini silmek için bu talimatları uygulayın.

GCS paketini silme

Bu eğitimin bir parçası olarak oluşturduğunuz GCS paketini silmek için aşağıdaki talimatları uygulayın.

Cloud Data Fusion örneğini silme

Cloud Data Fusion örneğinizi silmek için aşağıdaki talimatları uygulayın.

Projeyi silme

Faturalandırılmanın önüne geçmenin en kolay yolu, eğitim için oluşturduğunuz projeyi silmektir.

Projeyi silmek için:

Dikkat: Bir projeyi silmenin aşağıdaki etkileri olur:

• Projedeki her şey silinir. Bu eğitim için mevcut bir proje kullandıysanız projeyi sildiğinizde projede yaptığınız diğer tüm çalışmalar da silinir.
• Özel proje kimlikleri kaybolur. Bu projeyi oluştururken gelecekte kullanmak istediğiniz özel bir proje kimliği oluşturmuş olabilirsiniz. Proje kimliğini kullanan URL'leri (ör. appspot.com URL'si) korumak için projenin tamamını silmek yerine proje içindeki seçili kaynakları silin.

Birden fazla eğitici ve hızlı başlangıç kılavuzunu incelemeyi planlıyorsanız projeleri yeniden kullanmak proje kotası sınırlarını aşmanızı önleyebilir.

1. GCP Console'da Projeler sayfasına gidin. PROJELER SAYFASINA GİTME
2. Proje listesinde, silmek istediğiniz projeyi seçin ve Sil'i tıklayın.
3. İletişim kutusunda proje kimliğini yazın ve projeyi silmek için Kapat'ı tıklayın.

9. Tebrikler

Tebrikler, Cloud Data Fusion'ı kullanarak BigQuery'ye sağlık hizmetleri verilerini alma ile ilgili kod laboratuvarını başarıyla tamamladınız.

CSV verilerini Pub/Sub konusuna yayınlayıp BigQuery'ye yüklediniz.

Sağlık hizmeti verilerini gerçek zamanlı olarak yüklemek, dönüştürmek ve maskelemek için görsel olarak bir veri entegrasyonu ardışık düzeni oluşturdunuz.

Artık Google Cloud Platform'da BigQuery ile sağlık hizmetleri veri analizi yolculuğunuza başlamak için gereken temel adımları biliyorsunuz.