MaxCompute：Integrasi pemrosesan data offline dan real-time dalam dataset peristiwa publik GitHub

Instance Elastic Compute Service (ECS) mengumpulkan data peristiwa real-time dan offline dari GitHub sebagai sumber data. Data tersebut kemudian diproses secara terpisah dan ditulis ke Hologres. Hologres menyediakan layanan data ke aplikasi eksternal.

• Pemrosesan real-time: Realtime Compute for Apache Flink memproses data real-time dalam Layanan Log dan menulis data hasil pemrosesan ke Hologres. Realtime Compute for Apache Flink adalah mesin komputasi aliran yang kuat. Hologres mendukung penulisan dan pembaruan data real-time, serta memungkinkan query instan setelah data ditulis. Integrasi native antara Realtime Compute for Apache Flink dan Hologres mendukung gudang data real-time dengan throughput tinggi, latensi rendah, pemodelan, dan kinerja berkualitas tinggi. Gudang data real-time memenuhi persyaratan wawasan bisnis real-time dalam skenario seperti ekstraksi peristiwa terbaru dan analisis tren peristiwa.

• Pemrosesan offline: MaxCompute memproses dan mengarsipkan sejumlah besar data offline. Object Storage Service (OSS) adalah layanan penyimpanan cloud yang disediakan oleh Alibaba Cloud. OSS dapat digunakan untuk menyimpan berbagai jenis data. Data mentah dalam praktik ini berada dalam format JSON. OSS menyediakan penyimpanan yang nyaman, aman, hemat biaya, dan andal. MaxCompute adalah gudang data perangkat lunak sebagai layanan (SaaS) tingkat perusahaan yang cocok untuk analitik data. MaxCompute dapat langsung membaca dan mengurai data semi-terstruktur dalam OSS menggunakan tabel eksternal, menyimpan data bernilai tinggi, dan kemudian menggunakan DataWorks untuk mengembangkan data serta membangun gudang data offline.

• Hologres terintegrasi dengan MaxCompute secara mulus di lapisan bawah. Anda dapat menggunakan Hologres untuk mempercepat query dan analitik sejumlah besar data historis dalam MaxCompute guna memenuhi kebutuhan bisnis Anda untuk query historis frekuensi rendah namun kinerja tinggi. Anda juga dapat dengan mudah memodifikasi data real-time melalui pemrosesan offline untuk menyelesaikan masalah seperti kehilangan data yang mungkin terjadi dalam tautan real-time.

Solusi gudang data terintegrasi memberikan keuntungan sebagai berikut:

• Pemrosesan offline yang stabil dan efisien: Data dapat ditulis dan diperbarui setiap jam. Sejumlah besar data dapat diproses dalam batch untuk komputasi dan analitik kompleks, mengurangi biaya komputasi serta meningkatkan efisiensi pemrosesan data.

• Pemrosesan real-time yang matang: Penulisan real-time, komputasi peristiwa real-time, dan analitik real-time didukung. Data real-time dapat di-query dalam hitungan detik.

• Penyimpanan dan layanan terpadu: Hologres digunakan untuk menyediakan layanan. Hologres menyimpan data secara terpusat dan menyediakan antarmuka SQL terpadu untuk query pemrosesan analitik online (OLAP) serta query pasangan kunci-nilai.

• Integrasi data real-time dan offline: Redundansi data lebih sedikit, migrasi data jarang dilakukan, dan data dapat dikoreksi dengan mudah.

Pengenalan GitHub

GitHub digunakan untuk manajemen kode dan juga berfungsi sebagai platform komunikasi antar pengembang. GitHub melibatkan tiga objek entitas level-1: pengembang, repositori, dan organisasi.

Dalam praktik ini, events disimpan dan dicatat sebagai objek entitas.

Pengenalan data peristiwa publik asli

Data sampel dalam format JSON dari peristiwa asli:

{
    "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"
}

Dalam contoh ini, 15 jenis peristiwa publik terlibat, tidak termasuk peristiwa yang tidak terjadi atau tidak lagi dicatat. Untuk informasi lebih lanjut tentang jenis peristiwa dan deskripsi, lihat Jenis Peristiwa GitHub.

Unduh file data mentah menggunakan instance ECS dan unggah file ke OSS

