Hologres:Menerapkan analitik batch dan real-time terpadu menggunakan dataset event publik GitHub

Dalam arsitektur ini, sebuah Instance ECS mengumpulkan dan mengagregasi data event real-time dan batch dari GitHub, yang kemudian berfungsi sebagai sumber data. Data ini dialirkan ke pipeline real-time dan pipeline batch. Data dari kedua pipeline tersebut kemudian dikonsolidasikan di Hologres, yang menyediakan lapisan layanan terpadu.

• Pipeline real-time: Realtime Compute for Apache Flink memproses data dari Simple Log Service (SLS) secara real-time dan menuliskannya ke Hologres. Flink adalah mesin pemrosesan aliran yang andal. Hologres mendukung penulisan dan pembaruan data secara real-time, dengan data dapat langsung diakses setelah ditulis. Integrasi natif mereka mendukung pengembangan gudang data real-time berkinerja tinggi, berlatensi rendah, berbasis model, dan berkualitas tinggi. Hal ini memenuhi kebutuhan wawasan bisnis real-time, seperti mengekstraksi event terbaru dan menganalisis tren event.

• Pipeline batch: MaxCompute memproses dan mengarsipkan data batch dalam jumlah besar. Object Storage Service (OSS) adalah layanan Alibaba Cloud untuk menyimpan berbagai jenis data. Data mentah yang digunakan dalam tutorial ini berformat JSON, dan OSS menyediakan penyimpanan yang nyaman, aman, berbiaya rendah, dan andal. MaxCompute adalah gudang data cloud SaaS tingkat enterprise yang dirancang untuk analitik data. Layanan ini dapat langsung membaca dan mengurai data semi-terstruktur di OSS menggunakan tabel eksternal, mengintegrasikan data bernilai tinggi ke penyimpanan internalnya, lalu mengintegrasikannya dengan DataWorks untuk membangun gudang data batch.

• Hologres terintegrasi secara mulus dengan MaxCompute pada lapisan penyimpanan. Hal ini memungkinkan Anda menggunakan Hologres untuk mempercepat kueri dan analisis terhadap volume besar data historis di MaxCompute, memenuhi kebutuhan bisnis akan kueri berfrekuensi rendah namun berkinerja tinggi terhadap data historis. Anda juga dapat dengan mudah menggunakan pipeline batch untuk mengoreksi data real-time, guna menyelesaikan isu seperti kelalaian data yang dapat terjadi pada pipeline real-time.

Solusi ini menawarkan keunggulan berikut:

• Pipeline batch yang stabil dan efisien: Mendukung penulisan dan pembaruan data per jam, mampu memproses data skala besar secara batch, melakukan perhitungan dan analisis kompleks, mengurangi biaya komputasi, serta meningkatkan efisiensi pemrosesan data.

• Pipeline real-time yang matang: Mendukung ingestion real-time, komputasi event real-time, dan analitik real-time. Pipeline real-time yang disederhanakan memberikan respons dalam hitungan detik.

• Penyimpanan dan layanan terpadu: Hologres menyediakan lapisan layanan terpadu dengan penyimpanan data terpusat dan antarmuka eksternal yang konsisten (satu antarmuka SQL untuk kueri OLAP dan Key-Value).

• Analitik batch dan real-time terpadu: Mengurangi redundansi dan perpindahan data, serta memungkinkan koreksi data.

Memahami bisnis GitHub

Bisnis inti GitHub adalah mengelola kode dan interaksi. Ini melibatkan tiga entitas utama: Developer, Repository, dan Organization.

Untuk analisis data ini, sebuah Event juga disimpan dan dicatat sebagai entitas.

Memahami data event publik mentah

Contoh berikut menunjukkan data JSON untuk sebuah event mentah:

{
    "id": "19541192931",
    "type": "WatchEvent",
    "actor":
    {
        "id": 23286640,
        "login": "herekeo",
        "display_login": "herekeo",
        "gravatar_id": "",
        "url": "https://api.github.com/users/herekeo",
        "avatar_url": "https://avatars.githubusercontent.com/u/23286640?"
    },
    "repo":
    {
        "id": 52760178,
        "name": "crazyguitar/pysheeet",
        "url": "https://api.github.com/repos/crazyguitar/pysheeet"
    },
    "payload":
    {
        "action": "started"
    },
    "public": true,
    "created_at": "2022-01-01T00:03:04Z"
}

Analisis ini mencakup 15 jenis event publik. Analisis tidak mencakup event yang tidak pernah terjadi atau tidak lagi direkam. Untuk detail jenis event tersebut, lihat Github public event types.

Unduh file data mentah menggunakan instance ECS dan unggah ke OSS

Gunakan instance Elastic Compute Service (ECS) untuk mengunduh file data JSON dari GH Archive.

• Unduh data historis menggunakan perintah wget. Misalnya, jalankan wget https://data.gharchive.org/{2012..2022}-{01..12}-{01..31}-{0..23}.json.gz untuk mengunduh data per jam dari tahun 2012 hingga 2022.

