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

HDFS テーブルエンジンの使用

HDFS テーブルエンジンは、HDFS 内のファイルをバックエンドとする ClickHouse テーブルを作成します。構文は次のとおりです：

CREATE TABLE [IF NOT EXISTS] [db.]table_name
(
    name1 [type1],
    name2 [type2],
    ...
)
ENGINE = HDFS(uri, format);

手順

1. orders.csv サンプルファイルをダウンロードし、ご利用の Hadoop クラスターにアップロードします。この例では、ファイルをルートディレクトリにアップロードします。

2. HDFS データベースと、アップロードされたファイルを指す HDFS バックエンドのテーブルを作成します：

   CREATE DATABASE IF NOT EXISTS hdfs ON CLUSTER cluster_emr;
   
   CREATE TABLE IF NOT EXISTS hdfs.orders ON CLUSTER cluster_emr
   (
       `uid`           UInt32,
       `date`          DateTime,
       `skuId`         UInt32,
       `order_revenue` UInt32
   )
   ENGINE = HDFS('hdfs://192.168.**.**:9000/orders.csv', 'CSV');

   192.168.. を、ご利用の Hadoop クラスターの core-1-1 ノードのプライベート IP アドレスに置き換えてください。この IP アドレスを確認するには、EMR コンソールに移動し、左側のナビゲーションウィンドウで [EMR on ECS] をクリックし、Hadoop クラスターの行の [操作] 列にある [ノード] をクリックします。IP アドレスは [ノード] タブに表示されます。

3. データを product.orders_all にインポートします：

   INSERT INTO product.orders_all
   SELECT uid, date, skuId, order_revenue
   FROM hdfs.orders;

4. hdfs.orders にあって product.orders_all にない行を確認することで、データ整合性を検証します：

   SELECT a.*
   FROM hdfs.orders a
   LEFT ANTI JOIN product.orders_all
   USING uid;

   結果セットが空であれば、インポートが成功したことを示します。

HDFS テーブル関数の使用

hdfs() テーブル関数は、永続テーブルを作成せずに HDFS からデータをインラインで読み取ります。構文は次のとおりです：

hdfs(uri, format, structure)

手順

1. orders.csv サンプルファイルをダウンロードし、ご利用の Hadoop クラスターにアップロードします。この例では、ファイルをルートディレクトリにアップロードします。

2. データを product.orders_all にインポートします：

   INSERT INTO product.orders_all
   SELECT uid, date, skuId, order_revenue
   FROM hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV',
             'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32');

   192.168.. を、ご利用の Hadoop クラスターの core-1-1 ノードのプライベート IP アドレスに置き換えてください。この IP アドレスを確認するには、EMR コンソールに移動し、左側のナビゲーションウィンドウで [EMR on ECS] をクリックし、Hadoop クラスターの行の [操作] 列にある [ノード] をクリックします。IP アドレスは [ノード] タブに表示されます。

3. データ整合性を検証します：

   SELECT a.*
   FROM hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV',
             'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32')
   LEFT ANTI JOIN product.orders_all a
   USING uid;

   結果セットが空であれば、インポートが成功したことを示します。

HDFS テーブルエンジンの使用

1. エクスポート先を指す HDFS バックエンドのテーブルを作成します：

   CREATE DATABASE IF NOT EXISTS hdfs ON CLUSTER cluster_emr;
   
   CREATE TABLE IF NOT EXISTS hdfs.orders ON CLUSTER cluster_emr
   (
       `uid`           UInt32,
       `date`          DateTime,
       `skuId`         UInt32,
       `order_revenue` UInt32
   )
   ENGINE = HDFS('hdfs://192.168.**.**:9000/orders.csv', 'CSV');

   192.168.. を、ご利用の Hadoop クラスターの core-1-1 ノードのプライベート IP アドレスに置き換えてください。この IP アドレスを確認するには、EMR コンソールに移動し、左側のナビゲーションウィンドウで [EMR on ECS] をクリックし、Hadoop クラスターの行の [操作] 列にある [ノード] をクリックします。IP アドレスは [ノード] タブに表示されます。

2. データをエクスポートします：

   INSERT INTO hdfs.orders
   SELECT uid, date, skuId, order_revenue
   FROM product.orders_all;

3. product.orders_all にあって hdfs.orders にない行を確認することで、データ整合性を検証します：

   SELECT a.*
   FROM hdfs.orders
   RIGHT ANTI JOIN product.orders_all a
   USING uid;

   結果セットが空であれば、エクスポートが成功したことを示します。

HDFS テーブル関数の使用

1. データをエクスポートします：

   INSERT INTO FUNCTION
       hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV',
            'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32')
   SELECT uid, date, skuId, order_revenue
   FROM product.orders_all;

2. データ整合性を検証します：

   SELECT a.*
   FROM hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV',
             'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32')
   RIGHT ANTI JOIN product.orders_all a
   USING uid;

   結果セットが空であれば、エクスポートが成功したことを示します。