Hologres:Simple Log Service からのデータのインポート

この例では、Simple Log Service によって提供されるシミュレートデータがソースデータとして使用されます。データは、ゲームプラットフォームのログオンと消費ログをシミュレートすることによって生成されます。ビジネスデータを使用することもできます。

Hologres でデータを受信するために使用するテーブルを作成します。クエリの要件に基づいて、テーブルのフィールドにインデックスを作成できます。これは、クエリの効率を向上させるのに役立ちます。インデックスの詳細については、概要をご参照ください。次の DDL ステートメントは、テーブルを作成するために使用されます。

CREATE TABLE sls_flink_holo (
    content JSONB ,
    operation TEXT,
    uid TEXT,
    topic  TEXT ,
    source TEXT ,
    c__timestamp TIMESTAMPTZ,
    receive_time BIGINT,
    PRIMARY KEY (uid)
  );
  -- テーブルを作成するためのDDLステートメントです。

Realtime Compute for Apache Flink を使用して Simple Log Service から Hologres にデータを書き込むには、次の手順を実行します。

1. Realtime Compute for Apache Flink を使用して Simple Log Service からデータを読み取ります。詳細については、Simple Log Serviceコネクタをご参照ください。

2. Realtime Compute for Apache Flink を使用して Hologres にデータを書き込みます。詳細については、Hologres コネクタをご参照ください。

次の SQL ジョブのサンプルは、Realtime Compute for Apache Flink を使用して Simple Log Service から Hologres にデータを書き込む方法の例を示しています。JSON 形式のフィールドは、Hologres テーブルの JSON データ型として直接書き込まれます。JSON データ型は、Realtime Compute for Apache Flink の VARCHAR 型に対応します。

CREATE TEMPORARY TABLE sls_input (
    content STRING,
    operation STRING,
    uid STRING,
    `__topic__` STRING METADATA VIRTUAL,
    `__source__` STRING METADATA VIRTUAL,
    `__timestamp__` BIGINT METADATA VIRTUAL,
    `__tag__` MAP<VARCHAR, VARCHAR> METADATA VIRTUAL
  )
WITH (
    'connector' = 'sls',
    'endpoint' = 'Internal endpoint of the Simple Log Service project',-- Simple Log Serviceプロジェクトにアクセスするために使用される内部エンドポイント。
    'accessid'='AccessKey ID of your Alibaba Cloud account',-- Alibaba CloudアカウントのAccessKey ID。
    'accesskey'='AccessKey secret of your Alibaba Cloud account',-- Alibaba CloudアカウントのAccessKeyシークレット。
    'starttime' = '2024-08-30 00:00:00',-- ログの消費を開始する時刻。
    'project'='Project name',-- Simple Log Serviceプロジェクトの名前。
    'logstore'='Logstore name'-- ログストアの名前。
  );

CREATE TEMPORARY TABLE hologres_sink (
    content VARCHAR,
    operation VARCHAR,
    uid VARCHAR,
    topic  STRING ,
    source STRING ,
    c__timestamp TIMESTAMP ,
    receive_time BIGINT
  )
WITH (
    'connector' = 'hologres',
    'dbname'='Name of the Hologres database',-- Hologresデータベースの名前。
    'tablename'='Name of the Hologres table',-- データを書き込むHologresテーブルの名前。
    'username'='AccessKey ID of your Alibaba Cloud account',-- Alibaba CloudアカウントのAccessKey ID。
    'password'='AccessKey secret of your Alibaba Cloud account',-- Alibaba CloudアカウントのAccessKeyシークレット。
    'endpoint'='VPC endpoint of your Hologres instance.'-- HologresインスタンスのVPCエンドポイント。
  );

INSERT INTO hologres_sink
SELECT
   content,
   operation,
   uid,
   `__topic__` ,
   `__source__` ,
    CAST (
    FROM_UNIXTIME (`__timestamp__`) AS TIMESTAMP
  ),
   CAST (__tag__['__receive_time__'] AS BIGINT) AS receive_time
FROM
  sls_input;

Realtime Compute for Apache Flink を使用して Simple Log Service から Hologres に書き込まれたデータをクエリできます。ビジネス要件に基づいてデータを開発することもできます。

この例では、Simple Log Service によって提供されるシミュレートデータがソースデータとして使用されます。データは、ゲームプラットフォームのログオンと消費ログをシミュレートすることによって生成されます。ビジネスデータを使用することもできます。

Hologres でデータを受信するために使用するテーブルを作成します。クエリの要件に基づいて、テーブルのフィールドにインデックスを作成できます。これは、クエリの効率を向上させるのに役立ちます。インデックスの詳細については、概要をご参照ください。次の DDL ステートメントは、テーブルを作成するために使用されます。

BEGIN;
CREATE  TABLE sls_dw_holo (
    content JSONB ,
    operation TEXT,
    uid TEXT,
    C_Topic  TEXT ,
    C_Source TEXT ,
    timestamp BIGINT,
    PRIMARY KEY (uid)
  );
  CALL set_table_property('sls_dw_holo', 'distribution_key', 'uid');
  CALL set_table_property('sls_dw_holo', 'event_time_column', 'timestamp');
COMMIT;
                        -- テーブルを作成するためのDDLステートメントです。

DataWorks の Data Integration サービスを使用して Hologres にデータを書き込む前に、指定された DataWorks ワークスペースに Simple Log Service データソースと Hologres データソースを追加する必要があります。

• Log Service LogHub データソースを設定します。詳細については、Simple Log Service データソースをご参照ください。

• Hologres データソースを設定します。詳細については、Hologresデータソースをご参照ください。

Data Integration でリアルタイム同期ノードを作成して実行します。詳細については、DataStudio でリアルタイム同期タスクを作成するとリアルタイム同期ノードの運用と保守をご参照ください。

リアルタイム同期ノードが実行された後、DataWorks の Data Integration サービスを使用して Hologres に書き込まれたデータを Hologres でクエリできます。