• Untuk mengunduh data baru yang dihasilkan setiap jam, atur tugas terjadwal per jam sebagai berikut.

  Catatan

  • Pastikan ossutil telah diinstal pada instance ECS. Untuk informasi selengkapnya, lihat Instal ossutil. Unduh paket instalasi ossutil dan unggah ke instance ECS. Jalankan yum install unzip untuk menginstal perangkat lunak unzip. Lalu, ekstrak paket ossutil dan pindahkan file yang dapat dieksekusi ke direktori /usr/bin/.

  • Pastikan Anda telah membuat bucket Object Storage Service (OSS) di wilayah yang sama dengan instance ECS Anda. Anda dapat menggunakan nama bucket kustom. Contoh ini menggunakan nama bucket githubevents.

  • Dalam contoh ini, file diunduh ke direktori /opt/hourlydata/gh_data pada instance ECS. Anda dapat menggunakan direktori berbeda.

  1. Jalankan perintah berikut untuk membuat file bernama download_code.sh di direktori /opt/hourlydata.

     cd /opt/hourlydata
     vim download_code.sh

  2. Tekan i untuk masuk ke mode edit dan tambahkan skrip berikut.

     d=$(TZ=UTC date --date='1 hour ago' '+%Y-%m-%d-%-H')
     h=$(TZ=UTC date --date='1 hour ago' '+%Y-%m-%d-%H')
     url=https://data.gharchive.org/${d}.json.gz
     echo ${url}
     
     # Unduh data ke direktori ./gh_data/. Anda dapat menggunakan direktori berbeda.
     wget ${url} -P ./gh_data/
     
     # Pindah ke direktori gh_data.
     cd gh_data
     
     # Ekstrak data yang diunduh menjadi file JSON.
     gzip -d ${d}.json
     
     echo ${d}.json
     
     # Pindah ke direktori root.
     cd /root
     
     # Gunakan ossutil untuk mengunggah data ke OSS.
     # Buat direktori hr=${h} di bucket OSS githubevents.
     ossutil mkdir oss://githubevents/hr=${h}
     
     # Unggah data dari direktori /opt/hourlydata/gh_data ke OSS. Anda dapat menggunakan direktori berbeda.
     ossutil cp -r /opt/hourlydata/gh_data oss://githubevents/hr=${h} -u
     echo oss uploaded successfully!
     
     rm -rf /opt/hourlydata/gh_data/${d}.json
     echo ecs deleted!

  3. Tekan tombol Esc, ketik :wq, lalu tekan Enter untuk menyimpan dan menutup file.

  4. Jalankan perintah berikut untuk mengeksekusi skrip download_code.sh pada menit ke-10 setiap jam.

     # 1. Jalankan perintah berikut dan tekan I untuk masuk ke mode edit.
     crontab -e
     
     # 2. Tambahkan perintah berikut. Lalu, tekan Esc, ketik :wq, dan tekan Enter untuk keluar.
     10 * * * * cd /opt/hourlydata && sh download_code.sh > download.log

     Setelah skrip dijalankan, file JSON dari jam sebelumnya diunduh pada menit ke-10 setiap jam. File tersebut kemudian diekstrak pada instance ECS dan diunggah ke OSS di path oss://githubevents. Untuk hanya membaca file dari jam sebelumnya, direktori bernama 'hr=%Y-%M-%D-%H' dibuat sebagai partisi untuk setiap file selama pengunggahan. Hal ini memastikan bahwa operasi penulisan data selanjutnya hanya membaca file dari partisi terbaru.

Impor data OSS ke MaxCompute menggunakan tabel eksternal

Jalankan perintah berikut di klien MaxCompute atau node ODPS SQL di DataWorks. Untuk informasi selengkapnya, lihat Hubungkan ke MaxCompute menggunakan klien (odpscmd) atau Kembangkan tugas ODPS SQL.

1. Buat tabel eksternal githubevents untuk membaca file JSON yang disimpan di OSS:

   CREATE EXTERNAL TABLE IF NOT EXISTS githubevents
   (
       col  STRING
   )
   PARTITIONED BY 
   (
       hr   STRING
   )
   STORED AS textfile
   LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/githubevents/'
   ;

   Untuk informasi selengkapnya tentang membuat tabel eksternal untuk mengakses data OSS di MaxCompute, lihat Akses data tidak terstruktur di OSS.

