Analyze full and incremental GitHub data with near real-time synchronization

Step

Procedure

Expected result

Step 0

Register for the invitational preview of new MaxCompute features

1. Open the invitational preview application form for new MaxCompute features.

2. Select Delta Table incremental table format, Incremental computing based on incremental materialized views, and MCQA 2.0 SQL engine query acceleration.

3. Click Register.

The new MaxCompute features are enabled.

Step 1

Initialize a new MaxCompute project

1. Open the MaxCompute console and click Create Project.

2. In the Create Project dialog box, configure the following parameters.

   Set Project Name to delta_compute_yunqi. Set Project Type to Apsara Conference New Feature Invitational Preview Project. Configure other parameters as needed or keep the default settings.

   You can select the China (Beijing), China (Shanghai), or China (Hangzhou) region.

3. Click OK.

1. In the navigation pane on the left of the console, choose Workspace > SQL Analysis to open the Ad-Hoc Query page.

2. On the left side of the SQL Analysis page, find the new project delta_compute_yunqi.

Step 2

Purchase a new quota in Quota Management

1. In the navigation pane on the left of the MaxCompute console, choose Workspace > Quota Management and click Purchase Quota.

2. Create a subscription level-1 quota and complete the payment as prompted on the page.

3. Under the created level-1 quota group, create two different level-2 quota groups:

   • Batch processing quota group: batch_demo_yunqi

   • Interactive quota group: mcqa2_demo_yunqi

4. Set the resources for the batch processing quota group and the interactive quota group.

On the Quota Management page, confirm that the batch processing quota group and the interactive quota group are created.

Step 3

Create a MaxCompute Delta Table

On the SQL Analysis page, run the following sample commands to create two MaxCompute internal tables:

• Create the yunqi_github_events_odps_cdc table. Data will be written to this table in real-time using Flink CDC.

  -- Enable the upsertable feature for ODPS SQL.
  SET odps.sql.upsertable.table.enable=true;
  SET odps.storage.orc.enable.memcmp.sort.key=true; 
  
  -- Flink CDC
  CREATE TABLE IF NOT EXISTS yunqi_github_events_odps_cdc
  (
      id                     BIGINT NOT NULL COMMENT 'Event ID'
      ,actor_id              BIGINT COMMENT 'Event initiator ID'
      ,actor_login           STRING COMMENT 'Event initiator logon name'
      ,repo_id               BIGINT COMMENT 'Repo ID'
      ,repo_name             STRING COMMENT 'Full repo name: owner/Repository_name'
      ,org_id                BIGINT COMMENT 'ID of the organization that owns the repo'
      ,org_login             STRING COMMENT 'Name of the organization that owns the repo'
      ,`type`                STRING COMMENT 'Event type'
      ,created_at            TIMESTAMP NOT NULL COMMENT 'Time of event occurrence'
      ,action                STRING COMMENT 'Event action'
      ,commit_id             STRING COMMENT 'Commit record ID'
      ,member_id             BIGINT COMMENT 'Member ID'
      ,language              STRING COMMENT 'Programming language'
      ,PRIMARY KEY(id, created_at))
  STORED AS ALIORC  
  TBLPROPERTIES ('acid.data.retain.hours'='24',
       'acid.incremental.query.out.of.time.range.enabled'='true',
       'columnar.nested.type'='true',
       'transactional'='true',
       'write.bucket.num'='64',
       'acid.ingest.commit.num.check.limit'='10',
       'cdc.insert.into.passthrough.enable'='true', 
       'acid.cdc.mode.enable' = 'true',-- If asynchronous mode is enabled, you must make additional settings.
       'acid.cdc.build.async'='true', 
       'acid.cdc.build.interval'='60') 
  LIFECYCLE 36500;
  

