1) Giriş

Birçok zaman belli bir kod betiğini periyodik olarak otomatik şekilde çalıştırma ihtiyacımız olmaktadır. Bu yazıda güncel verilerin bir web sitesi üzerinden her yarım saatte bir alınarak, Google Cloud Platform üzerinde depolanması ve analiz süreçlerinden bahsedeceğim.

Bu iş için Google Cloud Platform’daki Cloud Scheduler, Cloud Functions, Cloud Storage ve BigQuery servislerini kullanacağız. Crawler işlemini gerçekleştirecek Python betiği, Cloud Functions üzerinde çalıştırılacak, elde edilen veri dosya olarak Cloud Storage Bucket üzerinde tutulurken, bu dosya BigQuery’de tablo üzerine append edilecektir. Cloud Functions’da Python betiğinin her yarım saatte tetiklenmesi için ise Cloud Scheduler servisi kullanılacaktır. Bir diğer fonksiyonun tetiklenmesi için Cloud Storage Trigger özelliğinden faydalanacağız. Böylece Cloud Functions servisinde hazırlanan fonksiyonun iki farklı şekilde tetiklenmesi için örnek görmüş olacağız.

Uygulamada bahsedilen servislerin kullanımı için Google Cloud Platformu projenizde ilgili servislerin enable edilmesi gerekmektedir.

Kullanılacak servislerden kısaca bahsedecek olursak;

1.1. Cloud Scheduler: Cloud Scheduler, belirli görev ve fonksiyonları belirli bir zaman diliminde ya da belirli bir periyotta otomatik olarak çalıştırmak için kullanılan ve Google Cloud Platform tarafından sunulan cron tabanlı bir servistir. HTTP, HTTPS ve Google Cloud Platform tarafından sunulan birçok farklı servis ile çalışabilmektedir.

1.2. Cloud Functions: Cloud Functions, yazılım geliştirmeciler tarafından kullanılan küçük parça kod betiklerinin fonksiyonlarının yürütülmesini sağlayan ve Google Cloud Platform tarafından sunulan sunucusuz bir servistir. Oluşturulan fonksiyonlar, Cloud Scheduler gibi servislerin tetiklemesi ile otomatik olarak çalıştırılabilmektedir.

1.3. Cloud Storage: Cloud Storage, kullanıcı verilerinin Google altyapısında tutulmasını ve rahatlıkla erişebilmesini sağlayan bir depoloma hizmetidir.

1.4. BigQuery: BigQuery, Google Cloud Platform tarafından sunulan, büyük veri setlerinin depolanabildiği, analiz edilebildiği, hızlı ve karmaşık sorguların çalıştırılabildiği, ölçeklenebilen ve farklı iş zekası uygulamaları ile entegre edilebilen bir cloud datawarehouse servisidir.

2) Cloud Storage’da Bucket Oluşturma

İlk adım olarak Cloud Storage servisinde bucket oluşturmak için GCP projenizde “Navigation Menu” üzerinden Cloud Storage’a gidilmesi gerekecektir. Cloud Storage’da Bucket sekmesinde “CREATE” butonuna tıklayarak istenilen bilgilerin doldurulması ile Bucket oluşturma işlemi gerçekleştirilmektedir.

Bu projede region kısmında “Multi-region” seçimi ile ilerledim. “Storage class”, “control access” ve “how to protect object data” seçimlerini default olarak bıraktım. Bucket name’in unique bir isim olması gerekmektedir. Bunun için isim verirken dikkat etmek gerekecektir.

3) Cloud Functions’da Function Oluşturma

Lokal makinede Python programlama dili ile yazılan ve test edilen web scraping kod betiği, Cloud Functions servisine deploy edilecektir. Bu fonksiyon adresinden döviz kuru verilerini çekerek CSV formatına çevirip, Cloud Storage’da oluşturulan Bucket içerisine upload etme işlemini gerçekleştirmektedir.

Cloud Function’a deploy işlemini yürütmek için Cloud Functions sayfasında “Create Function” butonu ile function oluşturma işlemine başlayacağız.

Karşımıza çıkan ekranda aşağıdaki gibi bilgilerimizi giriyoruz. Cloud Scheduler ile tetikleme yapacağımızdan dolayı trigger seçimimizi HTTP olarak veriyoruz.

Gelişmiş ayarlar için alt kısımdaki “Runtime, build, connections and security settings” butonu ile ekranı açıyoruz ve kullandığımız veri boyutu ve işlem hacmimize göre makine değerlerini güncelleyebilirken, connection, security gibi seçimlerimizde de güncellemeler gerçekleştirebiliriz. Bu uygulamamızda bu kısımda herhangi bir değişiklik yapmadan ilerlemekteyiz.