2. Buat tabel fakta dwd_github_events_odps untuk menyimpan data. Kode berikut menunjukkan pernyataan Data Definition Language (DDL):

   CREATE TABLE IF NOT EXISTS dwd_github_events_odps
   (
       id                     BIGINT COMMENT 'ID Event'
       ,actor_id              BIGINT COMMENT 'ID inisiator event'
       ,actor_login           STRING COMMENT 'Nama login inisiator event'
       ,repo_id               BIGINT COMMENT 'ID repositori'
       ,repo_name             STRING COMMENT 'Nama lengkap repositori dalam format owner/repository_name'
       ,org_id                BIGINT COMMENT 'ID organisasi tempat repositori berada'
       ,org_login             STRING COMMENT 'Nama organisasi tempat repositori berada'
       ,`type`                STRING COMMENT 'Jenis event'
       ,created_at            DATETIME COMMENT 'Waktu terjadinya event'
       ,action                STRING COMMENT 'Aksi event'
       ,iss_or_pr_id          BIGINT COMMENT 'ID issue atau pull request'
       ,number                BIGINT COMMENT 'Nomor issue atau pull request'
       ,comment_id            BIGINT COMMENT 'ID komentar'
       ,commit_id             STRING COMMENT 'ID commit'
       ,member_id             BIGINT COMMENT 'ID anggota'
       ,rev_or_push_or_rel_id BIGINT COMMENT 'ID review, push, atau rilis'
       ,ref                   STRING COMMENT 'Nama resource yang dibuat atau dihapus'
       ,ref_type              STRING COMMENT 'Jenis resource yang dibuat atau dihapus'
       ,state                 STRING COMMENT 'Status issue, pull request, atau review pull request'
       ,author_association    STRING COMMENT 'Hubungan antara aktor dan repositori'
       ,language              STRING COMMENT 'Bahasa kode dalam pull request'
       ,merged                BOOLEAN COMMENT 'Menunjukkan apakah pull request digabung'
       ,merged_at             DATETIME COMMENT 'Waktu kode digabung'
       ,additions             BIGINT COMMENT 'Jumlah baris kode yang ditambahkan'
       ,deletions             BIGINT COMMENT 'Jumlah baris kode yang dihapus'
       ,changed_files         BIGINT COMMENT 'Jumlah file yang diubah dalam pull request'
       ,push_size             BIGINT COMMENT 'Jumlah commit'
       ,push_distinct_size    BIGINT COMMENT 'Jumlah commit berbeda'
       ,hr                    STRING COMMENT 'Jam terjadinya event. Misalnya, jika event terjadi pukul 00:23, nilai hr adalah 00.'
       ,`month`               STRING COMMENT 'Bulan terjadinya event. Misalnya, jika event terjadi pada Oktober 2015, nilai month adalah 2015-10.'
       ,`year`                STRING COMMENT 'Tahun terjadinya event. Misalnya, jika event terjadi pada 2015, nilai year adalah 2015.'
   )
   PARTITIONED BY 
   (
       ds                     STRING COMMENT 'Tanggal terjadinya event, dalam format yyyy-mm-dd.'
   );

3. Uraikan data JSON dan tulis ke tabel fakta.

   Jalankan perintah berikut untuk menambahkan partisi, mengurai data JSON, dan menuliskan data ke tabel dwd_github_events_odps:

   msck repair table githubevents add partitions;
   
   set odps.sql.hive.compatible = true;
   set odps.sql.split.hive.bridge = true;
   INSERT into TABLE dwd_github_events_odps PARTITION(ds)
   SELECT  CAST(GET_JSON_OBJECT(col,'$.id')  AS BIGINT ) AS id
           ,CAST(GET_JSON_OBJECT(col,'$.actor.id')AS BIGINT) AS actor_id
           ,GET_JSON_OBJECT(col,'$.actor.login') AS actor_login
           ,CAST(GET_JSON_OBJECT(col,'$.repo.id')AS BIGINT) AS repo_id
           ,GET_JSON_OBJECT(col,'$.repo.name') AS repo_name
           ,CAST(GET_JSON_OBJECT(col,'$.org.id')AS BIGINT) AS org_id
           ,GET_JSON_OBJECT(col,'$.org.login') AS org_login
           ,GET_JSON_OBJECT(col,'$.type') as type
           ,to_date(GET_JSON_OBJECT(col,'$.created_at'), 'yyyy-mm-ddThh:mi:ssZ') AS created_at
           ,GET_JSON_OBJECT(col,'$.payload.action') AS action
           ,case    WHEN GET_JSON_OBJECT(col,'$.type')="PullRequestEvent" THEN CAST(GET_JSON_OBJECT(col,'$.payload.pull_request.id')AS BIGINT) 
                    WHEN GET_JSON_OBJECT(col,'$.type')="IssuesEvent" THEN CAST(GET_JSON_OBJECT(col,'$.payload.issue.id')AS BIGINT) 
            END AS iss_or_pr_id
           ,case    WHEN GET_JSON_OBJECT(col,'$.type')="PullRequestEvent" THEN CAST(GET_JSON_OBJECT(col,'$.payload.pull_request.number')AS BIGINT) 
                    WHEN GET_JSON_OBJECT(col,'$.type')="IssuesEvent" THEN CAST(GET_JSON_OBJECT(col,'$.payload.issue.number')AS BIGINT) 
                    ELSE CAST(GET_JSON_OBJECT(col,'$.payload.number')AS BIGINT)
            END AS number
           ,CAST(GET_JSON_OBJECT(col,'$.payload.comment.id')AS BIGINT) AS comment_id
           ,GET_JSON_OBJECT(col,'$.payload.comment.commit_id') AS commit_id
           ,CAST(GET_JSON_OBJECT(col,'$.payload.member.id')AS BIGINT) AS member_id
           ,case    WHEN GET_JSON_OBJECT(col,'$.type')="PullRequestReviewEvent" THEN CAST(GET_JSON_OBJECT(col,'$.payload.review.id')AS BIGINT)
                    WHEN GET_JSON_OBJECT(col,'$.type')="PushEvent" THEN CAST(GET_JSON_OBJECT(col,'$.payload.push_id')AS BIGINT)
                    WHEN GET_JSON_OBJECT(col,'$.type')="ReleaseEvent" THEN CAST(GET_JSON_OBJECT(col,'$.payload.release.id')AS BIGINT)
            END AS rev_or_push_or_rel_id
           ,GET_JSON_OBJECT(col,'$.payload.ref') AS ref
           ,GET_JSON_OBJECT(col,'$.payload.ref_type') AS ref_type
           ,case    WHEN GET_JSON_OBJECT(col,'$.type')="PullRequestEvent" THEN GET_JSON_OBJECT(col,'$.payload.pull_request.state')
                    WHEN GET_JSON_OBJECT(col,'$.type')="IssuesEvent" THEN GET_JSON_OBJECT(col,'$.payload.issue.state')
                    WHEN GET_JSON_OBJECT(col,'$.type')="PullRequestReviewEvent" THEN GET_JSON_OBJECT(col,'$.payload.review.state') 
            END AS state
           ,case    WHEN GET_JSON_OBJECT(col,'$.type')="PullRequestEvent" THEN GET_JSON_OBJECT(col,'$.payload.pull_request.author_association')
                    WHEN GET_JSON_OBJECT(col,'$.type')="IssuesEvent" THEN GET_JSON_OBJECT(col,'$.payload.issue.author_association')
                    WHEN GET_JSON_OBJECT(col,'$.type')="IssueCommentEvent" THEN GET_JSON_OBJECT(col,'$.payload.comment.author_association')
                    WHEN GET_JSON_OBJECT(col,'$.type')="PullRequestReviewEvent" THEN GET_JSON_OBJECT(col,'$.payload.review.author_association') 
            END AS author_association
           ,GET_JSON_OBJECT(col,'$.payload.pull_request.base.repo.language') AS language
           ,CAST(GET_JSON_OBJECT(col,'$.payload.pull_request.merged') AS BOOLEAN) AS merged
           ,to_date(GET_JSON_OBJECT(col,'$.payload.pull_request.merged_at'), 'yyyy-mm-ddThh:mi:ssZ') AS merged_at
           ,CAST(GET_JSON_OBJECT(col,'$.payload.pull_request.additions')AS BIGINT) AS additions
           ,CAST(GET_JSON_OBJECT(col,'$.payload.pull_request.deletions')AS BIGINT)  AS deletions
           ,CAST(GET_JSON_OBJECT(col,'$.payload.pull_request.changed_files')AS BIGINT) AS changed_files
           ,CAST(GET_JSON_OBJECT(col,'$.payload.size')AS BIGINT)  AS push_size
           ,CAST(GET_JSON_OBJECT(col,'$.payload.distinct_size')AS BIGINT)   AS push_distinct_size
           ,SUBSTR(GET_JSON_OBJECT(col,'$.created_at'),12,2) as hr
           ,REPLACE(SUBSTR(GET_JSON_OBJECT(col,'$.created_at'),1,7),'/','-') as month
           ,SUBSTR(GET_JSON_OBJECT(col,'$.created_at'),1,4) as year
           ,REPLACE(SUBSTR(GET_JSON_OBJECT(col,'$.created_at'),1,10),'/','-') as ds
   from githubevents 
   where hr = cast(to_char(dateadd(getdate(),-9,'hh'), 'yyyy-mm-dd-hh') as string);

