MaxCompute：使用阿里雲Flink（流式資料轉送）

該功能的使用限制如下：

• 本外掛程式僅支援Blink 3.2.1及以上版本。

• MaxCompute中的聚簇表不支援作為MaxCompute結果表。

命令樣本如下：

create table odps_output(
    id INT,
    user_name VARCHAR,
    content VARCHAR
) with (
    type ='custom',
    class = 'com.alibaba.blink.customersink.MaxComputeStreamTunnelSink',
    endpoint = '<YourEndPoint>',
    project = '<YourProjectName>',
    `table` = '<YourtableName>',
    access_id = '<yourAccessKeyId>',
    access_key = '<yourAccessKeySecret>',
    `partition` = 'ds=2018****'
);

包含MaxCompute結果表的Realtime ComputeFlink版作業程式碼範例如下：

• 寫入固定分區

  create table source (
     id INT,
     len INT,
     content VARCHAR
  ) with (
     type = 'random'
  );
  create table odps_sink (
     id INT,
     len INT,
     content VARCHAR
  ) with (
     type='custom',
     class = 'com.alibaba.blink.customersink.MaxComputeStreamTunnelSink',
     endpoint = '<yourEndpoint>', 
     project = '<yourProjectName>',
     `table` = '<yourTableName>',
     accessId = '<yourAccessId>',
     accessKey = '<yourAccessPassword>',
     `partition` = 'ds=20180418'
  );
  insert into odps_sink 
  select 
     id, len, content 
  from source;

• 寫入動態分區

  create table source (
     id INT,
     len INT,
     content VARCHAR,
     c TIMESTAMP 
  ) with (
     type = 'random'
  );
  create table odps_sink (
     id INT,
     len INT,
     content VARCHAR,
     ds VARCHAR                        --動態分區列需要顯式寫在建表語句中。
  ) with (
     type = 'odps',
     endpoint = '<yourEndpoint>', 
     project = '<yourProjectName>',
     `table` = '<yourTableName>',
     accessId = '<yourAccessId>',
     accessKey = '<yourAccessPassword>',
     `partition`='ds'                 --不寫分區的值，表示根據ds欄位的值寫入不同分區。
     ,enable_dynamic_partition = 'true' --啟用動態分區。
     ,dynamic_partition_limit='50' --最大並發分區數50。
     ,flush_batch_size = '524288' --緩衝區512 KB。
     ,flush_interval_ms = '60000' --Flush間隔60秒。
     ,flush_retry_count = '5' --Flush失敗重試5次。
     ,flush_retry_interval_sec = '2' --失敗稍候再試單位2秒。
     ,flush_retry_strategy = 'linear' --連續失敗重試時間間隔線性增長。
  );
  insert into odps_sink 
  select 
     id, 
     len, 
     content,
     date_dormat(c, 'yyMMdd') as ds
  from source;