:Log Service テーブルの作成

Log Service の概要

Log Service は、Alibaba Group が多くのビッグデータシナリオで開発し、テストしたオールインワンのリアルタイムデータロギングサービスです。 Log Service では、追加の開発作業なしで、データの取り込み、消費、配信、照会、分析などのタスクをすばやく完了できます。 これにより、O＆M と運用効率が向上し、データテクノロジー時代に大量のログを処理する機能を構築できます。 Log Service は、ストリーミングデータストレージシステムです。 Realtime Compute は、Log Service テーブルをソーステーブルとして作成することをサポートしています。 Log Service では、各データレコードは JSON に類似した形式です。 例は次のとおりです。

{
    "a": 1000,
    "b": 1234,
    "c": "li"
}
			

Realtime Compute は、次の DDL を定義する必要があります (sls は Log Service を示します )。

create table sls_stream(
  a int,
  b int,
  c VARCHAR
) with (
  type ='sls',
  endPoint ='yourEndpoint',
  accessId ='yourAccessId',
  accessKey ='yourAccessKey',
  startTime = 'yourStartTime',
  project ='yourProjectName',
  logStore ='yourLogStoreName',
  consumerGroup ='yourConsumerGroupName'
);
			

属性フィールド

現在、Flink SQL は、 Log Service の次の 3 つの属性フィールドの取得と、他のカスタムフィールドの書き込みをデフォルトでサポートしています。

属性フィールドを取得するには、通常のロジックに従ってフィールドを最初に宣言する必要があります。 次に、キーワード HEADER を型宣言の最後に追加します。 例：

• テストデータ

       __topic__:  ens_altar_flow  
          result:  {"MsgID":"ems0a","Version":"0.0.1"}
  					

• テスト文

  CREATE TABLE sls_log (
    __topic__  varchar HEADER,
    result     varchar  
  )
  WITH
  (
    type ='sls',
  );
  
  CREATE TABLE sls_out (
    name     varchar,
    MsgID    varchar,
    Version  varchar 
  )
  WITH
  (
    type ='RDS'
  );
  
  INSERT INTO sls_out
  SELECT 
  __topic__,
  JSON_VALUE(result,'$. MsgID'),
  JSON_VALUE(result,'$. Version')
  FROM
  sls_log
  					

• テスト結果

  name(VARCAHR)	MsgID(VARCAHR)	Version(VARCAHR)
  ens_altar_flow	ems0a	0.0.1

WITH パラメーター

フィールドタイプマッピング

次の表に、Log Service フィールドタイプと Realtime Compute フィールドタイプ間のマッピングを示します。 DDL 宣言でマッピングを使用することを推奨します。