“Next” butonu ile ilerlediğimizde ise “source” kısmına geçiş yapılmaktadır. Bu kısımda ise çalıştıracağımız kod betiği hangi programlama dilinde ise onun seçimini yapmamız gerekecek ve Cloud Functions yapısına göre fonksiyon kodumuzu betiğe alıyoruz. “Runtime” kısmından Python 3.9 seçiyoruz. Kodumuz eğer hazır ise “source code” kısmında file path verip ilerleyebiliriz. Ancak bu uygulamada biz kodumuzu manuel olarak main.py dosyası içerisine ekliyoruz.

import requests
from bs4 import BeautifulSoup
from datetime import datetime
import json
import time
from google.cloud import storage
import unicodedata
import pandas as pd
import os


def crawler(request):
    date = datetime.today().strftime('%Y-%m-%d %H:%M:%S')
    print("Tarih", date)
    r = requests.get("https://bigpara.hurriyet.com.tr/doviz/")
    soup = BeautifulSoup(r.content, "html.parser")
    currency_list_len = len(
        soup.select('#content > div.contentLeft > div > div.tableCnt > div > div.tableBox.srbstPysDvz > '
                    'div.tBody > ul'))
    exchange_rate_dict = []
    for i in range(1, currency_list_len + 1):
        rate_info = {'date': date}

        main_search_string = f'#content > div.contentLeft > div > div.tableCnt > div > div.tableBox.srbstPysDvz > ' \
                             f'div.tBody ul:nth-child({i})'

        data = soup.select_one(main_search_string)

        currency_code = data.select_one('li.cell010.tal > h3 > a > img')['src'].replace('/Assets/images/mini_bayrak/',
                                                                                        '').replace('.png', '')
        rate_info['currency_code'] = currency_code
        currency = data.select_one('li.cell010.tal > h3 > a').text
        rate_info['currency'] = currency
        forex_buying = data.select_one('li:nth-child(3)').text.replace(".", "").replace(",", ".")
        rate_info['forex_buying'] = forex_buying
        forex_selling = data.select_one('li:nth-child(4)').text.replace(".", "").replace(",", ".")
        rate_info['forex_selling'] = forex_selling
        forex_percent = data.select_one('li:nth-child(5)').text.replace(".", "").replace(",", ".").replace("%", "")
        rate_info['forex_percent'] = forex_percent

        exchange_rate_dict.append(rate_info)

    gold_index_list = [2, 3, 5, 6, 8]

    gold_currency_mapping = {
        'Gram Altın': 'GAU',
        'Ons Altın': 'ONS'
    }

    r2 = requests.get("https://uzmanpara.milliyet.com.tr/altin-fiyatlari/")
    soup = BeautifulSoup(r2.content, "html.parser")
    for i in gold_index_list:
        rate_info = {'date': date}

        main_search_string = f'#altinfiyat > tbody > tr:nth-child({i})'

        data = soup.select_one(main_search_string)
        currency = data.select_one('td.currency > a').text
        rate_info['currency'] = currency
        currency_code = gold_currency_mapping.get(currency, currency)
        rate_info['currency_code'] = currency_code

        forex_buying = data.select_one('td:nth-child(3)').text.replace(".", "").replace(",", ".").replace(' TL', '')
        rate_info['forex_buying'] = forex_buying
        forex_selling = data.select_one('td:nth-child(4)').text.replace(".", "").replace(",", ".").replace(' TL', '')
        rate_info['forex_selling'] = forex_selling
        forex_percent = data.select_one('td:nth-child(5)').text.replace('% ', '').replace(".", "").replace(",", ".")
        rate_info['forex_percent'] = forex_percent

        exchange_rate_dict.append(rate_info)

    silver_main_search_string = 'body > div:nth-child(12) > div.detMain.borsaMain.goldMain > div.detL > ' \
                                'div:nth-child(4) > table > tbody > tr:nth-child(2)'
    data = soup.select_one(silver_main_search_string)
    silver_buy = data.select_one('td:nth-child(3)').text.replace(".", "").replace(",", ".")
    silver_sell = data.select_one('td:nth-child(4)').text.replace(".", "").replace(",", ".")
    silver_exchange_rate = data.select_one('td:nth-child(5)').text.replace('% ', '').replace(".", "").replace(",", ".")

    exchange_rate_dict.append({
        'date': date,
        'currency_code': 'XAG',
        'currency': 'Gümüs',
        'forex_buying': silver_buy,
        'forex_selling': silver_sell,
        'forex_percent': silver_exchange_rate

    })
    coin_index_list = [2, 3, 8]

    gold_currency_mapping = {
        'Gram Altın': 'GAU',
        'Ons Altın': 'ONS'
    }

    r3 = requests.get("https://uzmanpara.milliyet.com.tr/kripto-paralar/")
    soup = BeautifulSoup(r3.content, "html.parser")

    for i in coin_index_list:
        rate_info = {'date': date}

        main_search_string = f'#myTable > tbody > tr:nth-child({i})'

        data = soup.select_one(main_search_string)

        raw_currency = data.select_one('td.currency > a').text.split(' ')

        currency = raw_currency[0]
        rate_info['currency'] = currency
        currency_code = raw_currency[1].replace('(', '').replace(')', '')
        rate_info['currency_code'] = currency_code

        forex_buying = data.select_one('td:nth-child(2)').text.replace(".", "").replace(",", ".")
        rate_info['forex_buying'] = forex_buying
        forex_selling = data.select_one('td:nth-child(3)').text.replace(".", "").replace(",", ".")
        rate_info['forex_selling'] = forex_selling
        forex_percent = data.select_one('td:nth-child(4)').text.replace('% ', '').replace(".", "").replace(",", ".")
        rate_info['forex_percent'] = forex_percent

        exchange_rate_dict.append(rate_info)
    time.sleep(2)
    bucket_name = "test-bucket"
    dfItem = pd.DataFrame.from_records(exchange_rate_dict)
    time.sleep(2)
    storage_client = storage.Client(project='kartaca-test')
    bucket = storage_client.get_bucket(bucket_name)
    bucket.blob('test-berke/forex_rate.csv').upload_from_string(
        dfItem.to_csv(sep=',', encoding='utf-8', header=False, index=False), 'text/csv')
    time.sleep(2)
    status = "Basarili"

    return status

