Lindorm:Access column-oriented data

Background information

Lindorm column-oriented storage is a columnar distributed storage service. Column-oriented storage is suitable for storing large amounts of semi-structured and structured data in scenarios such as Internet of Vehicles (IoV), Internet of Things (IoT), orders, and logs. Column-oriented storage provides the following core capabilities:

• Computing and analytics

  The Lindorm compute engine can access column-oriented data to perform interactive analytics and online computing. Column-oriented storage provides the data distribution feature and rich indexing capabilities, which can effectively accelerate data positioning and arrangement in the computing process. You can add, remove, modify, and query large amounts of primary key data by using SQL statements.

• High throughput

  Column-oriented storage supports horizontal scaling and provides the capability to read and write terabytes of data per minute. This makes it suitable for high-throughput data scenarios such as quick import of IoV data, model training dataset access, and large-scale report analysis and production.

• Cost-effectiveness

  Lindorm implements technologies such as columnar high compression ratio algorithms, high-density low-cost media, hot and cold data separation, compression encoding, and cold data archive storage. Compared with self-managed storage, Lindorm column-oriented storage significantly reduces storage costs, allowing you to archive and store large amounts of data at reduced costs.

• High availability

  By using technologies such as erasure coding, Lindorm column-oriented storage ensures the high availability of distributed datasets and eliminates the risk of single points of failure (SPOFs).

• Compatibility with open source services

  Column-oriented storage is compatible with multiple open-source standard APIs, such as the Iceberg API. Column-oriented storage can also connect to compute engines such as Spark and Flink. This way, column-oriented storage can be seamlessly integrated into mainstream data ecosystems.

• Hot and cold data separation

  You can store hot and cold data on different storage media based on your business requirements. This helps reduce the performance overhead caused by access to cold data, while effectively lowering storage costs.

Prerequisites

• You have read Precautions.

• Depending on the job mode, make sure that you have completed the following operations:

  • JDBC jobs: Use JDBC in application development.

  • JAR jobs: Create a job in Java.

  • Python jobs: Create a job in Python.

• If you want to use the cold and hot data separation feature, make sure that you have enabled Capacity storage. For more information, see Enable Capacity storage.

Feature description

DDL

DML

Hot and cold data storage

In most cases, frequently accessed data is stored in high-performance storage and infrequently accessed historical data is migrated to low-cost storage. This helps reduce the storage costs and improve performance. Lindorm supports hot and cold data separation at three levels: L1, L2, and L3. You can develop policies for conversion between hot and cold data based on your business requirements. You can use the data lake service to automatically dump data between hot and cold data storage. This helps you effectively manage storage costs without extending the response time of key services.

Parameters

When you create a column-oriented table, you can define the cold or hot storage (CHS) attribute to specify how to dump data between cold and hot storage based on the time partition of the column-oriented table. The following table describes how to configure CHS parameters.

Examples

• Example 1:

  Create a table named table0. The partition field names are year, month, and day. Based on the hot and cold data separation policy that you specify, data that was stored one month (2,592,000 seconds) ago is automatically dumped to Capacity storage. You can execute the following statement to create a table:

  CREATE TABLE table0 (col0 INT,year STRING,month STRING,day STRING)
  PARTITIONED BY  (year,month)
  TBLPROPERTIES 
  'CHS'='2592000',
  'CHS_L2'='storagetype=CAPACITY_CLOUD_STORAGE',
  'CHS_EXP'='toSec(year,yyyy,month,MM,day,dd)'
  );

• Example 2:

  Change the hot and cold data separation policy of table0. After the change, data that was stored one month (2,592,000 seconds) ago is automatically dumped to Capacity storage, and data that was stored three months (5,184,000 seconds) ago is automatically dumped to archive storage. You can execute the following statement to update a table:

  ALTER TABLE table0
  SET TBLPROPERTIES (
  'CHS'='2592000,5184000',
  'CHS_L2'='storagetype=CAPACITY_CLOUD_STORAGE',
  'CHS_L3'='storagetype=CLOUD_ARCHIVE_STORAGE',
  'CHS_EXP'='toSec(year,yyyy,month,MM,day,dd)'
  );

