E-MapReduce:Import and export data between HDFS and ClickHouse

Use HDFS table engines to import data

HDFS table engines of ClickHouse can read file data in a specific format from a specified HDFS address. Syntax:

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

1. Create an HDFS engine table and prepare data.

   1. Download the sample file orders.csv and upload the file to a directory of the Hadoop cluster. In this example, the sample file is uploaded to the root directory of the Hadoop cluster.

   2. Execute the following statements to create a database named hdfs and an HDFS table named hdfs.orders:

      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');

      Note

      In this example, the sample data is uploaded to the root directory of the Hadoop cluster. In the sample code, 192.168.**.** is the private IP address of the core-1-1 node in the Hadoop cluster. You can log on to the EMR console, click EMR on ECS in the left-side navigation pane, click Nodes in the Actions column of the Hadoop cluster, and then view the private IP address on the Nodes tab.

2. Execute the following statements to import data into the product.orders_all table:

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

3. Execute the following statement to check data consistency:

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

Use HDFS table functions to import data

HDFS table functions of ClickHouse can read file data from a specified HDFS address and return a table with a specified schema. Syntax:

hdfs(uri, format, structure);

1. Download the sample file orders.csv and upload the file to a directory of the Hadoop cluster. In this example, the sample file is uploaded to the root directory of the Hadoop cluster.

2. Execute the following statements to import data into the product.orders_all table:

   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');

   Note

   In this example, the sample data is uploaded to the root directory of the Hadoop cluster. In the sample code, 192.168.**.** is the private IP address of the core-1-1 node in the Hadoop cluster. You can log on to the EMR console, click EMR on ECS in the left-side navigation pane, click Nodes in the Actions column of the Hadoop cluster, and then view the private IP address on the Nodes tab.

3. Execute the following statement to check data consistency:

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

Use HDFS table engines to export data

1. Execute the following statements to create an HDFS table:

   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');

   Note

   In this example, data is exported to the root directory of the Hadoop cluster. In the sample code, 192.168.**.** is the private IP address of the core-1-1 node in the Hadoop cluster. You can log on to the EMR console, click EMR on ECS in the left-side navigation pane, click Nodes in the Actions column of the Hadoop cluster, and then view the private IP address on the Nodes tab.

2. Execute the following statements to export data. The data is stored in the related directory by using an HDFS table engine:

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

   Note

   During the data export, ClickHouse creates a file in the related path and writes data to the file. By default, if the file to which you want to write data already exists in HDFS, the data export fails. You can configure parameters for EMR clusters of V5.8.0, V3.45.0, and their later minor versions to prevent the failure.

3. Execute the following statement to check data consistency:

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

Use HDFS table functions to export data

1. Execute the following statements to export data:

   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;

   Note

   During the data export, ClickHouse creates a file in the related path and writes data to the file. By default, if the file to which you want to write data already exists in HDFS, the data export fails. You can configure parameters for EMR clusters of V5.8.0, V3.45.0, and their later minor versions to prevent the failure.

2. Execute the following statement to check data consistency:

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;