Yukarıdaki fonksiyon kodu ile veri çekme işlemi ve CSV formatında Storage Bucket içerisine file upload etme işlemleri gerçekleştirilmektedir. Sonrasında ise kullanılan Python kütüphanelerinin “requirements.txt” dosyası içerisine eklenmesi gerekmektedir.

“Test Function” ile fonksiyonu test ettikten sonra başarılı sonuç aldığımızda “Deploy” butonu ile fonksiyon oluşturuyoruz.

Fonksiyon oluşturulduktan sonra “Cloud Functions” sayfasında oluşturduğumuz function listelenecektir.

4) Cloud Scheduler’da Job Oluşturma

Cloud Function’da function oluşturma işleminin ardından fonksiyonu tetikleyecek Cloud Scheduler job oluşturulması gerekecektir. Bunun için Cloud Scheduler sayfasına geçiş yapıyoruz ve “Create Job” butonu ile ilerliyoruz.

Oluşturacağımız Scheduler Job’a bir isim tanımladıktan sonra frequency kısmına her yarım saatte bir işlem yaptırmak istediğimizden dolayı “30 * * * *” yazıyoruz. (Crontab adresinden faydalanabilirsiniz.) Region olarak ise bir önceki adımda oluşturduğumuz function “us-central1” bölgesinde olduğundan dolayı aynı şekilde seçiyoruz.Timezone kısmına ise bulunduğunuz konuma göre seçim yapabilirsiniz.

Execution config parametleri için seçimlerimizde “Target Type” parametresini “HTTP” olarak seçiyoruz. Bu seçimden sonra diğer parametre seçimleri açılacaktır. URL kısmına oluşturduğumuz Cloud Function’dan “Trigger URL” bilgisini yapıştırıyoruz. Bu bilgiye ayrı sekmede oluşturduğumuz Cloud Function içerisine gidip “Trigger” sekmesinden ulaşabiliriz.

HTTP Methodumuzu ise “Post” olarak seçiyoruz. “Auth header” parametresini ise “Add OAuth token” olarak seçiyoruz ve seçimden sonra iki farklı parametre ile karşılaşıyoruz. Service account parametresini default compute engine service account olarak bırakırken “Audience” kısmına yine Cloud Function üzerinden elde ettiğimiz trigger URL değerini giriyoruz.

Bu kısımdaki bilgileri tamamladıktan sonra “Create” diyerek Scheduler Job oluşturabiliriz.