• Create the yunqi_github_events_odps_dw table. Data will be written to this table in real-time using DataWorks data integration.

  -- Enable the upsertable feature for ODPS SQL.
  SET odps.sql.upsertable.table.enable=true;
  SET odps.storage.orc.enable.memcmp.sort.key=true; 
  
  -- DataWorks real-time data synchronization
  CREATE TABLE IF NOT EXISTS yunqi_github_events_odps_dw
  (
      id                     BIGINT NOT NULL COMMENT 'Event ID'
      ,actor_id              BIGINT COMMENT 'Event initiator ID'
      ,actor_login           STRING COMMENT 'Event initiator logon name'
      ,repo_id               BIGINT COMMENT 'Repo ID'
      ,repo_name             STRING COMMENT 'Full repo name: owner/Repository_name'
      ,org_id                BIGINT COMMENT 'ID of the organization that owns the repo'
      ,org_login             STRING COMMENT 'Name of the organization that owns the repo'
      ,`type`                STRING COMMENT 'Event type'
      ,created_at            TIMESTAMP COMMENT 'Time of event occurrence'
      ,action                STRING COMMENT 'Event action'
      ,commit_id             STRING COMMENT 'Commit record ID'
      ,member_id             BIGINT COMMENT 'Member ID'
      ,language              STRING COMMENT 'Programming language'
      ,PRIMARY KEY(id))
  STORED AS ALIORC  
  TBLPROPERTIES ('acid.data.retain.hours'='24',
       'acid.incremental.query.out.of.time.range.enabled'='true',
       'columnar.nested.type'='true',
       'transactional'='true',
       'write.bucket.num'='64',
       'acid.ingest.commit.num.check.limit'='10',
       'cdc.insert.into.passthrough.enable'='true', 
       'acid.cdc.mode.enable' = 'true',-- CDC is synchronized by default.
       'acid.cdc.build.async'='true', -- If asynchronous mode is enabled, you must make additional settings.
       'acid.cdc.build.interval'='60') 
  LIFECYCLE 36500;

On the SQL Analysis page, select the project delta_compute_yunqi and confirm that the following two Delta Tables exist:

• yunqi_github_events_odps_cdc

• yunqi_github_events_odps_dw

Step

Procedure

Expected result

Step 1

Use the interactive resource group for near real-time data analysis

1. In the navigation pane on the left of the MaxCompute console, choose Workspace > SQL Analysis to open the Ad-Hoc Query page.

2. Select the interactive quota group created in Step 1 to perform near real-time query analysis.

The specific SQL analysis steps are as follows:

• Q1: What are the most active projects in the last 24 hours?

  --Q1: Top 5 most active projects in the last 24 hours
  SELECT
      repo_name,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      repo_name
  ORDER BY
      events DESC
  LIMIT 5;

• Q2: Who are the most active developers in the past 24 hours?

  --Q2: Top 5 most active developers in the last 24 hours
  SELECT
      actor_login,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      actor_login
  ORDER BY
      events 
  LIMIT 5;

• Q3: Total number of public events today

  --Q3: Total number of public events today
  SELECT count(*) FROM yunqi_github_events_odps_dw WHERE TO_DATE(created_at) >= date_add(now(),-1);

• Q4: Display real-time events

  --Q4: Display real-time events
  SELECT
      cast(created_at as STRING ),
      actor_login,
      type,
      repo_name
  FROM
      yunqi_github_events_odps_dw
  ORDER BY
      created_at DESC
  LIMIT 5;

Observe the SQL query results.

Step 2

Use the batch processing resource group for incremental computing and aggregate queries with incremental MVs

1. In the navigation pane on the left of the MaxCompute console, choose Workspace > SQL Analysis to open the Ad-Hoc Query page.

2. Select the batch processing quota group created in Step 1 for batch MV applications.

The specific SQL analysis steps are as follows:

1. Confirm that the Delta Table is created and that historical and incremental data is being updated.

   Create CDC information to track the Delta Table (Optional: CDC is automatically generated by default).

   --Q1: Create CDC information to track the Delta Table.
   ALTER TABLE yunqi_github_events_odps_dw build cdc; --Trigger the subsequent CDC process.

2. Perform real-time incremental MV queries. Based on historical and incremental data, create a lightweight aggregation incremental MV that aggregates and counts the number of events by repo_name, language, and create_date.

   --Q2: Generate incremental materialized view MV1-yunqi_incre_mv1.
   SET useTunnel=FALSE;
   CREATE materialized VIEW IF NOT EXISTS yunqi_incre_mv1
   refresh every 2 minutes
   tblproperties("enable_auto_refresh"="true", "refresh_mode"="incremental")
   AS SELECT
       repo_name,
       language,
       COUNT(*) AS events,
       to_date(created_at) AS create_date
   FROM
       yunqi_github_events_odps_dw
   WHERE
       date_add(created_at,-1)>'2024-07-26' AND language IS NOT NULL
   GROUP BY
     repo_name,language,to_date(created_at);

   --Q3: Observe the source table status - Verify using SHOW HISTORY to view table behavior / Confirm CDC Build status.
   SHOW history FOR TABLE yunqi_github_events_odps_dw;
   SELECT * FROM table_changes('yunqi_github_events_odps_dw', 1);

