ApsaraDB for SelectDB:Group Commit

Nasıl çalışır?

Group Commit ayrı bir içe aktarma yöntemi değildir. Nitelikli INSERT INTO VALUES, Stream Load ve HTTP Stream isteklerini yakalar ve bunları tek bir dahili commit işleminde toplar. Aşağıdaki eşiklerden herhangi birine ulaşıldığında otomatik commit tetiklenir:

• Commit aralığı dolduğunda (varsayılan: 10 saniye)

• Birikmiş veri boyutu sınıra ulaştığında (varsayılan: 64 MB)

Bu davranışı kontrol eden üç mod bulunur:

Sync ve async modları arasında seçim yapma:

• Verilerin içe aktarıldıktan hemen sonra görünür olması gereken yüksek eşzamanlı senaryolar için sync_mode kullanın. Bu mod, işlem commit edilene kadar bloke olur; böylece dönen sonuç verilerin kalıcı hale geldiğini ve hemen sorgulanabilir olduğunu doğrular.

• Yazma gecikme süresinin en yüksek önceliğe sahip olduğu durumlarda async_mode kullanın. Sunucu, dahili commit işleminin tamamlanmasını beklemeden veriler WAL'a yazıldığı anda içe aktarmayı onaylar. Dahili commit başarısız olursa verileri kurtarmak için WAL günlükleri kullanılır. Veriler, içe aktarma yanıtı döndükten hemen sonra görünür olmaz.

Örnek tabloyu oluşturun

Bu konudaki örnekler aşağıdaki tabloyu kullanır:

CREATE TABLE `dt` (
    `id` int(11) NOT NULL,
    `name` varchar(50) NULL,
    `score` int(11) NULL
) ENGINE=OLAP
DUPLICATE KEY(`id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 1;

JDBC kullanarak veri içe aktarın

ApsaraDB for SelectDB, Java Database Connectivity (JDBC) üzerinden MySQL'in hazırlanmış deyim (prepared statement) özelliğini destekler. Hazırlanmış deyimler kullanıldığında, SQL deyimleri ve içe aktarma planları oturum düzeyindeki bellekte önbelleğe alınır; bu da tekrarlanan ekleme işlemlerinde CPU ek yükünü azaltır.

1. MySQL konektör bağımlılığını projenize ekleyin:

    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.49</version>
    </dependency>

2. Sunucu tarafında hazırlanmış deyimler etkinleştirilmiş şekilde JDBC URL'sini oluşturun:

    jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030/db?useServerPrepStmts=true

3. group_commit oturum değişkenini ayarlayın. Aşağıdaki yaklaşımlardan birini kullanın:

   • JDBC URL'sine ekleyin: `` jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030/db?useServerPrepStmts=true&sessionVariables=group_commit=async_mode ``

   • Bağlantı sırasında bir SQL deyimi yürütün: ``java try (Statement statement = conn.createStatement()) { statement.execute("SET group_commit = async_mode;"); } ``

4. Satır eklemek için hazırlanmış bir deyim kullanın:

    private static final String JDBC_DRIVER = "com.mysql.jdbc.Driver";
    private static final String URL_PATTERN = "jdbc:mysql://%s:%d/%s?useServerPrepStmts=true";
    private static final String HOST = "selectdb-cn-****.selectdbfe.rds.aliyuncs.com";
    private static final int PORT = 9030;
    private static final String DB = "db";
    private static final String TBL = "dt";
    private static final String USER = "admin";
    private static final String PASSWD = "***";
    private static final int INSERT_BATCH_SIZE = 10;
   
    public static void main(String[] args) {
        groupCommitInsert();
        //groupCommitInsertBatch
    }
   
    private static void groupCommitInsert() throws Exception {
        Class.forName(JDBC_DRIVER);
        try (Connection conn = DriverManager.getConnection(String.format(URL_PATTERN, HOST, PORT, DB), USER, PASSWD)) {
            // set session variable 'group_commit'
            try (Statement statement = conn.createStatement()) {
                statement.execute("SET group_commit = async_mode;");
            }
   
            String query = "INSERT INTO " + TBL + " VALUES(?, ?, ?)";
            try (PreparedStatement stmt = conn.prepareStatement(query)) {
                for (int i = 0; i < INSERT_BATCH_SIZE; i++) {
                    stmt.setInt(1, i);
                    stmt.setString(2, "name" + i);
                    stmt.setInt(3, i + 10);
                    int result = stmt.executeUpdate();
                    System.out.println("rows: " + result);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
   
    private static void groupCommitInsertBatch() throws Exception {
        Class.forName(JDBC_DRIVER);
        // add rewriteBatchedStatements=true and cachePrepStmts=true in JDBC url
        // set session variables by sessionVariables=group_commit=async_mode in JDBC url
        try (Connection conn = DriverManager.getConnection(
                String.format(URL_PATTERN + "&rewriteBatchedStatements=true&cachePrepStmts=true&sessionVariables=group_commit=async_mode", HOST, PORT, DB), USER, PASSWD)) {
   
            String query = "INSERT INTO " + TBL + " VALUES(?, ?, ?)";
            try (PreparedStatement stmt = conn.prepareStatement(query)) {
                for (int j = 0; j < 5; j++) {
                    // 10 rows per insert
                    for (int i = 0; i < INSERT_BATCH_SIZE; i++) {
                        stmt.setInt(1, i);
                        stmt.setString(2, "name" + i);
                        stmt.setInt(3, i + 10);
                        stmt.addBatch();
                    }
                    int[] result = stmt.executeBatch();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

INSERT INTO kullanarak veri içe aktarın

INSERT INTO deyimlerini çalıştırmadan önce group_commit oturum değişkenini ayarlayarak Group Commit'i etkinleştirin.

Eşzamansız mod — veriler arka planda toplu halde commit edilir:

-- Enable group commit in async mode. The default value is off_mode.
mysql> SET group_commit = async_mode;

-- The returned label starts with "group_commit", confirming group commit is active.
mysql> INSERT INTO dt VALUES(1, 'Bob', 90), (2, 'Alice', 99);
Query OK, 2 rows affected (0.05 sec)
{'label':'group_commit_a145ce07f1c972fc-bd2c54597052a9ad', 'status':'PREPARE', 'txnId':'181508'}

-- Consecutive inserts sharing the same label and txnId are batched into one import job.
mysql> INSERT INTO dt(id, name) VALUES(3, 'John');
Query OK, 1 row affected (0.01 sec)
{'label':'group_commit_a145ce07f1c972fc-bd2c54597052a9ad', 'status':'PREPARE', 'txnId':'181508'}

-- Data is not visible immediately after the import returns.
mysql> SELECT * FROM dt;
Empty SET (0.01 sec)

-- After ~10 seconds (controlled by group_commit_interval), data becomes visible.
mysql> SELECT * FROM dt;
+------+-------+-------+
| id   | name  | score |
+------+-------+-------+
|    1 | Bob   |    90 |
|    2 | Alice |    99 |
|    3 | John  |  NULL |
+------+-------+-------+
3 rows in set (0.02 sec)

Eşzamanlı mod — yalnızca işlem commit edildikten sonra yanıt döner; veriler hemen görünür olur:

-- Enable group commit in sync mode.
mysql> SET group_commit = sync_mode;

-- The commit interval is controlled by group_commit_interval. The call blocks until the transaction commits.
mysql> INSERT INTO dt VALUES(4, 'Bob', 90), (5, 'Alice', 99);
Query OK, 2 rows affected (10.06 sec)
{'label':'group_commit_d84ab96c09b60587_ec455a33cb0e9e87', 'status':'PREPARE', 'txnId':'3007', 'query_id':'fc6b94085d704a94-a69bfc9a202e66e2'}

-- Data is visible immediately.
mysql> SELECT * FROM dt;
+------+-------+-------+
| id   | name  | score |
+------+-------+-------+
|    1 | Bob   |    90 |
|    2 | Alice |    99 |
|    3 | John  |  NULL |
|    4 | Bob   |    90 |
|    5 | Alice |    99 |
+------+-------+-------+
5 rows in set (0.03 sec)

Group Commit'i devre dışı bırakın:

mysql> SET group_commit = off_mode;

Stream Load kullanarak veri içe aktarın

Günlük kaydı veya HTTP tabanlı işlem hatları için, Stream Load isteklerinde Group Commit'i etkinleştirmek amacıyla group_commit üst bilgisini iletin. Stream Load hakkında daha fazla bilgi için bkz. Stream Load.

1. data.csv adlı bir dosya oluşturun:

    6,Amy,60
    7,Ross,98

2. İçe aktarmayı uygun mod üst bilgisiyle çalıştırın: Eşzamansız mod:

    # Pass group_commit:async_mode as a request header.
    curl --location-trusted -u {user}:{passwd} -T data.csv \
      -H "group_commit:async_mode" \
      -H "column_separator:," \
      http://{selectdbHost}:{selectdbHttpPort}/api/db/dt/_stream_load

   Beklenen yanıt:

    {
        "TxnId": 7009,
        "Label": "group_commit_c84d2099208436ab_96e33fda01eddba8",
        "Comment": "",
        "GroupCommit": true,
        "Status": "Success",
        "Message": "OK",
        "NumberTotalRows": 2,
        "NumberLoadedRows": 2,
        "NumberFilteredRows": 0,
        "NumberUnselectedRows": 0,
        "LoadBytes": 19,
        "LoadTimeMs": 35,
        "StreamLoadPutTimeMs": 5,
        "ReadDataTimeMs": 0,
        "WriteDataTimeMs": 26
    }

   Eşzamanlı mod:

    # Pass group_commit:sync_mode as a request header.
    curl --location-trusted -u {user}:{passwd} -T data.csv \
      -H "group_commit:sync_mode" \
      -H "column_separator:," \
      http://{selectdbHost}:{selectdbHttpPort}/api/db/dt/_stream_load

   Beklenen yanıt:

    {
        "TxnId": 3009,
        "Label": "group_commit_d941bf17f6efcc80_ccf4afdde9881293",
        "Comment": "",
        "GroupCommit": true,
        "Status": "Success",
        "Message": "OK",
        "NumberTotalRows": 2,
        "NumberLoadedRows": 2,
        "NumberFilteredRows": 0,
        "NumberUnselectedRows": 0,
        "LoadBytes": 19,
        "LoadTimeMs": 10044,
        "StreamLoadPutTimeMs": 4,
        "ReadDataTimeMs": 0,
        "WriteDataTimeMs": 10038
    }

   Yanıttaki "GroupCommit": true değeri, Group Commit'in etkin olduğunu doğrular. Etiket her zaman group_commit ile başlar.

Otomatik commit eşiklerini yapılandırın

ALTER TABLE kullanarak tablo başına commit aralığını veya veri boyutu eşiklerini ayarlayın.

-- Change the commit interval to 2 seconds.
ALTER TABLE dt SET ("group_commit_interval_ms" = "2000");

-- Change the data size threshold to 128 MB.
ALTER TABLE dt SET ("group_commit_data_bytes" = "134217728");

Sınırlamalar

insert table ${table_name} is blocked on schema change