Scheduler job oluşturulduktan sonra Cloud Scheduler konsolunda job listenelecektir. Çalışmanızı test etmek isterseniz sağ kısımdaki “Actions” kısmında işi “Force run” diyerek çalıştırabiliriz ve Function’ı çalıştırmasını bekleyebiliriz. “View logs” ile çalışan job için Cloud Logging sayfasına yönlendirilerek de log incelemesi yapabilirsiniz.

5) BigQuery

Cloud Functions kullanımı elde edilen veri dosyası Cloud Storage’a yazıldıktan sonra bu dosyanın BigQuery’de ilgili tabloya eklenmesi ve üzerine gerekirse SQL sorguları ile analiz çalışmalarının yapılması gerekecektir. Bunun için BigQuery’de bir dataset ve içerisine veri şemasına uygun bir tablo oluşturulmalıdır.

Bunun için ilk adım olarak BigQuery üzerinde proje içerisinde dataset oluşturulacaktır.

Yukarıdaki görselde proje isminizin yer aldığı action kısmından “Create dataset” butonu ile beraber dataset id ve region seçimi ile beraber dataset oluşturulabilmektedir.

Sonrasında ise oluşturulan dataset içerisine gidip yine aynı şekilde action kısmından “Create table” butonu ile ilerlemek gerekecektir.

1. Tablo oluşturulurken “Create table from” seçeniğinde “Google Cloud Storage” seçiyoruz ve upload edilen veri dosyasını seçiyoruz.
2. “Destination” kısmında “Project” ve “Dataset” kısımları zaten dataset üzerinden geldiğimiz için hazır gelmekte. “Table” kısmına tablomuza verilecek ismi giriyoruz.
3. “Table Type” kısmında ise “Native Table” olarak seçim yapıyoruz. Native tablolar, BigQuery’de depolanan ve yönetilen tablolardır. External tablolar ise harici veri kaynaklarından BigQuery dışındaki kaynaklara bağlanan tablolardır. Buna örnek olarak Bigtable, Cloud Storage gibi yerlerde tutulan verileri verebiliriz. BigQuery tarafında yönetilmez, veri kaynağı referans olarak tutulur ve sorgulanır.
4. Schema kısmında ise veri kaynağımızdaki sütunların veri türüne göre şema belirlenmelidir. Veri türünün değişebileceği sütunlar için uygun bir type seçilmelidir. Bu uygulama için aşağıdaki şekilde verilmiştir.

5. “Partition and cluster settings” bölümünde ise “Partitioning” kısmında “Partition by ingestion time” seçimi yapıyoruz. “Partitioning type” kısmında da “By day” seçerek, tablomuzu “day” bilgisine göre parçalamış oluyoruz.
6. “Partitioning filter” kısmında “Require WHERE clause to query data” seçimi bulunmaktadır. Bu tabloya sorgu attığınızda “day” bilgisine göre “WHERE” koşulu zorunlu tutmak isterseniz “Enable” edilmesi gerekecektir. Çok büyük veri boyutuna sahip tablolarda bu seçenek önemli olacaktır.
7. Bu işlemlerin ardından “Create Table” diyerek dataset altında tablo oluşturulabilmektedir. İçerisine de Cloud Storage üzerinde seçtiğimiz veri dosyasındaki veriler eklenmiş şekilde görülebilmektedir.

BigQuery üzerinde gerekli aksiyonları tamamladıktan sonra buradaki süreci otomatize etmemiz gerekiyor. Yani Cloud Storage’a veri yazıldıktan sonra verinin BigQuery’de oluşturduğumuz tabloya eklenmesini istiyoruz. Bunun için Cloud Functions servisini kullanacağız.

6) Cloud Functions ile BigQuery’e Veri Aktarımı

Bu aşamada oluşturacağımız Cloud Functions, Cloud Storage üzerinde elde edilen veri dosyasını okuyacak ve BigQuery’de ilgili dataset altındaki tabloya veri insert edecek. Ancak bu fonksiyonun Cloud Storage’a veri yüklendikten sonra çalışması gerekmekte. Bu nedenle oluşturulan Cloud Functions’ı veri eklendiği zaman tetiklenecek şekilde düzenleyeceğiz.

Tekrardan Cloud Functions sayfasına geçiyoruz ve “Create Function” butonu ile ilerliyoruz.

1. “Environment”, “Function name”, “Region” seçimlerimizi önceki adımlardaki gibi yapıyoruz.
2. “Trigger kısmında ise alt kısımda bulunan “Add Trigger” butonuna tıklıyoruz ve açılan seçeneklerden “Cloud Storage Trigger” seçiyoruz.