Anda dapat menggunakan instance ECS untuk mengunduh file data dalam format JSON dari GH Archive. Jika Anda ingin mengunduh data historis, Anda dapat menjalankan perintah wget. Misalnya, Anda dapat menjalankan perintah wget https://data.gharchive.org/{2012..2022}-{01..12}-{01..31}-{0..23}.json.gz untuk mengunduh data yang diarsipkan setiap jam dari tahun 2012 hingga tahun 2022. Jika Anda ingin mengunduh data real-time yang diarsipkan setiap jam, lakukan langkah-langkah berikut untuk mengonfigurasi tugas terjadwal yang berjalan setiap jam.

1. Jalankan perintah berikut untuk membuat file bernama download_code.sh:

   vim download_code.sh

2. Masukkan i dalam file untuk masuk ke mode edit dan jalankan skrip. Contoh skrip:

   Catatan

   Sebelum menjalankan skrip, pastikan bahwa alat ossutil diinstal dalam instance ECS. Untuk informasi lebih lanjut, lihat Instal ossutil. Dalam contoh ini, nama Bucket OSS adalah githubevents.

   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}
   
   wget ${url} -P ./gh_data/
   cd gh_data
   gzip -d ${d}.json
   
   echo ${d}.json
   # Gunakan alat ossutil untuk mengunggah data ke OSS.
   cd /root
   ./ossutil64 mkdir oss://githubevents/hr=${h}
   ./ossutil64 cp -r /hourlydata/gh_data oss://githubevents/hr=${h} -u
   echo oss uploaded successfully!
   
   rm -rf /hourlydata/gh_data/${d}.json
   echo ecs deleted!

3. Tekan tombol Esc untuk keluar dari mode edit. Masukkan :wq dan tekan tombol Enter untuk menyimpan dan menutup skrip.

4. Jalankan perintah berikut untuk mengonfigurasi tugas terjadwal di mana skrip download_code.sh dijalankan pada menit ke-10 setiap jam.

   crontab -e
   10 * * * * cd /hourlydata && sh download_code.sh > download.log

   Pada menit ke-10 setiap jam, file JSON yang diarsipkan pada jam sebelumnya diunduh dan didekompresi ke path oss://githubevents OSS. Untuk memastikan bahwa MaxCompute hanya membaca file yang diarsipkan pada jam sebelumnya, kami sarankan Anda membuat partisi dengan nama dalam format ‘hr=%Y-%M-%D-%H’ untuk setiap file. Ini memastikan bahwa MaxCompute membaca data dari partisi terbaru.

Impor data dari OSS ke MaxCompute menggunakan tabel eksternal

Lakukan langkah-langkah berikut pada klien MaxCompute atau node ODPS SQL di konsol DataWorks. Untuk informasi lebih lanjut, lihat Klien MaxCompute (odpscmd) atau Kembangkan Tugas SQL MaxCompute.

1. Buat tabel eksternal bernama githubevents untuk mengonversi file JSON yang disimpan dalam 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 lebih lanjut tentang cara membuat tabel eksternal OSS dalam MaxCompute, lihat Tabel Eksternal ORC.