3. SQL analysis: Query the total number of events and the language for each repo within a specific time frame, based on repo_name and language.

   --Q4: Query the total number of events and the language, sorted by repo_name and language in descending order.
   SELECT repo_name, language, SUM(events) AS events_sum, COUNT(create_date) AS day_count 
   FROM yunqi_incre_mv1 
   WHERE events > 100 
   GROUP BY repo_name,language ORDER BY events_sum DESC;

   --Q5: View the historical operation changes of incremental materialized view MV1.
   SHOW history FOR TABLE yunqi_incre_mv1;
   -- Data is updated in 5-minute batches. If there are no changes, run the following command.
   ALTER materialized VIEW yunqi_incre_mv1 rebuild;
   SELECT * FROM table_changes('yunqi_github_events_odps_dw', 1);

4. Build a cascading incremental materialized view MV2. Based on the existing incremental materialized view MV1, create a heavyweight aggregation layer.

   • Generate incremental materialized view MV2-yunqi_incre_mv2. Based on incremental MV1, materialize the SQL analysis results into incremental MV2. This is based on the query from Q4.

     --Q6: Generate cascading incremental materialized view MV2-yunqi_incre_mv2.
     SET useTunnel=FALSE;
     CREATE materialized VIEW IF NOT EXISTS yunqi_incre_mv2
     refresh every 2 minutes
     tblproperties("enable_auto_refresh"="true", "refresh_mode"="incremental")
     AS 
     SELECT repo_name, language, SUM(events) AS events_sum, COUNT(create_date) AS day_count 
     FROM yunqi_incre_mv1 
     WHERE events > 100 
     GROUP BY repo_name,language;

   • SQL analysis: Within a specific time frame, group by programming language and sort by the number of repos to query the total number of events for each language.

     --Q7: Sort popular programming languages.
     SELECT language, COUNT(repo_name) AS repo_cnt,MAX(events_sum) AS max_events FROM yunqi_incre_mv2 
     GROUP BY language ORDER BY repo_cnt DESC;

     --Q8: Observe the historical operation status of incremental MV2 / Table change information.
     SHOW history FOR TABLE yunqi_incre_mv2;
     -- Data is updated in 5-minute batches. If there are no changes, run the following command.
     ALTER materialized VIEW yunqi_incre_mv2 rebuild;
     SELECT * FROM table_changes('yunqi_incre_mv2', 2);

Observe the SQL query results.

Step 3

Use the batch processing resource group for incremental computing with Stream and Task applications

1. In the navigation pane on the left of the MaxCompute console, choose Workspace > SQL Analysis to open the Ad-Hoc Query page.

2. Select the batch processing quota group created in Step 1 for incremental CDC.

The specific SQL analysis steps are as follows:

1. Create a stream object to query the newly added CDC records for subsequent incremental computing.

   --Q1: Query the latest version information of the table data.
   SELECT get_latest_version('yunqi_github_events_odps_dw'); 

   Create a Stream object. Use the information from get_latest_version to set the starting version for tracking incremental data. Select a historical version that is less than or equal to the latest version to start tracking incremental CDC records.

   --Q2: Create a Stream object.
   CREATE stream stream_githubs ON TABLE yunqi_github_events_odps_dw version AS OF 10 strmproperties('read_mode'='cdc') ;
   
   DESC stream stream_githubs;

2. Create and run a Task. The task automatically performs incremental updates for repo_name.

   • Create the aggregation table yunqi_github_events_ads.

     --Create the aggregation table yunqi_github_events_ads.
     SET odps.sql.upsertable.table.enable=true;
     CREATE TABLE IF NOT EXISTS 
     yunqi_github_events_ads
     (
         repo_name             STRING COMMENT 'Full repo name: owner/Repository_name'
         ,language              STRING COMMENT 'Programming language'
         ,events_cnt            BIGINT COMMENT 'Number of events'
         ,create_date           DATE COMMENT 'Time of event occurrence'
     )
     STORED AS ALIORC  
     TBLPROPERTIES (
          'transactional'='true',
          'write.bucket.num'='64') 
     LIFECYCLE 36500;

   • The task automatically performs incremental updates for repo_name.

     --The task automatically performs incremental updates for repo_name.
     SET odps.sql.periodic.task.enabled=true;
     CREATE task yunqi_incre_task1
     schedule='45 second' 
     taskproperties('schedule_strategy'='200')
     WHEN stream_has_data('stream_githubs')
     AS INSERT overwrite yunqi_github_events_ads SELECT
         repo_name,
         language,
         COUNT(*) AS events_cnt,
         to_date(created_at) AS create_date
     FROM
         yunqi_github_events_odps_dw
     WHERE
         date_add(created_at,-1)>'2024-08-26' AND language IS NOT NULL
     GROUP BY
     repo_name,language,to_date(created_at);
     

     Every minute, the task automatically queries the incremental CDC data from the DWD layer, analyzes it, and writes the results to the ADS table yunqi_github_events_ads. Although this custom pipeline is complex to write, it lets you flexibly define the incremental computing and analysis logic using SQL. (Duration setting: You can set the schedule in seconds for intervals under 1 minute. For intervals over 1 minute, set it in minutes, up to a maximum of 59 minutes.)

