このチュートリアルでは、GitHub イベントデータを使用して、統合されたバッチおよびリアルタイム分析ソリューションを構築する方法を説明します。MaxCompute を使用してバッチデータウェアハウスを構築し、Realtime Compute for Apache Flink および Hologres を使用してリアルタイムデータウェアハウスを構築します。その後、Hologres および MaxCompute を使用して、リアルタイムおよびバッチデータの分析を実行します。
背景情報
ビジネスのデジタル化が進むにつれて、より新鮮なデータへの需要が高まっています。大規模データに対する従来のバッチ処理に加えて、多くの企業ではリアルタイムデータの処理、ストレージ、分析を行う必要があります。統合バッチおよびリアルタイム分析は、こうしたニーズに対応するために開発されました。
統合バッチおよびリアルタイム分析は、単一のプラットフォーム上でリアルタイムデータとバッチデータの両方を管理・処理します。これにより、リアルタイム処理とバッチ分析とのシームレスな接続が可能になり、データ分析の効率性および精度が向上します。主なメリットは以下のとおりです。
データ処理効率の向上：リアルタイムデータとバッチデータを単一のプラットフォーム上で統合することで、処理効率が大幅に向上し、データ転送および変換コストが削減されます。
分析精度の向上：リアルタイムデータとバッチデータを組み合わせて分析することで、結果の精度および正確性が向上します。
このアプローチにより、データ管理および処理が簡素化され、これらのタスクの効率が向上します。
データの価値を最大限に活用し、ビジネス意思決定をより適切にサポートできます。
Alibaba Cloud は、バッチおよびリアルタイムの両方のシナリオに対応した、シンプルな統合データウェアハウスソリューションを提供しています。このソリューションでは、バッチ処理に MaxCompute を、リアルタイム分析に Hologres を使用します。Realtime Compute for Apache Flink のリアルタイム処理機能と組み合わせることで、これらのサービスは Alibaba Cloud の統合データウェアハウスの中核エンジンを形成します。
ソリューションアーキテクチャ
次の図は、MaxCompute および Hologres を使用して GitHub パブリックイベントデータセットに対して統合バッチおよびリアルタイム分析を実装するための完全なパイプラインを示しています。
このアーキテクチャでは、ECS インスタンスが GitHub からリアルタイムおよびバッチのイベントデータを収集・集約し、それがデータソースとなります。このデータは、リアルタイムパイプラインおよびバッチパイプラインに供給されます。両方のパイプラインからのデータは Hologres で統合され、統一されたサービスレイヤーを提供します。
リアルタイムパイプライン：Realtime Compute for Apache Flink が Simple Log Service (SLS) からのデータをリアルタイムで処理し、Hologres に書き込みます。Flink は強力なストリーム処理エンジンです。Hologres はリアルタイムでのデータ書き込みおよび更新をサポートしており、書き込まれた直後にデータをクエリできます。ネイティブな統合により、高スループット、低遅延、モデル駆動型、高品質なリアルタイムデータウェアハウス開発が可能になります。これにより、最新のイベントの抽出やトレンドイベントの分析など、リアルタイムビジネスインサイトのニーズに対応できます。
バッチパイプライン：MaxCompute は大量のバッチデータを処理およびアーカイブします。Object Storage Service (OSS) は、さまざまなデータタイプを保存するための Alibaba Cloud サービスです。このチュートリアルで使用する生データは JSON 形式であり、OSS は便利で安全、低コストかつ信頼性の高いストレージを提供します。MaxCompute はデータ分析向けに設計されたエンタープライズグレードの SaaS クラウドデータウェアハウスです。外部テーブルを使用して OSS 内の半構造化データを直接読み取り・解析し、高付加価値データを内部ストレージに統合した後、DataWorks と連携してバッチデータウェアハウスを構築できます。
Hologres はストレージレイヤーで MaxCompute とシームレスに統合されています。これにより、Hologres を使用して MaxCompute 内の大量の既存データに対するクエリおよび分析を高速化でき、既存データに対する低頻度かつ高性能なクエリというビジネスニーズに対応できます。また、リアルタイムパイプラインで発生する可能性のあるデータ欠落などの問題を解決するために、バッチパイプラインを使用してリアルタイムデータを簡単に修正できます。
このソリューションには、以下の利点があります。
安定性と効率性に優れたバッチパイプライン：毎時データの書き込みおよび更新をサポートし、大規模データをバッチ処理して複雑な計算および分析を実行でき、コンピューティングコストを削減し、データ処理効率を向上させます。
成熟したリアルタイムパイプライン：リアルタイム取り込み、リアルタイムイベント計算、リアルタイム分析をサポートします。簡略化されたリアルタイムパイプラインにより、数秒以内に応答を提供します。
統一されたストレージおよびサービス：Hologres は、一元化されたデータストレージおよび一貫した外部インターフェイス（OLAP および Key-Value クエリの両方に単一の SQL インターフェイスを提供）を備えた統一サービスレイヤーを提供します。
統合バッチおよびリアルタイム分析：データ冗長性およびデータ移動を削減し、データ修正を可能にします。
このワンストップ開発アプローチにより、最終的に秒単位のデータ応答、エンドツーエンドのステータス可視性、コンポーネントおよび依存関係が少ない簡素化されたアーキテクチャ、および O&M コストおよび人件費の効果的な削減を実現します。
ビジネスおよびデータの理解
開発者は GitHub 上でオープンソースプロジェクトに取り組む際に、多くのイベントを作成します。GitHub は各イベントの詳細を記録します。これらの詳細には、イベントタイプ、開発者、コードリポジトリが含まれます。GitHub は、リポジトリのスター付けやコードのコミットなど、パブリックイベントを公開しています。イベントタイプの完全なリストについては、「Webhook events and payloads」をご参照ください。
GitHub は、OpenAPI を通じてパブリックイベントを提供しています。この API は、5 分の遅延があるリアルタイムデータを提供します。詳細については、「イベント」をご参照ください。
GH Archive プロジェクトは、GitHub パブリックイベントの毎時のアーカイブを収集・提供しています。これらのアーカイブを使用してオフラインデータを取得できます。詳細については、「GH Archive」をご参照ください。
GitHub ビジネスの理解
GitHub のコアビジネスは、コードおよびインタラクションの管理です。これには、Developer（開発者）、Repository（リポジトリ）、Organization（組織）の 3 つの主要エンティティが関与しています。
このデータ分析では、Event もエンティティとして保存・記録されます。
生のパブリックイベントデータの理解
次の例は、生イベントの JSON データを示しています。
{
"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"
}この分析では、15 種類の公開イベントを対象としています。発生しなかったイベントや記録されなくなったイベントは含まれていません。これらのイベントタイプの詳細については、「GitHub の公開イベントタイプ」をご参照ください。
前提条件
Elastic Compute Service (ECS) インスタンスが作成済みで、Elastic IP Address (EIP) が関連付けられています。このインスタンスは、GitHub API からリアルタイムイベントデータを抽出するために使用します。詳細については、「作成ガイド」および「Elastic IP Address」をご参照ください。
Object Storage Service (OSS) が有効化されており、GH Archive からの JSON データファイルを保存するために、ossutil ツールが ECS インスタンスにインストールされています。詳細については、「OSS の有効化」および「ossutil のインストール」をご参照ください。
MaxCompute が有効化されており、プロジェクトが作成されています。詳細については、「MaxCompute プロジェクトの作成」をご参照ください。
DataWorks が有効化されており、ワークスペースが作成されてオフラインスケジューリングタスクを作成できるようになっています。詳細については、「ワークスペースの作成」をご参照ください。
Simple Log Service (SLS) が有効化されており、プロジェクトおよび Logstore が作成されて ECS インスタンスからログとしてデータを収集できるようになっています。詳細については、「LoongCollector を使用した ECS テキストログの収集および分析」をご参照ください。
Realtime Compute for Apache Flink インスタンスが有効化されており、SLS からのログデータをリアルタイムで Hologres に書き込むことができます。詳細については、「Realtime Compute for Apache Flink の有効化」をご参照ください。
Hologres が有効化されています。詳細については、「Hologres インスタンスの購入」をご参照ください。
オフラインデータウェアハウスの構築（毎時更新）
ECS インスタンスを使用した生データファイルのダウンロードおよび OSS へのアップロード
Elastic Compute Service (ECS) インスタンスを使用して、GH Archive から JSON データファイルをダウンロードします。
wgetコマンドを使用して履歴データをダウンロードします。たとえば、wget https://data.gharchive.org/{2012..2022}-{01..12}-{01..31}-{0..23}.json.gzを実行すると、2012 年から 2022 年までの毎時のデータをダウンロードできます。毎時生成される新しいデータをダウンロードするには、次のように毎時のスケジュールタスクを設定します。
説明ECS インスタンスに ossutil がインストールされていることを確認してください。詳細については、「ossutil のインストール」をご参照ください。ossutil インストールパッケージをダウンロードして ECS インスタンスにアップロードし、
yum install unzipを実行して unzip ソフトウェアをインストールします。その後、ossutil パッケージを解凍して実行可能ファイルを/usr/bin/ディレクトリに移動します。ECS インスタンスと同じリージョンに Object Storage Service (OSS) バケットを作成していることを確認してください。バケット名は任意で指定できます。この例では、バケット名として
githubeventsを使用します。この例では、ファイルは ECS インスタンス上の
/opt/hourlydata/gh_dataディレクトリにダウンロードされます。別のディレクトリを使用することも可能です。
次のコマンドを実行して、
/opt/hourlydataディレクトリにdownload_code.shという名前のファイルを作成します。cd /opt/hourlydata vim download_code.shiキーを押して編集モードに入り、次のスクリプトを追加します。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} # データを ./gh_data/ ディレクトリにダウンロードします。別のディレクトリを使用することも可能です。 wget ${url} -P ./gh_data/ # gh_data ディレクトリに移動します。 cd gh_data # ダウンロードしたデータを JSON ファイルに解凍します。 gzip -d ${d}.json echo ${d}.json # ルートディレクトリに移動します。 cd /root # ossutil を使用してデータを OSS にアップロードします。 # githubevents OSS バケット内に hr=${h} ディレクトリを作成します。 ossutil mkdir oss://githubevents/hr=${h} # /opt/hourlydata/gh_data ディレクトリからデータを OSS にアップロードします。別のディレクトリを使用することも可能です。 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!Esc キーを押し、
:wqと入力して Enter キーを押すと、ファイルが保存されて閉じられます。次のコマンドを実行して、毎時 10 分に
download_code.shスクリプトを実行します。# 1. 次のコマンドを実行し、I キーを押して編集モードに入ります。 crontab -e # 2. 次のコマンドを追加します。その後、Esc キーを押し、:wq と入力して Enter キーを押すと終了します。 10 * * * * cd /opt/hourlydata && sh download_code.sh > download.logスクリプトが実行されると、毎時 10 分に前時間の JSON ファイルがダウンロードされます。その後、ECS インスタンス上でファイルが解凍され、
oss://githubeventsのパスに OSS へアップロードされます。前時間のファイルのみを読み取るようにするため、アップロード時に各ファイルに対して'hr=%Y-%M-%D-%H'という名前のディレクトリがパーティションとして作成されます。これにより、後続のデータ書き込み操作が最新のパーティションからのみファイルを読み取ることが保証されます。
外部テーブルを使用した OSS データの MaxCompute へのインポート
MaxCompute クライアントまたは DataWorks の ODPS SQL ノードで次のコマンドを実行します。詳細については、「クライアント (odpscmd) を使用した MaxCompute への接続」または「ODPS SQL タスクの開発」をご参照ください。
OSS に保存されている JSON ファイルを読み取るための外部テーブル
githubeventsを作成します。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/' ;MaxCompute で OSS データにアクセスするための外部テーブルの作成に関する詳細については、「OSS 内の非構造化データへのアクセス」をご参照ください。
データを保存するファクトテーブル
dwd_github_events_odpsを作成します。次のコードは、データ定義言語 (DDL) ステートメントを示しています。CREATE TABLE IF NOT EXISTS dwd_github_events_odps ( id BIGINT COMMENT 'イベント ID' ,actor_id BIGINT COMMENT 'イベント開始者の ID' ,actor_login STRING COMMENT 'イベント開始者のログイン名' ,repo_id BIGINT COMMENT 'リポジトリ ID' ,repo_name STRING COMMENT 'owner/repository_name 形式のリポジトリのフルネーム' ,org_id BIGINT COMMENT 'リポジトリが属する組織の ID' ,org_login STRING COMMENT 'リポジトリが属する組織の名前' ,`type` STRING COMMENT 'イベントタイプ' ,created_at DATETIME COMMENT 'イベントが発生した時刻' ,action STRING COMMENT 'イベントアクション' ,iss_or_pr_id BIGINT COMMENT 'issue または pull request の ID' ,number BIGINT COMMENT 'issue または pull request の番号' ,comment_id BIGINT COMMENT 'コメント ID' ,commit_id STRING COMMENT 'コミット ID' ,member_id BIGINT COMMENT 'メンバー ID' ,rev_or_push_or_rel_id BIGINT COMMENT 'レビュー、プッシュ、またはリリースの ID' ,ref STRING COMMENT '作成または削除されたリソースの名前' ,ref_type STRING COMMENT '作成または削除されたリソースのタイプ' ,state STRING COMMENT 'issue、pull request、または pull request レビューのステータス' ,author_association STRING COMMENT 'アクターとリポジトリの関係' ,language STRING COMMENT 'pull request 内のコードの言語' ,merged BOOLEAN COMMENT 'pull request がマージされたかどうかを示します' ,merged_at DATETIME COMMENT 'コードがマージされた時刻' ,additions BIGINT COMMENT '追加されたコード行数' ,deletions BIGINT COMMENT '削除されたコード行数' ,changed_files BIGINT COMMENT 'pull request で変更されたファイル数' ,push_size BIGINT COMMENT 'コミット数' ,push_distinct_size BIGINT COMMENT '個別のコミット数' ,hr STRING COMMENT 'イベントが発生した時刻（時）。たとえば、イベントが 00:23 に発生した場合、hr の値は 00 です。' ,`month` STRING COMMENT 'イベントが発生した月。たとえば、イベントが 2015 年 10 月に発生した場合、month の値は 2015-10 です。' ,`year` STRING COMMENT 'イベントが発生した年。たとえば、イベントが 2015 年に発生した場合、year の値は 2015 です。' ) PARTITIONED BY ( ds STRING COMMENT 'イベントが発生した日付（yyyy-mm-dd 形式）。' );JSON データを解析してファクトテーブルに書き込みます。
次のコマンドを実行してパーティションを追加し、JSON データを解析して
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);データをクエリします。
次のコマンドを実行して
dwd_github_events_odpsテーブルからデータをクエリします。SET odps.sql.allow.fullscan=true; SELECT * FROM dwd_github_events_odps where ds = '2023-03-31' limit 10;次のサンプル結果が返されます。
リアルタイムデータウェアハウスの構築
ECS を使用したリアルタイムデータの取得
Elastic Compute Service (ECS) インスタンスは、GitHub API からリアルタイムイベントデータを抽出するために使用されます。このトピックでは、GitHub API からリアルタイムデータを収集する方法を示すサンプルスクリプトを提供します。
スクリプトが実行されるたびに、1 分間実行されます。この期間中に API が提供するリアルタイムイベントデータを収集し、各イベントを JSON 形式で保存します。
このスクリプトは、すべてのリアルタイムイベントデータを確実に収集することを保証するものではありません。
GitHub API から継続的にデータを収集するには、Accept ヘッダーおよび Authorization ヘッダーを指定する必要があります。Accept の値は固定です。Authorization には、GitHub から取得した個人アクセストークンを入力します。個人アクセストークンの作成方法の詳細については、「このドキュメント」をご参照ください。
次のコマンドを実行して、
/opt/realtimeディレクトリにdownload_realtime_data.pyという名前のファイルを作成します。cd /opt/realtime vim download_realtime_data.pyiキーを押して編集モードに入り、ファイルに次のサンプル内容を追加します。#!python import requests import json import sys import time # API URL を取得します 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 # API から 1 ページ分のデータを収集します def download(link, fname): # GitHub API の Accept ヘッダーおよび Authorization ヘッダーを定義します 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 # 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 # 現在時刻を定義します def get_current_ms(): return round(time.time()*1000) # スクリプトの実行時間を 1 分に定義します def main(fname): current_ms = get_current_ms() while get_current_ms() - current_ms < 60*1000: download_all_data(fname) time.sleep(0.1) # スクリプトを実行します if __name__ == '__main__': if len(sys.argv) < 2: print('usage: python {} <log_file>'.format(sys.argv[0])) exit(0) main(sys.argv[1])Esc キーを押し、
:wqと入力して Enter キーを押すと、ファイルが保存されて閉じられます。download_realtime_data.pyを実行して、各実行から収集されたデータを個別に保存するためのrun_py.shファイルを作成します。内容は次のとおりです。python /opt/realtime/download_realtime_data.py /opt/realtime/gh_realtime_data/$(date '+%Y-%m-%d-%H:%M:%S').json履歴データを削除するための
delete_log.shファイルを作成します。内容は次のとおりです。d=$(TZ=UTC date --date='2 day ago' '+%Y-%m-%d') rm -f /opt/realtime/gh_realtime_data/*${d}*.json次のコマンドを実行して、GitHub データを毎分収集し、履歴データを毎日削除します。
#1. 次のコマンドを実行し、I キーを押して編集モードに入ります。 crontab -e #2. 次のコマンドを追加します。その後、Esc キーを押し、:wq と入力して Enter キーを押すと終了します。 * * * * * bash /opt/realtime/run_py.sh 1 1 * * * bash /opt/realtime/delete_log.sh
SLS を使用した ECS データの収集
Simple Log Service (SLS) は、ECS インスタンスから抽出されたリアルタイムイベントデータをログとして収集するために使用されます。
SLS は、Logtail を使用して ECS インスタンスからログを収集することをサポートしています。このトピックのデータは JSON 形式であるため、Logtail の JSON モードを使用して、ECS インスタンスから増分 JSON ログを迅速に収集できます。詳細については、「JSON モードでのログ収集」をご参照ください。このトピックでは、SLS を構成して生データのトップレベルのキーと値のペアを解析します。
この例では、Logtail 構成のログパスパラメーターを /opt/realtime/gh_realtime_data/**/*.json に設定します。
構成が完了すると、SLS は ECS インスタンスから増分イベントデータを継続的に収集します。次の図は、収集されたデータの例を示しています。
Flink を使用した SLS データの Hologres へのリアルタイム書き込み
Flink は、SLS によって収集されたログデータをリアルタイムで Hologres に書き込むために使用されます。Flink の SLS ソーステーブルおよび Hologres 結果テーブルを使用することで、SLS から Hologres にデータをリアルタイムで書き込むことができます。詳細については、「Simple Log Service からのデータインポート」をご参照ください。
Hologres 内部テーブルを作成します。
このトピックで作成される内部テーブルは、生 JSON データの一部のキーと値のペアのみを保持します。イベント
idおよび日付dsはプライマリキーとして設定されます。イベントidは分散キーとして設定されます。日付dsはパーティションキーとして設定されます。イベント時間created_atは event_time_column として設定されます。必要に応じて他のフィールドにインデックスを作成して、クエリ効率を向上させることができます。インデックスの詳細については、「CREATE TABLE」をご参照ください。この例でテーブルを作成するために使用されるデータ定義言語 (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'; COMMENT ON COLUMN public.gh_realtime_data.actor_id IS 'イベント開始者の ID'; COMMENT ON COLUMN public.gh_realtime_data.actor_login IS 'イベント開始者のログイン名'; COMMENT ON COLUMN public.gh_realtime_data.repo_id IS 'リポジトリ ID'; COMMENT ON COLUMN public.gh_realtime_data.repo_name IS 'リポジトリ名'; COMMENT ON COLUMN public.gh_realtime_data.org_id IS 'リポジトリが属する組織の ID'; COMMENT ON COLUMN public.gh_realtime_data.org_login IS 'リポジトリが属する組織の名前'; COMMENT ON COLUMN public.gh_realtime_data.type IS 'イベントタイプ'; COMMENT ON COLUMN public.gh_realtime_data.created_at IS 'イベントが発生した時刻'; COMMENT ON COLUMN public.gh_realtime_data.action IS 'イベントアクション'; COMMENT ON COLUMN public.gh_realtime_data.iss_or_pr_id IS 'issue/pull_request ID'; COMMENT ON COLUMN public.gh_realtime_data.number IS 'issue/pull_request 番号'; COMMENT ON COLUMN public.gh_realtime_data.comment_id IS 'コメント ID'; COMMENT ON COLUMN public.gh_realtime_data.commit_id IS 'コミット ID'; COMMENT ON COLUMN public.gh_realtime_data.member_id IS 'メンバー ID'; COMMENT ON COLUMN public.gh_realtime_data.rev_or_push_or_rel_id IS 'レビュー/プッシュ/リリース ID'; COMMENT ON COLUMN public.gh_realtime_data.ref IS '作成または削除されたリソースの名前'; COMMENT ON COLUMN public.gh_realtime_data.ref_type IS '作成または削除されたリソースのタイプ'; COMMENT ON COLUMN public.gh_realtime_data.state IS 'issue/pull_request/pull_request_review のステータス'; COMMENT ON COLUMN public.gh_realtime_data.author_association IS 'アクターとリポジトリの関係'; COMMENT ON COLUMN public.gh_realtime_data.language IS 'プログラミング言語'; COMMENT ON COLUMN public.gh_realtime_data.merged IS 'マージが承認されたかどうかを指定します'; COMMENT ON COLUMN public.gh_realtime_data.merged_at IS 'コードがマージされた時刻'; COMMENT ON COLUMN public.gh_realtime_data.additions IS '追加されたコード行数'; COMMENT ON COLUMN public.gh_realtime_data.deletions IS '削除されたコード行数'; COMMENT ON COLUMN public.gh_realtime_data.changed_files IS 'pull request で変更されたファイル数'; COMMENT ON COLUMN public.gh_realtime_data.push_size IS 'プッシュ数'; COMMENT ON COLUMN public.gh_realtime_data.push_distinct_size IS '個別のプッシュ数'; COMMENT ON COLUMN public.gh_realtime_data.hr IS 'イベントが発生した時刻（時）。たとえば、時刻が 00:23 の場合、hr=00 です。'; COMMENT ON COLUMN public.gh_realtime_data.month IS 'イベントが発生した月。たとえば、日付が 2015 年 10 月の場合、month=2015-10 です。'; COMMENT ON COLUMN public.gh_realtime_data.year IS 'イベントが発生した年。たとえば、年が 2015 年の場合、year=2015 です。'; COMMENT ON COLUMN public.gh_realtime_data.ds IS 'イベントが発生した日。ds=yyyy-mm-dd。'; COMMIT;Flink を使用してリアルタイムでデータを書き込みます。
Flink を使用して SLS データをさらに解析し、リアルタイムで Hologres に書き込みます。Flink で次のステートメントを使用して、書き込むデータをフィルターします。これにより、イベント ID またはイベント時間 (
created_at) が null のダーティデータは破棄され、最近のイベントデータのみが保持されます。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>',--SLS のプライベートエンドポイント 'accessid' = '<accesskey id>',--ご利用のアカウントの AccessKey ID 'accesskey' = '<accesskey secret>',--ご利用のアカウントの AccessKey Secret 'project' = '<project name>',--SLS プロジェクトの名前 'logstore' = '<logstore name>'--SLS Logstore の名前 'starttime' = '2023-04-06 00:00:00',--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>', --Hologres データベースの名前 'tablename' = '<hologres tablename>', --データを受信する Hologres テーブルの名前 'username' = '<accesskey id>', --現在の Alibaba Cloud アカウントの AccessKey ID 'password' = '<accesskey secret>', --現在の Alibaba Cloud アカウントの AccessKey Secret 'endpoint' = '<endpoint>', --現在の Hologres インスタンスの VPC エンドポイント 'jdbcretrycount' = '1', --接続失敗時のリトライ回数 'partitionrouter' = 'true', --パーティションテーブルへの書き込みを指定します 'createparttable' = 'true', --パーティションを自動作成するかどうかを指定します 'mutatetype' = 'insertorignore' --データ書き込みモード ); 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);パラメーターの詳細については、「Simple Log Service (SLS)」および「Hologres」をご参照ください。
説明GitHub からの生イベントデータは UTC タイムゾーンを使用しており、タイムゾーン属性を持っていません。Hologres のデフォルトのタイムゾーンは UTC + 08:00 です。そのため、Flink から Hologres にリアルタイムでデータを書き込む際にタイムゾーンを調整する必要があります。Flink SQL のソーステーブルデータに UTC タイムゾーン属性を割り当てます。手順は次のとおりです。
ステップ 1：ジョブ編集ページに移動します
Realtime Compute for Apache Flink コンソール にログインします
対象の ワークスペース に移動します
ご利用の Flink SQL または JAR ジョブを見つけ、編集 をクリックします
ステップ 2：デプロイメント詳細タブを開きます
⚠️ 重要な変更：「Flink 構成」は個別のセクションではなくなりました。「デプロイメント詳細」に統合されています。
ジョブ編集ページの上部で、デプロイメント詳細 タブに切り替えます。
パラメーター構成 セクションまで下にスクロールします。
ステップ 3：カスタム構成を追加します
パラメーター構成 の右側にある 編集 ボタンをクリックします。
表示されるダイアログボックスで、その他の構成 テキストボックスを見つけます。
テキストボックスに、Flink パラメーター
table.local-time-zone:Asia/Shanghaiを キーと値のペア として追加して、Flink システムタイムゾーンをAsia/Shanghaiに設定します。
データをクエリします。
Flink を介して Hologres に書き込まれた SLS データをクエリします。その後、必要に応じてデータ開発を実行できます。
SELECT * FROM public.gh_realtime_data limit 10;次の結果はその例です。
オフラインデータを使用したリアルタイムデータの修正
このトピックで説明するシナリオでは、リアルタイムデータが欠落している可能性があります。オフラインデータを使用してリアルタイムデータを修正できます。次の手順は、前日のリアルタイムデータを修正する方法を示しています。データ修正期間は必要に応じて調整できます。
Hologres に外部テーブルを作成して、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');パラメーターの詳細については、「IMPORT FOREIGN SCHEMA」をご参照ください。
オフラインデータを使用して前日のリアルタイムデータを修正するための一時テーブルを作成します。
説明-- 潜在的な一時テーブルをクリーンアップします DROP TABLE IF EXISTS gh_realtime_data_tmp; -- 一時テーブルを作成します SET hg_experimental_enable_create_table_like_properties = ON; CALL HG_CREATE_TABLE_LIKE ('gh_realtime_data_tmp', 'select * from gh_realtime_data'); -- (オプション) サーバーレスコンピューティングを使用して、大規模なオフラインデータインポートおよび ETL ジョブを実行します。 SET hg_computing_resource = 'serverless'; -- 一時テーブルにデータを挿入し、統計情報を更新します 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; -- 設定をリセットして、不要な SQL ステートメントがサーバーレスリソースを使用しないようにします。 RESET hg_computing_resource; -- 原子テーブルを既存の一時子テーブルに置き換えます 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;
データ分析
収集した大量のデータに対して、幅広い分析を実行できます。ビジネスで必要な時間範囲に基づいて、レイヤー化されたデータウェアハウスを設計します。これにより、リアルタイム分析、オフライン分析、リアルタイムおよびオフラインの統合分析など、さまざまなニーズに対応できます。
次の例では、前述のリアルタイムデータを分析します。特定のコードリポジトリまたは開発者を対象に分析することもできます。
今日のパブリックイベントの総数をクエリします。
SELECT count(*) FROM gh_realtime_data WHERE created_at >= date_trunc('day', now());次のサンプル結果が得られます。
count ------ 1006過去 1 日間に最もアクティブなプロジェクト（イベント数が最も多いプロジェクト）をクエリします。
SELECT repo_name, COUNT(*) AS events FROM gh_realtime_data WHERE created_at >= now() - interval '1 day' GROUP BY repo_name ORDER BY events DESC LIMIT 5;次のサンプル結果が得られます。
repo_name events ----------------------------------------+------ leo424y/heysiri.ml 29 arm-on/plan 10 Christoffel-T/fiverr-pat-20230331 9 mate-academy/react_dynamic-list-of-goods 9 openvinotoolkit/openvino 7過去 1 日間に最もアクティブな開発者（イベント数が最も多い開発者）をクエリします。
SELECT actor_login, COUNT(*) AS events FROM gh_realtime_data WHERE created_at >= now() - interval '1 day' AND actor_login NOT LIKE '%[bot]' GROUP BY actor_login ORDER BY events DESC LIMIT 5;次のサンプル結果が得られます。
actor_login events ------------------+------ direwolf-github 13 arm-on 10 sergii-nosachenko 9 Christoffel-T 9 yangwang201911 7過去 1 時間に最も人気のあるプログラミング言語のランキングをクエリします。
SELECT language, count(*) total FROM gh_realtime_data WHERE created_at > now() - interval '1 hour' AND language IS NOT NULL GROUP BY language ORDER BY total DESC LIMIT 10;次のサンプル結果が得られます。
language total -----------+---- JavaScript 25 C++ 15 Python 14 TypeScript 13 Java 8 PHP 8過去 1 日間にスターを受け取った数でプロジェクトのランキングをクエリします。
説明この例では、ユーザーがプロジェクトのスターを外すケースは考慮していません。
SELECT repo_id, repo_name, COUNT(actor_login) total FROM gh_realtime_data WHERE type = 'WatchEvent' AND created_at > now() - interval '1 day' GROUP BY repo_id, repo_name ORDER BY total DESC LIMIT 10;次のサンプル結果が得られます。
repo_id repo_name total ---------+----------------------------------+----- 618058471 facebookresearch/segment-anything 4 619959033 nomic-ai/gpt4all 1 97249406 denysdovhan/wtfjs 1 9791525 digininja/DVWA 1 168118422 aylei/interview 1 343520006 joehillen/sysz 1 162279822 agalwood/Motrix 1 577723410 huggingface/swift-coreml-diffusers 1 609539715 e2b-dev/e2b 1 254839429 maniackk/KKCallStack 1今日のデイリーアクティブユーザーおよびプロジェクト数をクエリします。
SELECT uniq (actor_id) actor_num, uniq (repo_id) repo_num FROM gh_realtime_data WHERE created_at > date_trunc('day', now());次のサンプル結果が得られます。
actor_num repo_num ---------+-------- 743 816