• Example 3:

  Create a table named table1. The partition field name is dt. Example: 2020/12/1. Based on the hot and cold data separation policy that you specify, data that was stored one month (2,592,000 seconds) ago is automatically dumped to Capacity storage, and data that was stored three months (5,184,000 seconds) ago is automatically dumped to archive storage. You can execute the following statement to create a table:

  CREATE TABLE table1 (col0 INT,dt STRING)
  PARTITIONED BY  (dt)
  TBLPROPERTIES (
  'CHS'='2592000,5184000',
  'CHS_L2'='storagetype=CAPACITY_CLOUD_STORAGE',
  'CHS_L3'='storagetype=CLOUD_ARCHIVE_STORAGE',
  'CHS_EXP'='toSec(dt,yyyy/MM/dd)'
  );

Precautions

• You can specify the CHS parameter when you create a table. If you want to change the hot and cold data separation policy, you can execute the ALTER TABLE ...SET TBLPROPERTIES... statement.

• If you specify the CHS parameter incorrectly, the table can be created and updated as expected, but data cannot be automatically dumped between cold and hot storage.

• Hot and cold data dumping is triggered in asynchronous mode. During and after data dumping, data access is not affected, but access performance may vary based on different storage media.

• You can only implement the hot and cold data separation policy based on the time partition in a column-oriented table.

Best practices

You can use the following methods to accelerate data queries or computing.

Specify a primary key

If large amounts of datasets are stored in a table, you can specify a primary key to accelerate queries. A smaller primary key data range provides a better acceleration result.

In this example, a sample table is created by executing the following statements:

USE lindorm_columnar;
CREATE TABLE orders (
o_orderkey       INT NOT NULL,
o_custkey        INT,
o_orderstatus    STRING,
o_totalprice     DOUBLE,
o_orderdate      STRING,
o_orderpriority  STRING,
o_clerk          STRING,
o_shippriority   INT,
o_comment        STRING)
PARTITIONED BY (bucket(1024,o_orderkey))
TBLPROPERTIES(
'primary-key' = 'o_orderkey');           

• Example 1:

  USE lindorm_columnar;
  SELECT * FROM orders WHERE o_orderkey=18394;

• Example 2:

  USE lindorm_columnar;
  SELECT count(*) FROM orders WHERE o_orderkey>100000 AND o_orderkey<200000;

• Example 3:

  USE lindorm_columnar;
  SELECT count(*) FROM orders WHERE o_orderkey>100000;

Add a partition

In a column-oriented storage engine of Lindorm, partitions are physically isolated from each other. You can add a partition to accelerate data queries.

In this example, a sample table is created by executing the following statements:

USE lindorm_columnar;
CREATE TABLE orders (
o_orderkey       INT NOT NULL,
o_custkey        INT,
o_orderstatus    STRING,
o_totalprice     DOUBLE,
o_orderdate      STRING NOT NULL,
o_orderpriority  STRING,
o_clerk          STRING,
o_shippriority   INT,
o_comment        STRING)
PARTITIONED BY (o_orderdate, bucket(1024,o_orderkey))
TBLPROPERTIES(
'primary-key' = 'o_orderdate,o_orderkey');

• Example 1:

  USE lindorm_columnar;
  SELECT o_orderdate, count(*) FROM orders WHERE o_orderdate='2022-01-01' GROUP BY o_orderdate;

• Example 2:

  USE lindorm_columnar;
  SELECT o_orderdate, count(*) FROM orders WHERE o_orderdate>='2022-01-01' AND o_orderdate<='2022-01-07' GROUP BY o_orderdate;

CREATE TABLE mydb.mytable (
  id INT NOT NULL, 
  city STRING NOT NULL, 
  name STRING, 
  score INT)
partitioned by (city, bucket(4, id))
tblproperties('primary-key' = 'id,city');

ALTER TABLE mydb.mytable SET TBLPROPERTIES ('read.scan-major-rewritten-files-only' = true);

USE lindorm_columnar;
CREATE TABLE orders (
o_orderkey       INT NOT NULL,
o_custkey        INT,
o_orderstatus    STRING,
o_totalprice     DOUBLE ,
o_orderdate      STRING ,
o_orderpriority  STRING,
o_clerk          STRING,
o_shippriority   INT,
o_comment        STRING)
PARTITIONED BY (bucket(1024,o_orderkey))
TBLPROPERTIES(
'primary-key' = 'o_orderkey',
'read.scan-major-rewritten-files-only' = 'true');

ALTER TABLE orders WRITE ORDERED BY o_shippriority,o_totalprice;

CALL lindorm_columnar.system.rewrite_data_files(table => 'orders');