E-MapReduce:OSS と ClickHouse 間のデータのインポートとエクスポート

S3 テーブルエンジンを使用してデータをインポートする

ClickHouse の Hadoop Distributed File System（HDFS）テーブルエンジンは、指定された OSS アドレスから特定の形式のファイルデータを読み取ることができます。 構文：

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
  name1 [type1] [NULL|NOT NULL] [DEFAULT|MATERIALIZED|ALIAS expr1] [compression_codec] [TTL expr1],
  name2 [type2] [NULL|NOT NULL] [DEFAULT|MATERIALIZED|ALIAS expr2] [compression_codec] [TTL expr2],
  ...
)
ENGINE = S3(path, [access_key_id, access_key_secret,] format, [compression]);

1. OSS からデータを読み取るテーブルを作成します。

   1. サンプルオブジェクト orders.csv をダウンロードし、OSS にアップロードします。 この例では、オブジェクトは test という名前の OSS バケットのルートディレクトリにアップロードされます。

   2. 次のステートメントを実行して、OSS テーブルを作成し、S3 テーブルエンジンを使用します。

      CREATE DATABASE IF NOT EXISTS oss ON CLUSTER cluster_emr;
      CREATE TABLE oss.orders_oss
      (
        uid UInt32,
        date DateTime,
        skuId UInt32,
        order_revenue UInt32
      ) ENGINE = S3('http://test.oss-cn-beijing-internal.aliyuncs.com/orders.csv', '<access_key_id>', '<access_key_secret>', 'CSV');

      説明

      サンプルコードでは、http://test.oss-cn-beijing-internal.aliyuncs.com/orders.csv は、中国（北京）リージョンの test という名前の OSS バケットにある orders.csv オブジェクトへのパスです。

2. 次のステートメントを実行して、product.orders_all テーブルにデータをインポートします。

   INSERT INTO product.orders_all
   SELECT
     uid,
     date,
     skuId,
     order_revenue
   FROM
     oss.orders_oss;

3. 次のステートメントを実行して、テーブル内のデータを表示し、データの整合性を確認します。

   • orders_all テーブルのデータをクエリします。

     SELECT count(1) FROM product.orders_all;

   • orders_oss テーブルのデータをクエリします。

     SELECT count(1) FROM oss.orders_oss;

S3 テーブル関数を使用してデータをインポートする

ClickHouse の S3 テーブル関数は、指定された HDFS アドレスからファイルデータを読み取り、指定されたスキーマを持つテーブルを返します。 構文：

s3(path, [access_key_id, access_key_secret,] format, structure, [compression])

1. S3 テーブル関数を使用して、ClickHouse クラスターにデータをインポートします。

   INSERT INTO product.orders_all
   SELECT
     uid,
     date,
     skuId,
     order_revenue
   FROM
     s3('http://test.oss-cn-beijing-internal.aliyuncs.com/orders.csv',
         '<your-access-key>',
         '<your-access-secret>',
         'CSV',
         'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32');

2. 次のステートメントを実行して、テーブル内のデータを表示し、データの整合性を確認します。

   • orders_all テーブルのデータをクエリします。

     SELECT count(1) FROM product.orders_all;

   • orders_oss テーブルのデータをクエリします。

     SELECT count(1) FROM oss.orders_oss;

S3 テーブルエンジンを使用してデータをエクスポートする

1. 次のステートメントを実行して、S3 テーブルを作成します。

   CREATE TABLE oss.orders_oss
   (
     uid UInt32,
     date DateTime,
     skuId UInt32,
     order_revenue UInt32
   ) ENGINE = S3('http://test.oss-cn-beijing-internal.aliyuncs.com/orders.csv', '<access_key_id>', '<access_key_secret>', 'CSV');

2. 次のステートメントを実行して、テーブルにデータを書き込みます。

   -- この例では、ビジネス テーブル product.orders_all が使用されます。
   INSERT INTO oss.orders_oss
   SELECT
     uid,
     date,
     skuId,
     order_revenue
   FROM
     product.orders_all;

   説明

   データのエクスポート中、ClickHouse は関連パスにオブジェクトを作成し、オブジェクトにデータを書き込みます。 デフォルトでは、データを書き込むオブジェクトが OSS に既に存在する場合、データのエクスポートは失敗します。 V5.8.0、V3.45.0 以降のマイナーバージョンの EMR クラスターの パラメーターを設定 して、エラーを防ぐことができます。

3. OSS コンソール でデータを表示します。

HDFS テーブル関数を使用してデータをエクスポートする

1. 次のステートメントを実行して、データをエクスポートします。

   INSERT INTO FUNCTION
   s3('http://test.oss-cn-beijing-internal.aliyuncs.com/orders.csv',
         '<your-access-key>',
         '<your-access-secret>',
         'CSV',
         'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32')
   SELECT
     uid,
     date,
     skuId,
     order_revenue
   FROM
     product.orders_all;

   説明

   データのエクスポート中、ClickHouse は関連パスにオブジェクトを作成し、オブジェクトにデータを書き込みます。 デフォルトでは、データを書き込むオブジェクトが OSS に既に存在する場合、データのエクスポートは失敗します。 V5.8.0、V3.45.0 以降のマイナーバージョンの EMR クラスターの パラメーターを設定 して、エラーを防ぐことができます。

2. OSS コンソール でデータを表示します。