4. Kueri data.

   Jalankan perintah berikut untuk mengkueri data dari tabel dwd_github_events_odps:

   SET odps.sql.allow.fullscan=true;
   SELECT * FROM dwd_github_events_odps where ds = '2023-03-31' limit 10;

   Hasil sampel berikut dikembalikan:

   

Peroleh data real-time menggunakan ECS

Instance Elastic Compute Service (ECS) digunakan untuk mengekstraksi data event real-time dari API GitHub. Topik ini menyediakan skrip contoh untuk menunjukkan cara mengumpulkan data real-time dari API GitHub.

1. Jalankan perintah berikut untuk membuat file bernama download_realtime_data.py di direktori /opt/realtime.

   cd /opt/realtime
   vim download_realtime_data.py

2. Tekan i untuk masuk ke mode edit dan tambahkan konten contoh berikut ke file tersebut.

   #!python
   
   import requests
   import json
   import sys
   import time
   
   # Dapatkan URL API
   def get_next_link(resp):
       resp_link = resp.headers['link']
       link = ''
       for l in resp_link.split(', '):
           link = l.split('; ')[0][1:-1]
           rel = l.split('; ')[1]
           if rel == 'rel="next"':
               return link
       return None
   
   # Kumpulkan satu halaman data dari API
   def download(link, fname):
   # Definisikan header Accept dan Authorization untuk API GitHub
       headers = {"Accept": "application/vnd.github+json","Authorization": "<Bearer> <github_api_token>"}
       resp = requests.get(link, headers=headers)
   
       if int(resp.status_code) != 200:
           return None
   
       with open(fname, 'a') as f:
           for j in resp.json():
               f.write(json.dumps(j))
               f.write('\n')
   
       print('downloaded {} events to {}'.format(len(resp.json()), fname))
       return resp
   
   # Kumpulkan beberapa halaman data dari API
   def download_all_data(fname):
       link = 'https://api.github.com/events?per_page=100&page=1'
       while True:
           resp = download(link, fname)
           if resp is None:
               break
           link = get_next_link(resp)
           if link is None:
               break
   
   # Definisikan waktu saat ini
   def get_current_ms():
       return round(time.time()*1000)
   
   # Definisikan durasi eksekusi skrip sebagai 1 menit
   def main(fname):
       current_ms = get_current_ms()
       while get_current_ms() - current_ms < 60*1000:
           download_all_data(fname)
           time.sleep(0.1)
   
   # Jalankan skrip
   if __name__ == '__main__':
       if len(sys.argv) < 2:
           print('usage: python {} <log_file>'.format(sys.argv[0]))
           exit(0)
       main(sys.argv[1])