2. Buat tabel fakta bernama dwd_github_events_odps untuk menyimpan data. Contoh pernyataan bahasa definisi data (DDL):

   CREATE TABLE IF NOT EXISTS dwd_github_events_odps
   (
       id                     BIGINT COMMENT 'ID peristiwa'
       ,actor_id BIGINT COMMENT 'ID aktor peristiwa'
       ,actor_login           STRING COMMENT 'Nama pengguna aktor peristiwa'
       ,repo_id               BIGINT COMMENT 'ID repositori'
       ,repo_name             STRING COMMENT 'Nama repositori lengkap dalam format pemilik/Nama_repositori'
       ,org_id                BIGINT COMMENT 'ID organisasi tempat repositori milik'
       ,org_login             STRING COMMENT 'Nama organisasi tempat repositori milik'
       ,`type`                STRING COMMENT 'Jenis peristiwa'
       ,created_at            DATETIME COMMENT 'Waktu terjadinya peristiwa'
       ,action                STRING COMMENT 'Aksi peristiwa'
       ,iss_or_pr_id          BIGINT COMMENT 'ID isu atau pull_request'
       ,number                BIGINT COMMENT 'Nomor urut permintaan isu 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 tinjauan, push, atau rilis'
       ,ref                   STRING COMMENT 'Nama sumber daya yang dibuat atau dihapus'
       ,ref_type              STRING COMMENT 'Jenis sumber daya yang dibuat atau dihapus'
       ,state                 STRING COMMENT 'Status isu, pull_request, atau permintaan tinjauan pull_request'
       ,author_association    STRING COMMENT 'Hubungan antara aktor dan repositori'
       ,language              STRING COMMENT 'Bahasa yang digunakan untuk menggabungkan kode permintaan'
       ,merged                BOOLEAN COMMENT 'Menentukan apakah penggabungan diizinkan'
       ,merged_at             DATETIME COMMENT 'Waktu penggabungan kode'
       ,additions             BIGINT COMMENT 'Jumlah baris yang ditambahkan ke kode'
       ,deletions             BIGINT COMMENT 'Jumlah baris yang dihapus dari kode'
       ,changed_files         BIGINT COMMENT 'Jumlah file yang diubah oleh pull request'
       ,push_size             BIGINT COMMENT 'Ukuran push'
       ,push_distinct_size    BIGINT COMMENT 'Ukuran push yang berbeda'
       ,hr                    STRING COMMENT 'Jam terjadinya peristiwa. Misalnya, jika peristiwa terjadi pada 00:23, nilai parameter ini adalah 00.
       ,`month`               STRING COMMENT 'Bulan terjadinya peristiwa. Misalnya, jika peristiwa terjadi pada Oktober 2015, nilai parameter ini adalah 2015-10.'
       ,`year`                STRING COMMENT 'Tahun terjadinya peristiwa. Misalnya, jika peristiwa terjadi pada tahun 2015, nilai parameter ini adalah 2015.'
   )
   PARTITIONED BY 
   (
       ds                     STRING COMMENT 'Hari terjadinya peristiwa. Nilai parameter ini dalam format yyyy-mm-dd.'
   )
   ;

3. Parsing data dalam format JSON dan tulis ke tabel fakta.

   Jalankan perintah berikut untuk menambahkan partisi, parsing data dalam format JSON, dan menulis data yang diuraikan 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. Query data.

   Jalankan perintah berikut untuk query data dari tabel dwd_github_events_odps:

   set odps.sql.allow.fullscan=true;
   select * from dwd_github_events_odps limit 10;

   Gambar berikut menunjukkan hasil yang dikembalikan.

   

Gunakan instance ECS untuk mengumpulkan data real-time

Anda dapat menggunakan instance ECS untuk mengekstrak data peristiwa real-time dari API GitHub. Bagian ini menjelaskan cara mengumpulkan data real-time dari API GitHub.

1. Jalankan perintah berikut untuk membuat file bernama download_realtime_data.py.

   vim download_realtime_data.py

2. Masukkan i dalam file untuk masuk ke mode edit. Kemudian, tambahkan kode contoh berikut.

   #!python
   
   import requests
   import json
   import sys
   import time
   
   # Dapatkan URL API GitHub.
   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 GitHub.
   def download(link, fname):
     	# Tentukan 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 GitHub.
   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
   
   # Tentukan waktu saat ini.
   def get_current_ms():
       return round(time.time()*1000)
   
   # Tetapkan durasi setiap eksekusi skrip menjadi 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 untuk keluar dari mode edit. Masukkan :wq dan tekan tombol Enter untuk menyimpan dan menutup skrip.

4. Buat file bernama run_py.sh dan gunakan file tersebut untuk menjalankan skrip dalam file download_realtime_data.py serta menyimpan data yang dikumpulkan secara terpisah berdasarkan waktu eksekusi. Contoh kode:

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

5. Buat file delete_log.sh yang digunakan untuk menghapus data historis. Contoh kode:

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

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

   crontab -e
   * * * * * bash /root/run_py.sh
   1 1 * * * bash /root/delete_log.sh

Gunakan Layanan Log untuk mengumpulkan data dari instance ECS

Anda dapat menggunakan Layanan Log untuk mengumpulkan data peristiwa real-time yang diekstraksi oleh instance ECS dan menyimpannya sebagai log.

Layanan Log menyediakan Logtail untuk mengumpulkan data log dari instance ECS. Data sampel dalam topik ini berada dalam format JSON. Anda dapat menggunakan mode JSON Logtail untuk dengan cepat mengumpulkan log JSON inkremental dari instance ECS. Untuk informasi lebih lanjut, lihat Kumpulkan Log dalam Mode JSON. Dalam topik ini, Layanan Log digunakan untuk mengurai pasangan kunci-nilai data mentah yang termasuk dalam tingkat atas.

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