Bu seçimden sonra karşımıza çıkan “Eventarc trigger kısmında” Bucket içerisindeki dosya yolumuzu seçiyoruz. Seçeceğimiz service account için de IAM üzerinden Cloud Storage ve BigQuery yetkilerini verilmesi gerekmektedir. Bu yetkilere sahip olmayan Service Account seçildiği takdirde fonksiyon hata alacaktır.

3. Seçimlerden sonra “Next” butonu ile beraber “Code” kısmına geçiliyor ve burada Cloud Storage’dan dosyayı okuyan ve BigQuery’de ilgili tabloya veriyi insert eden Python kod betiğini fonksiyon olarak eklenmesi gerekecektir. Entry point kısmına “storage_to_bq” olarak giriyoruz. Requirements.txt içerisine gerekli Python kütüphaneleri eklenmelidir.


“Main.py” içerisine ise fonksyiona “storage_to_bq” ismini veriyoruz. Bu “Entry point” ile aynı olmalıdır. Sonrasında ise fonksiyon içerisine python kodumuzu yazıyoruz.



import json
from google.cloud import storage
from google.cloud import bigquery
import os
import time

def stroage_to_bq(data, context) : 
    project_id = 'kartaca-test'
    dataset_id = 'berke_test_exchange'
    table_id = 'kartaca-test.berke_test_exchange.forex_rate'
    client = bigquery.Client(project='kartaca-test')
    job_config = bigquery.LoadJobConfig(
        schema=[
            bigquery.SchemaField("date", "DATETIME"),
            bigquery.SchemaField("currency_code", "STRING"),
            bigquery.SchemaField("currency", "STRING"),
            bigquery.SchemaField("forex_buying", "FLOAT"),
            bigquery.SchemaField("forex_selling", "FLOAT"),
            bigquery.SchemaField("forex_percent", "FLOAT"),
        ],
        source_format=bigquery.SourceFormat.CSV,
        write_disposition = 'WRITE_APPEND',
        time_partitioning=bigquery.TimePartitioning(
            type_=bigquery.TimePartitioningType.DAY,
            field="date",  # Name of the column to use for partitioning.
            expiration_ms=7776000000,  # 90 days.
        ),
    )
    uri = "gs://test-bucket/test-berke/forex_rate.csv"
    load_job = client.load_table_from_uri(
        uri, table_id, job_config=job_config
    )  
    load_job.result()  
    time.sleep(2)
    table = client.get_table(table_id)
    view_id = "kartaca-test.berke_test_exchange.dollar_forex_rate_view"
    view = client.get_table(view_id)
    job_config_dolar = bigquery.QueryJobConfig(destination="kartaca-test.berke_test_exchange.dollar_forex_rate")
    job_config_dolar.write_disposition = 'WRITE_TRUNCATE'   
    sql = view.view_query
    query_job = client.query(sql, job_config=job_config_dolar)
    query_job.result()
    return project_id

4. “Test Function” ile test işleminden sonra fonksiyonumuzu deploy edebiliriz.
5. Oluşturulan fonksiyon Cloud Functions sayfasında listelenecektir.
6. Cloud Logging servisi ile de fonksiyon logları takip edilebilmektedir.

Bu işlemlerle aslında akışımız tamamlanmıştır. Özet olarak, gittiğimiz web adresinden istediğimiz bilgileri Python fonksiyonu ile alabiliyoruz ve bu fonksiyonu Google Cloud Functions servisi ile çalıştırıyoruz. Bu fonksiyonun periyodik olarak cron ile tetiklenip çalıştırılması için de Cloud Scheduler servisini kullanıyoruz. Elde edilen veri dosyası Cloud Storage’da oluşturulan bucket içerisine upload edilmekte. Bunun sonrasında ise veriyi görebilmek ve analiz edebilmek için BigQuery ortamına ihtiyaç duyuyoruz. Burada başka bir Cloud Function oluşturularak Python betiği Cloud Storage’a yazılan veriyi okur ve BigQuery’de ilgili tabloya veriyi insert eder. Bu fonksiyonun Cloud Storage’a veri yüklendiğinde çalıştırılması için de trigger olarak Cloud Storage Trigger seçilmiştir. Böylelikle gün içerisinde web sitesinden elde edilen en güncel verileri BigQuery tablomuzda toplayabilir ve üzerinde analiz çalışmaları gerçekleştirebiliriz.

BigQuery’de veriler elde edildikten sonra tablomuzu bir görselleştirme aracına entegre ederek de farklı grafikler ile görselleştirme yapabiliriz. Looker ile veri görselleştirme süreçlerinden detaylı olarak bahsettiğimiz yazımıza buradan göz atabilirsiniz.