3. Tekan tombol Esc, ketik :wq, lalu tekan Enter untuk menyimpan dan menutup file.

4. Buat file run_py.sh untuk menjalankan download_realtime_data.py dan menyimpan data yang dikumpulkan dari setiap eksekusi secara terpisah. Isinya sebagai berikut.

   python /opt/realtime/download_realtime_data.py /opt/realtime/gh_realtime_data/$(date '+%Y-%m-%d-%H:%M:%S').json

5. Buat file delete_log.sh untuk menghapus data historis. Isinya sebagai berikut.

   d=$(TZ=UTC date --date='2 day ago' '+%Y-%m-%d')
   rm -f /opt/realtime/gh_realtime_data/*${d}*.json

6. Jalankan perintah berikut untuk mengumpulkan data GitHub setiap menit dan menghapus data historis setiap hari.

   #1. Jalankan perintah berikut dan tekan I untuk masuk ke mode edit.
   crontab -e
   
   #2. Tambahkan perintah berikut. Lalu, tekan Esc, ketik :wq, dan tekan Enter untuk keluar.
   * * * * * bash /opt/realtime/run_py.sh
   1 1 * * * bash /opt/realtime/delete_log.sh

Kumpulkan data ECS menggunakan SLS

Simple Log Service (SLS) digunakan untuk mengumpulkan data event real-time yang diekstraksi dari instance ECS sebagai log.

SLS mendukung pengumpulan log dari instance ECS menggunakan Logtail. Karena data dalam topik ini berformat JSON, Anda dapat menggunakan mode JSON Logtail untuk dengan cepat mengumpulkan log JSON inkremental dari instance ECS. Untuk informasi selengkapnya, lihat Kumpulkan log dalam mode JSON. Dalam topik ini, SLS dikonfigurasi untuk mengurai pasangan kunci-nilai tingkat atas dari data mentah.

Setelah konfigurasi selesai, SLS terus-menerus mengumpulkan data event inkremental dari instance ECS. Gambar berikut menunjukkan contoh data yang dikumpulkan.

Tuliskan data SLS ke Hologres secara real-time menggunakan Flink

Flink digunakan untuk menuliskan data log yang dikumpulkan oleh SLS ke Hologres secara real-time. Dengan menggunakan tabel sumber SLS dan tabel hasil Hologres di Flink, Anda dapat menuliskan data dari SLS ke Hologres secara real-time. Untuk informasi selengkapnya, lihat Impor data dari Simple Log Service.

1. Buat tabel internal Hologres.

   Tabel internal yang dibuat dalam topik ini hanya menyimpan beberapa pasangan kunci-nilai dari data JSON mentah. ID event id dan tanggal ds diatur sebagai kunci primer. ID event id diatur sebagai kunci distribusi. Tanggal ds diatur sebagai kunci partisi. Waktu event created_at diatur sebagai event_time_column. Anda dapat membuat indeks untuk bidang lain sesuai kebutuhan untuk meningkatkan efisiensi kueri. Untuk informasi selengkapnya tentang indeks, lihat CREATE TABLE. Pernyataan Data Definition Language (DDL) berikut digunakan untuk membuat tabel dalam contoh ini.

   DROP TABLE IF EXISTS gh_realtime_data;
   
   BEGIN;
   
   CREATE TABLE gh_realtime_data (
       id bigint,
       actor_id bigint,
       actor_login text,
       repo_id bigint,
       repo_name text,
       org_id bigint,
       org_login text,
       type text,
       created_at timestamp with time zone NOT NULL,
       action text,
       iss_or_pr_id bigint,
       number bigint,
       comment_id bigint,
       commit_id text,
       member_id bigint,
       rev_or_push_or_rel_id bigint,
       ref text,
       ref_type text,
       state text,
       author_association text,
       language text,
       merged boolean,
       merged_at timestamp with time zone,
       additions bigint,
       deletions bigint,
       changed_files bigint,
       push_size bigint,
       push_distinct_size bigint,
       hr text,
       month text,
       year text,
       ds text,
       PRIMARY KEY (id,ds)
   )
   PARTITION BY LIST (ds);
   CALL set_table_property('public.gh_realtime_data', 'distribution_key', 'id');
   CALL set_table_property('public.gh_realtime_data', 'event_time_column', 'created_at');
   CALL set_table_property('public.gh_realtime_data', 'clustering_key', 'created_at');
   
   COMMENT ON COLUMN public.gh_realtime_data.id IS 'ID Event';
   COMMENT ON COLUMN public.gh_realtime_data.actor_id IS 'ID inisiator event';
   COMMENT ON COLUMN public.gh_realtime_data.actor_login IS 'Nama login inisiator event';
   COMMENT ON COLUMN public.gh_realtime_data.repo_id IS 'ID repo';
   COMMENT ON COLUMN public.gh_realtime_data.repo_name IS 'Nama repo';
   COMMENT ON COLUMN public.gh_realtime_data.org_id IS 'ID organisasi tempat repo berada';
   COMMENT ON COLUMN public.gh_realtime_data.org_login IS 'Nama organisasi tempat repo berada';
   COMMENT ON COLUMN public.gh_realtime_data.type IS 'Jenis event';
   COMMENT ON COLUMN public.gh_realtime_data.created_at IS 'Waktu terjadinya event';
   COMMENT ON COLUMN public.gh_realtime_data.action IS 'Aksi event';
   COMMENT ON COLUMN public.gh_realtime_data.iss_or_pr_id IS 'ID issue/pull_request';
   COMMENT ON COLUMN public.gh_realtime_data.number IS 'Nomor issue/pull_request';
   COMMENT ON COLUMN public.gh_realtime_data.comment_id IS 'ID komentar';
   COMMENT ON COLUMN public.gh_realtime_data.commit_id IS 'ID Commit';
   COMMENT ON COLUMN public.gh_realtime_data.member_id IS 'ID Anggota';
   COMMENT ON COLUMN public.gh_realtime_data.rev_or_push_or_rel_id IS 'ID review/push/rilis';
   COMMENT ON COLUMN public.gh_realtime_data.ref IS 'Nama resource yang dibuat atau dihapus';
   COMMENT ON COLUMN public.gh_realtime_data.ref_type IS 'Jenis resource yang dibuat atau dihapus';
   COMMENT ON COLUMN public.gh_realtime_data.state IS 'Status issue/pull_request/review pull_request';
   COMMENT ON COLUMN public.gh_realtime_data.author_association IS 'Hubungan antara aktor dan repo';
   COMMENT ON COLUMN public.gh_realtime_data.language IS 'Bahasa pemrograman';
   COMMENT ON COLUMN public.gh_realtime_data.merged IS 'Menentukan apakah merge diterima';
   COMMENT ON COLUMN public.gh_realtime_data.merged_at IS 'Waktu kode digabung';
   COMMENT ON COLUMN public.gh_realtime_data.additions IS 'Jumlah baris kode yang ditambahkan';
   COMMENT ON COLUMN public.gh_realtime_data.deletions IS 'Jumlah baris kode yang dihapus';
   COMMENT ON COLUMN public.gh_realtime_data.changed_files IS 'Jumlah file yang diubah dalam pull request';
   COMMENT ON COLUMN public.gh_realtime_data.push_size IS 'Jumlah push';
   COMMENT ON COLUMN public.gh_realtime_data.push_distinct_size IS 'Jumlah push berbeda';
   COMMENT ON COLUMN public.gh_realtime_data.hr IS 'Jam terjadinya event. Misalnya, jika waktu adalah 00:23, hr=00.';
   COMMENT ON COLUMN public.gh_realtime_data.month IS 'Bulan terjadinya event. Misalnya, jika tanggal adalah Oktober 2015, month=2015-10.';
   COMMENT ON COLUMN public.gh_realtime_data.year IS 'Tahun terjadinya event. Misalnya, jika tahun adalah 2015, year=2015.';
   COMMENT ON COLUMN public.gh_realtime_data.ds IS 'Hari terjadinya event. ds=yyyy-mm-dd.';
   
   COMMIT;

2. Tuliskan data secara real-time menggunakan Flink.

   Gunakan Flink untuk mengurai lebih lanjut data SLS dan menuliskannya ke Hologres secara real-time. Gunakan pernyataan berikut di Flink untuk memfilter data yang akan dituliskan. Hal ini membuang data kotor di mana ID event atau waktu event (created_at) bernilai null dan hanya menyimpan data event terbaru.

   CREATE TEMPORARY TABLE sls_input (
     actor varchar,
     created_at varchar,
     id bigint,
     org varchar,
     payload varchar,
     public varchar,
     repo varchar,
     type varchar
     )
   WITH (
       'connector' = 'sls',
       'endpoint' = '<endpoint>',--Titik akhir pribadi SLS
       'accessid' = '<accesskey id>',--ID AccessKey akun Anda
       'accesskey' = '<accesskey secret>',--Rahasia AccessKey akun Anda
       'project' = '<project name>',--Nama proyek SLS
       'logstore' = '<logstore name>'--Nama Logstore SLS
       'starttime' = '2023-04-06 00:00:00',--Waktu mulai pengumpulan data SLS
   );
   
   CREATE TEMPORARY TABLE hologres_sink (
       id bigint,
       actor_id bigint,
       actor_login string,
       repo_id bigint,
       repo_name string,
       org_id bigint,
       org_login string,
       type string,
       created_at timestamp,
       action string,
       iss_or_pr_id bigint,
       number bigint,
       comment_id bigint,
       commit_id string,
       member_id bigint,
       rev_or_push_or_rel_id bigint,
       `ref` string,
       ref_type string,
       state string,
       author_association string,
       `language` string,
       merged boolean,
       merged_at timestamp,
       additions bigint,
       deletions bigint,
       changed_files bigint,
       push_size bigint,
       push_distinct_size bigint,
       hr string,
       `month` string,
       `year` string,
       ds string
       )
   WITH (
       'connector' = 'hologres',
       'dbname' = '<hologres dbname>', --Nama database Hologres
       'tablename' = '<hologres tablename>', --Nama tabel Hologres yang menerima data
       'username' = '<accesskey id>', --ID AccessKey akun Alibaba Cloud saat ini
       'password' = '<accesskey secret>', --Rahasia AccessKey akun Alibaba Cloud saat ini
       'endpoint' = '<endpoint>', --Titik akhir VPC instance Hologres saat ini
       'jdbcretrycount' = '1', --Jumlah percobaan ulang saat koneksi gagal
       'partitionrouter' = 'true', --Menentukan apakah akan menuliskan data ke tabel partisi
       'createparttable' = 'true', --Menentukan apakah akan membuat partisi secara otomatis
       'mutatetype' = 'insertorignore' --Mode penulisan data
   );
   
   INSERT INTO hologres_sink
   SELECT id
           ,CAST(JSON_VALUE(actor, '$.id') AS bigint) AS actor_id
           ,JSON_VALUE(actor, '$.login') AS actor_login
           ,CAST(JSON_VALUE(repo, '$.id') AS bigint) AS repo_id
           ,JSON_VALUE(repo, '$.name') AS repo_name
           ,CAST(JSON_VALUE(org, '$.id') AS bigint) AS org_id
           ,JSON_VALUE(org, '$.login') AS org_login
           ,type
           ,TO_TIMESTAMP_TZ(replace(created_at,'T',' '), 'yyyy-MM-dd HH:mm:ss', 'UTC') AS created_at
           ,JSON_VALUE(payload, '$.action') AS action
           ,CASE    WHEN type='PullRequestEvent' THEN CAST(JSON_VALUE(payload, '$.pull_request.id') AS bigint)
                    WHEN type='IssuesEvent' THEN CAST(JSON_VALUE(payload, '$.issue.id') AS bigint)
            END AS iss_or_pr_id
           ,CASE    WHEN type='PullRequestEvent' THEN CAST(JSON_VALUE(payload, '$.pull_request.number') AS bigint)
                    WHEN type='IssuesEvent' THEN CAST(JSON_VALUE(payload, '$.issue.number') AS bigint)
                    ELSE CAST(JSON_VALUE(payload, '$.number') AS bigint)
            END AS number
           ,CAST(JSON_VALUE(payload, '$.comment.id') AS bigint) AS comment_id
           ,JSON_VALUE(payload, '$.comment.commit_id') AS commit_id
           ,CAST(JSON_VALUE(payload, '$.member.id') AS bigint) AS member_id
           ,CASE    WHEN type='PullRequestReviewEvent' THEN CAST(JSON_VALUE(payload, '$.review.id') AS bigint)
                    WHEN type='PushEvent' THEN CAST(JSON_VALUE(payload, '$.push_id') AS bigint)
                    WHEN type='ReleaseEvent' THEN CAST(JSON_VALUE(payload, '$.release.id') AS bigint)
            END AS rev_or_push_or_rel_id
           ,JSON_VALUE(payload, '$.ref') AS `ref`
           ,JSON_VALUE(payload, '$.ref_type') AS ref_type
           ,CASE    WHEN type='PullRequestEvent' THEN JSON_VALUE(payload, '$.pull_request.state')
                    WHEN type='IssuesEvent' THEN JSON_VALUE(payload, '$.issue.state')
                    WHEN type='PullRequestReviewEvent' THEN JSON_VALUE(payload, '$.review.state')
            END AS state
           ,CASE    WHEN type='PullRequestEvent' THEN JSON_VALUE(payload, '$.pull_request.author_association')
                    WHEN type='IssuesEvent' THEN JSON_VALUE(payload, '$.issue.author_association')
                    WHEN type='IssueCommentEvent' THEN JSON_VALUE(payload, '$.comment.author_association')
                    WHEN type='PullRequestReviewEvent' THEN JSON_VALUE(payload, '$.review.author_association')
            END AS author_association
           ,JSON_VALUE(payload, '$.pull_request.base.repo.language') AS `language`
           ,CAST(JSON_VALUE(payload, '$.pull_request.merged') AS boolean) AS merged
           ,TO_TIMESTAMP_TZ(replace(JSON_VALUE(payload, '$.pull_request.merged_at'),'T',' '), 'yyyy-MM-dd HH:mm:ss', 'UTC') AS merged_at
           ,CAST(JSON_VALUE(payload, '$.pull_request.additions') AS bigint) AS additions
           ,CAST(JSON_VALUE(payload, '$.pull_request.deletions') AS bigint) AS deletions
           ,CAST(JSON_VALUE(payload, '$.pull_request.changed_files') AS bigint) AS changed_files
           ,CAST(JSON_VALUE(payload, '$.size') AS bigint) AS push_size
           ,CAST(JSON_VALUE(payload, '$.distinct_size') AS bigint) AS push_distinct_size
           ,SUBSTRING(TO_TIMESTAMP_TZ(replace(created_at,'T',' '), 'yyyy-MM-dd HH:mm:ss', 'UTC'),12,2) as hr
           ,REPLACE(SUBSTRING(TO_TIMESTAMP_TZ(replace(created_at,'T',' '), 'yyyy-MM-dd HH:mm:ss', 'UTC'),1,7),'/','-') as `month`
           ,SUBSTRING(TO_TIMESTAMP_TZ(replace(created_at,'T',' '), 'yyyy-MM-dd HH:mm:ss', 'UTC'),1,4) as `year`
           ,SUBSTRING(TO_TIMESTAMP_TZ(replace(created_at,'T',' '), 'yyyy-MM-dd HH:mm:ss', 'UTC'),1,10) as ds
   FROM
           sls_input
   WHERE
           id IS NOT NULL
         	AND created_at IS NOT NULL
           AND to_date(replace(created_at,'T',' ')) >= date_add(CURRENT_DATE, -1); 

   Untuk informasi selengkapnya tentang parameter, lihat Simple Log Service (SLS) dan Hologres.

   Catatan

   Data event mentah dari GitHub menggunakan zona waktu UTC dan tidak memiliki atribut zona waktu. Zona waktu default Hologres adalah UTC+8. Oleh karena itu, sesuaikan zona waktu saat menuliskan data dari Flink ke Hologres secara real-time. Tetapkan atribut zona waktu UTC ke data tabel sumber di Flink SQL. Langkah-langkahnya sebagai berikut:

   Langkah 1: Buka halaman pengeditan pekerjaan

   • Masuk ke Konsol Realtime Compute for Apache Flink

   • Buka ruang kerja target

   • Temukan pekerjaan Flink SQL atau JAR Anda dan klik Edit

   Langkah 2: Buka tab Detail Penyebaran

   ⚠️ Perubahan penting: "Konfigurasi Flink" tidak lagi menjadi bagian terpisah. Bagian ini digabung ke dalam "Detail Penyebaran".

   • Di bagian atas halaman pengeditan pekerjaan, beralihlah ke tab Detail Penyebaran.

   • Gulir ke bawah ke bagian Konfigurasi Parameter.

   Langkah 3: Tambahkan konfigurasi kustom

   • Klik tombol Edit di sebelah kanan Konfigurasi Parameter.

   • Pada kotak dialog yang muncul, temukan kotak teks Other Configuration.

   • Di kotak teks tersebut, tambahkan parameter Flink table.local-time-zone:Asia/Shanghai sebagai pasangan kunci-nilai untuk mengatur zona waktu sistem Flink ke Asia/Shanghai.

3. Kueri data.

   Kueri data SLS yang dituliskan ke Hologres melalui Flink. Anda kemudian dapat melakukan pengembangan data sesuai kebutuhan.

   SELECT * FROM public.gh_realtime_data limit 10;

   Hasil berikut merupakan contoh:

   

Koreksi data real-time menggunakan data offline

Dalam skenario yang dijelaskan dalam topik ini, data real-time mungkin hilang. Anda dapat menggunakan data offline untuk mengoreksi data real-time. Langkah-langkah berikut menunjukkan cara mengoreksi data real-time dari hari sebelumnya. Anda dapat menyesuaikan periode koreksi data sesuai kebutuhan.

1. Buat tabel eksternal di Hologres untuk memperoleh data offline MaxCompute.

   IMPORT FOREIGN SCHEMA <maxcompute_project_name> LIMIT to
   (
       <foreign_table_name>
   ) 
   FROM SERVER odps_server INTO public OPTIONS(if_table_exist 'update',if_unsupported_type 'error');

   Untuk informasi selengkapnya tentang parameter, lihat IMPORT FOREIGN SCHEMA.

2. Buat tabel sementara untuk mengoreksi data real-time dari hari sebelumnya dengan data offline.

   Catatan

   -- Bersihkan tabel sementara yang mungkin ada
   DROP TABLE IF EXISTS gh_realtime_data_tmp;
   
   -- Buat tabel sementara
   SET hg_experimental_enable_create_table_like_properties = ON;
   CALL HG_CREATE_TABLE_LIKE ('gh_realtime_data_tmp', 'select * from gh_realtime_data');
   
   -- (Opsional) Gunakan Komputasi Tanpa Server untuk melakukan impor data offline skala besar dan pekerjaan ETL.
   SET hg_computing_resource = 'serverless';
   
   -- Masukkan data ke tabel sementara dan perbarui statistik
   INSERT INTO gh_realtime_data_tmp
   SELECT
       *
   FROM
       <foreign_table_name>
   WHERE
       ds = current_date - interval '1 day'
   ON CONFLICT (id, ds)
       DO NOTHING;
   ANALYZE gh_realtime_data_tmp;
   
   -- Atur ulang konfigurasi untuk memastikan pernyataan SQL non-esensial tidak menggunakan sumber daya Tanpa Server.
   RESET hg_computing_resource;
   
   -- Ganti tabel atomik dengan tabel anak sementara yang ada
   BEGIN;
   DROP TABLE IF EXISTS "gh_realtime_data_<yesterday_date>";
   ALTER TABLE gh_realtime_data_tmp RENAME TO "gh_realtime_data_<yesterday_date>";
   ALTER TABLE gh_realtime_data ATTACH PARTITION "gh_realtime_data_<yesterday_date>" FOR VALUES IN ('<yesterday_date>');
   COMMIT;