Gunakan Realtime Compute for Apache Flink untuk menulis data dari Layanan Log ke Hologres

Anda dapat menggunakan Realtime Compute for Apache Flink untuk menulis data log yang dikumpulkan oleh Layanan Log ke Hologres secara real-time. Dalam proses ini, tabel sumber Layanan Log dan tabel hasil Hologres digunakan dalam Realtime Compute for Apache Flink. Untuk informasi lebih lanjut, lihat Impor Data dari Layanan Log Sederhana.

1. Buat tabel internal Hologres.

   Dalam contoh ini, hanya beberapa kunci dari data mentah dalam format JSON yang disimpan dalam tabel internal Hologres. ID peristiwa yang ditentukan oleh id dikonfigurasi sebagai kunci distribusi, tanggal yang ditentukan oleh ds dikonfigurasi sebagai kunci partisi, dan waktu terjadinya peristiwa yang ditentukan oleh created_at dikonfigurasi sebagai event_time_column. id dan ds adalah kunci utama. Anda dapat membuat indeks untuk bidang lain dalam tabel internal Hologres berdasarkan kebutuhan query Anda. Ini membantu meningkatkan efisiensi query. Untuk informasi lebih lanjut tentang indeks, lihat CREATE TABLE. Contoh pernyataan DDL:

   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 peristiwa';
   COMMENT ON COLUMN public.gh_realtime_data.actor_id IS 'ID aktor';
   COMMENT ON COLUMN public.gh_realtime_data.actor_login IS 'Nama pengguna aktor peristiwa';
   COMMENT ON COLUMN public.gh_realtime_data.repo_id IS 'repoID';
   COMMENT ON COLUMN public.gh_realtime_data.repo_name IS 'Nama repositori';
   COMMENT ON COLUMN public.gh_realtime_data.org_id IS 'ID organisasi tempat repositori milik';
   COMMENT ON COLUMN public.gh_realtime_data.org_login IS 'Nama organisasi tempat repositori milik';
   COMMENT ON COLUMN public.gh_realtime_data.type IS 'Jenis peristiwa';
   COMMENT ON COLUMN public.gh_realtime_data.created_at IS 'Waktu terjadinya peristiwa.';
   COMMENT ON COLUMN public.gh_realtime_data.action IS 'Aksi peristiwa';
   COMMENT ON COLUMN public.gh_realtime_data.iss_or_pr_id IS 'ID isu atau pull_request';
   COMMENT ON COLUMN public.gh_realtime_data.number IS 'Nomor urut isu atau 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 tinjauan, push, atau rilis';
   COMMENT ON COLUMN public.gh_realtime_data.ref IS 'Nama sumber daya yang dibuat atau dihapus';
   COMMENT ON COLUMN public.gh_realtime_data.ref_type IS 'Jenis sumber daya yang dibuat atau dihapus';
   COMMENT ON COLUMN public.gh_realtime_data.state IS 'Status isu, pull_request, atau permintaan tinjauan pull_request';
   COMMENT ON COLUMN public.gh_realtime_data.author_association IS 'Hubungan antara aktor dan repositori';
   COMMENT ON COLUMN public.gh_realtime_data.language IS 'Bahasa pemrograman';
   COMMENT ON COLUMN public.gh_realtime_data.merged IS 'Menentukan apakah penggabungan diizinkan';
   COMMENT ON COLUMN public.gh_realtime_data.merged_at IS 'Waktu penggabungan kode';
   COMMENT ON COLUMN public.gh_realtime_data.additions IS 'Jumlah baris yang ditambahkan ke kode';
   COMMENT ON COLUMN public.gh_realtime_data.deletions IS 'Jumlah baris yang dihapus dari kode';
   COMMENT ON COLUMN public.gh_realtime_data.changed_files IS 'Jumlah file yang diubah oleh pull request';
   COMMENT ON COLUMN public.gh_realtime_data.push_size IS 'Ukuran push';
   COMMENT ON COLUMN public.gh_realtime_data.push_distinct_size IS 'Ukuran push yang berbeda';
   COMMENT ON COLUMN public.gh_realtime_data.hr IS 'Jam terjadinya peristiwa. Misalnya, jika peristiwa terjadi pada 00:23, nilai parameter ini adalah 00.';
   COMMENT ON COLUMN public.gh_realtime_data.month IS 'Bulan terjadinya peristiwa. Misalnya, jika peristiwa terjadi pada Oktober 2015, nilai parameter ini adalah 2015-10.';
   COMMENT ON COLUMN public.gh_realtime_data.year IS 'Tahun terjadinya peristiwa. Misalnya, jika peristiwa terjadi pada tahun 2015, nilai parameter ini adalah 2015.';
   COMMENT ON COLUMN public.gh_realtime_data.ds IS 'Hari terjadinya peristiwa. Nilai parameter ini dalam format yyyy-mm-dd.';
   
   COMMIT;