3. Update the Task status. The task automatically performs incremental updates for repo_name.

   --Update the Task status. The task automatically performs incremental updates for repo_name.
   ALTER task  yunqi_incre_task1  suspend;  --Pause
   ALTER task  yunqi_incre_task1  resume; --Resume

4. Query the CDC build task status.

   --Query the CDC build task status.
   SHOW history FOR task yunqi_incre_task1;
   SELECT *  FROM stream_githubs

Observe the SQL query results.

Step

Procedure

Expected result

Step 1

Scale out the quota

1. In the navigation pane on the left of the MaxCompute console, choose Workspace > Quota Management, and click Quota Configurations to the right of the target level-1 quota group.

2. On the Basic Configurations page, click Edit Basic Configurations.

3. Modify the interactive quota group mcqa2_demo_yunqi by scaling it out from 64 CUs to 128 CUs.

The target quota group is scaled out.

Step 2

Compare and analyze query performance and optimization

1. In the navigation pane on the left of the MaxCompute console, choose Workspace > SQL Analysis to open the Ad-Hoc Query page.

2. Select the interactive quota group created in Step 1 to perform near real-time query analysis and observe data changes.

The specific SQL analysis steps are as follows:

• Q1 (Optional): Perform a major compaction operation to optimize queries.

  --Q1: Perform a major compaction operation to optimize queries.
  SET odps.merge.task.mode=service;
  ALTER TABLE yunqi_github_events_odps_dw compact major;

  Compare MCQA 2.0 query performance using the interactive resource group.

  --Q2: Near real-time query - Display real-time events.
  SELECT
      cast(created_at AS STRING ),
      actor_login,
      type,
      repo_name
  FROM
      yunqi_github_events_odps_dw
  ORDER BY
      created_at DESC
  LIMIT 5;

• Incremental MV1 query: Query the total number of events and the language for each repo within a specific time frame, based on repo_name and language.

  --Q3: Query the total number of events and the language, sorted by repo_name and language in descending order.
  SELECT repo_name, language, SUM(events) AS events_sum, COUNT(create_date) AS day_count 
  FROM yunqi_incre_mv1 
  WHERE events > 100 
  GROUP BY repo_name,language ORDER BY events_sum DESC;

• Incremental MV2 query: Within a specific time frame, group by programming language to query the number of repos and the total number of events for each language.

  --Q4: Sort popular programming languages.
  SELECT language, COUNT(repo_name) AS repo_cnt,MAX(events_sum) AS max_events FROM yunqi_incre_mv2 
  GROUP BY language;

Observe the SQL query results.

Step

Procedure

Expected result

Step 1

Configure MaxCompute compute resources in the DataWorks IDE Data Development module

1. Go to the DataWorks IDE Data Development page. In the navigation pane on the left, click the icon to open the Data Catalog page.

2. In the upper-right corner, click Workspace Management, select Compute Resources, and click Bind Compute Resource.

3. On the right side of the page, configure Compute Resources, Compute Quota, and Resource Group.

4. Bind the MaxCompute project and run the MaxCompute SQL task.

None

Step 2

Perform data analysis on the DataWorks IDE Data Development platform

The specific SQL analysis steps are as follows:

• Q1: What are the most active projects in the past 24 hours?

  SELECT
      repo_name,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      repo_name
  ORDER BY
      events DESC
  LIMIT 5;

• Q2: Who are the most active developers in the past 24 hours?

  SELECT
      actor_login,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      actor_login
  ORDER BY
      events 
  LIMIT 5;

• Q3: Total number of public events today

  SELECT count(*) FROM yunqi_github_events_odps_dw WHERE TO_DATE(created_at) >= date_add(now(),-1);

• Q4: Display real-time events

  SELECT
      cast(created_at AS STRING ),
      actor_login,
      type,
      repo_name
  FROM
      yunqi_github_events_odps_dw
  ORDER BY
      created_at DESC
  LIMIT 5;

Observe the SQL query results.