2. Tulis data real-time menggunakan Realtime Compute for Apache Flink.

   Anda dapat menggunakan Realtime Compute for Apache Flink untuk mem-parsing lebih lanjut data dari Layanan Log dan menulis data yang diparsing ke Hologres secara real-time. Eksekusi pernyataan berikut dalam Realtime Compute for Apache Flink untuk memfilter data yang ingin Anda tulis ke Hologres. Dalam penyaringan data, data kotor seperti ID peristiwa dan waktu terjadinya peristiwa yang ditentukan oleh created_at dibuang. Hanya data peristiwa yang terjadi baru-baru ini yang disimpan.

   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>',-- Endpoint internal yang digunakan untuk mengakses Layanan Log.
       'accesssid'= '<accesskey id>',-- ID AccessKey akun Anda.
       'accesskey' = '<accesskey secret>',-- Rahasia AccessKey akun Anda.
       'project' = '<project name>',-- Nama proyek Layanan Log.
       'logstore' = '<logstore name>'-- Nama Logstore dalam Layanan Log.
   );
   
   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 ke mana Anda ingin menulis data.
       'username' = '<accesskey id>', -- ID AccessKey akun Alibaba Cloud Anda.
       'password' = '<accesskey secret>', -- Rahasia AccessKey akun Alibaba Cloud Anda.
       'endpoint' = '<endpoint>', -- Endpoint virtual private cloud (VPC) instance Hologres Anda.
       'jdbcretrycount' = '1', -- Jumlah maksimum percobaan ulang yang diizinkan jika koneksi gagal.
       'partitionrouter' = 'true', -- Menentukan apakah akan menulis data ke tabel yang dipartisi.
       '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 lebih lanjut tentang parameter, lihat Buat Tabel Sumber Layanan Log dan Buat Tabel Hasil Hologres.

   Catatan

   GitHub mencatat data peristiwa mentah dalam zona waktu UTC dan tidak menyertakan properti zona waktu dalam data. Namun, Hologres menggunakan zona waktu UTC+8 secara default. Saat Anda menggunakan Realtime Compute for Apache Flink untuk menulis data real-time ke Hologres, Anda harus mengonversi zona waktu untuk data dengan melakukan operasi berikut: Tambahkan properti zona waktu UTC ke data dalam tabel sumber di Flink SQL. Di bagian Konfigurasi Flink halaman Konfigurasi Mulai Penyebaran, tambahkan pernyataan table.local-time-zone:Asia/Shanghai untuk menetapkan zona waktu Realtime Compute for Apache Flink ke Asia/Shanghai.

3. Query data.

   Anda dapat query data yang ditulis dari Layanan Log ke Hologres menggunakan Realtime Compute for Apache Flink. Anda juga dapat mengembangkan data berdasarkan kebutuhan bisnis Anda.

   SELECT * FROM public.gh_realtime_data limit 10;

   Gambar berikut menunjukkan hasil yang dikembalikan.

   

Gunakan data offline untuk memperbaiki data real-time

Dalam skenario yang dijelaskan dalam topik ini, data real-time mungkin tidak lengkap. Anda dapat menggunakan data offline untuk memperbaiki data real-time. Bagian berikut menjelaskan cara memperbaiki data real-time yang dihasilkan pada hari sebelumnya. Anda dapat memodifikasi periode koreksi data berdasarkan kebutuhan bisnis Anda.

1. Buat tabel asing di Hologres untuk mendapatkan data offline dari 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 lebih lanjut tentang parameter, lihat IMPORT FOREIGN SCHEMA.

2. Buat tabel sementara dan gunakan data offline untuk memperbaiki data real-time yang dihasilkan pada hari sebelumnya.

   -- Hapus tabel sementara jika 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');
   
   -- 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;
   
   -- Ganti tabel anak asli dengan tabel anak sementara.
   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;