446 Lesungen

Entwicklung des Data Sharding hin zu Automatisierung und Flexibilität in Apache Doris

von Shirley H.29m2024/07/20

Zu lang; Lesen

Apache Doris führt Auto Partition in V2.1.0 ein. Es vereinfacht seit Apache Doris 2.1.0 die DDL- und Partitionsverwaltung. Es ist bei der Verarbeitung großer Datenmengen nützlich und erleichtert Benutzern die Migration von anderen Datenbanksystemen zu Apache Doris.

featured image - Entwicklung des Data Sharding hin zu Automatisierung und Flexibilität in Apache Doris

Um große Datensätze zu verarbeiten, führen verteilte Datenbanken Strategien wie Partitionierung und Bucketing ein. Daten werden basierend auf bestimmten Regeln in kleinere Einheiten unterteilt und auf verschiedene Knoten verteilt, sodass Datenbanken eine parallele Verarbeitung durchführen können, um eine höhere Leistung und Flexibilität bei der Datenverwaltung zu erzielen.

Wie in vielen Datenbanken werden Daten in Apache Doris in Partitionen aufgeteilt, und dann wird eine Partition weiter in Buckets unterteilt. Partitionen werden normalerweise durch Zeit oder andere kontinuierliche Werte definiert. Dadurch können Abfrage-Engines die Zieldaten bei Abfragen schnell finden, indem irrelevante Datenbereiche bereinigt werden.

Beim Bucketing werden die Daten hingegen auf Basis der Hashwerte einer oder mehrerer Spalten verteilt, wodurch eine Datenverzerrung verhindert wird.

Vor Version 2.1.0 gab es zwei Möglichkeiten zum Erstellen von Datenpartitionen in Apache Doris:

Manuelle Partitionierung : Benutzer geben die Partitionen in der Anweisung zur Tabellenerstellung an oder ändern sie anschließend über DDL-Anweisungen.
Dynamische Partition : Das System verwaltet Partitionen automatisch innerhalb eines vordefinierten Bereichs basierend auf der Datenübertragungszeit.

In Apache Doris 2.1.0 haben wir Auto Partition eingeführt. Es unterstützt die Partitionierung von Daten nach RANGE oder LIST und verbessert die Flexibilität zusätzlich zur automatischen Partitionierung.

Entwicklung von Partitionierungsstrategien in Doris

Beim Entwurf der Datenverteilung konzentrieren wir uns mehr auf die Partitionsplanung, da die Auswahl der Partitionsspalten und Partitionsintervalle stark von den tatsächlichen Datenverteilungsmustern abhängt und ein guter Partitionsentwurf die Abfrage- und Speichereffizienz der Tabelle erheblich verbessern kann.

In Doris wird die Datentabelle hierarchisch in Partitionen und dann in Buckets unterteilt. Die Daten im selben Bucket bilden dann ein Datentableau , das die minimale physische Speichereinheit in Doris für Datenreplikation, Datenplanung zwischen Clustern und Lastausgleich darstellt.

Manuelle Partitionierung

Doris ermöglicht Benutzern die manuelle Erstellung von Datenpartitionen nach RANGE und LIST.

Für mit Zeitstempeln versehene Daten wie Protokolle und Transaktionsdatensätze erstellen Benutzer normalerweise Partitionen basierend auf der Zeitdimension. Hier ist ein Beispiel für die Anweisung CREATE TABLE:

 CREATE TABLE IF NOT EXISTS example_range_tbl ( `user_id` LARGEINT NOT NULL COMMENT "User ID", `date` DATE NOT NULL COMMENT "Data import date", `timestamp` DATETIME NOT NULL COMMENT "Data import timestamp", `city` VARCHAR(20) COMMENT "Location of user", `age` SMALLINT COMMENT "Age of user", `sex` TINYINT COMMENT "Sex of user", `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "Last visit date of user", `cost` BIGINT SUM DEFAULT "0" COMMENT "User consumption", `max_dwell_time` INT MAX DEFAULT "0" COMMENT "Maximum dwell time of user", `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "Minimum dwell time of user" ) ENGINE=OLAP AGGREGATE KEY(`user_id`, `date`, `timestamp`, `city`, `age`, `sex`) PARTITION BY RANGE(`date`) ( PARTITION `p201701` VALUES LESS THAN ("2017-02-01"), PARTITION `p201702` VALUES LESS THAN ("2017-03-01"), PARTITION `p201703` VALUES LESS THAN ("2017-04-01"), PARTITION `p2018` VALUES [("2018-01-01"), ("2019-01-01")) ) DISTRIBUTED BY HASH(`user_id`) BUCKETS 16 PROPERTIES ( "replication_num" = "1" );

Die Tabelle ist nach dem Datenimportdatum date partitioniert, und es wurden 4 Partitionen vorab erstellt. Innerhalb jeder Partition werden die Daten basierend auf dem Hashwert der user_id weiter in 16 Buckets unterteilt.

Mit diesem Partitionierungs- und Bucketing-Design muss das System beim Abfragen von Daten ab 2018 nur die Partition p2018 scannen. So sieht die Abfrage-SQL aus:

 mysql> desc select count() from example_range_tbl where date >= '20180101'; +--------------------------------------------------------------------------------------+ | Explain String(Nereids Planner) | +--------------------------------------------------------------------------------------+ | PLAN FRAGMENT 0 | | OUTPUT EXPRS: | | count(*)[#11] | | PARTITION: UNPARTITIONED | | | | ...... | | | | 0:VOlapScanNode(193) | | TABLE: test.example_range_tbl(example_range_tbl), PREAGGREGATION: OFF. | | PREDICATES: (date[#1] >= '2018-01-01') | | partitions=1/4 (p2018), tablets=16/16, tabletList=561490,561492,561494 ... | | cardinality=0, avgRowSize=0.0, numNodes=1 | | pushAggOp=NONE | | | +--------------------------------------------------------------------------------------+

Wenn die Daten ungleichmäßig auf die Partitionen verteilt sind, kann der Hash-basierte Bucketing-Mechanismus die Daten basierend auf der user_id weiter aufteilen. Dies hilft, eine ungleiche Lastverteilung auf einigen Maschinen während der Abfrage und Speicherung zu vermeiden.

In realen Geschäftsszenarien kann ein Cluster jedoch Zehntausende von Tabellen umfassen, was bedeutet, dass eine manuelle Verwaltung unmöglich ist.

 CREATE TABLE `DAILY_TRADE_VALUE` ( `TRADE_DATE` datev2 NOT NULL COMMENT 'Trade date', `TRADE_ID` varchar(40) NOT NULL COMMENT 'Trade ID', ...... ) UNIQUE KEY(`TRADE_DATE`, `TRADE_ID`) PARTITION BY RANGE(`TRADE_DATE`) ( PARTITION p_200001 VALUES [('2000-01-01'), ('2000-02-01')), PARTITION p_200002 VALUES [('2000-02-01'), ('2000-03-01')), PARTITION p_200003 VALUES [('2000-03-01'), ('2000-04-01')), PARTITION p_200004 VALUES [('2000-04-01'), ('2000-05-01')), PARTITION p_200005 VALUES [('2000-05-01'), ('2000-06-01')), PARTITION p_200006 VALUES [('2000-06-01'), ('2000-07-01')), PARTITION p_200007 VALUES [('2000-07-01'), ('2000-08-01')), PARTITION p_200008 VALUES [('2000-08-01'), ('2000-09-01')), PARTITION p_200009 VALUES [('2000-09-01'), ('2000-10-01')), PARTITION p_200010 VALUES [('2000-10-01'), ('2000-11-01')), PARTITION p_200011 VALUES [('2000-11-01'), ('2000-12-01')), PARTITION p_200012 VALUES [('2000-12-01'), ('2001-01-01')), PARTITION p_200101 VALUES [('2001-01-01'), ('2001-02-01')), ...... ) DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10 PROPERTIES ( ...... );

Im obigen Beispiel werden die Daten monatlich partitioniert. Dies erfordert, dass der Datenbankadministrator (DBA) jeden Monat manuell eine neue Partition hinzufügt und das Tabellenschema regelmäßig pflegt. Stellen Sie sich den Fall der Echtzeit-Datenverarbeitung vor, bei der Sie möglicherweise täglich oder sogar stündlich Partitionen erstellen müssen. Dies manuell zu tun, ist keine Option mehr. Aus diesem Grund haben wir die dynamische Partitionierung eingeführt.

Dynamische Partition

Mit Dynamic Partition erstellt und gewinnt Doris automatisch Datenpartitionen zurück, solange der Benutzer die Partitionseinheit, die Anzahl der historischen Partitionen und die Anzahl der zukünftigen Partitionen angibt. Diese Funktion basiert auf einem festen Thread im Doris-Frontend. Er führt kontinuierlich Abfragen durch und prüft, ob neue Partitionen erstellt oder alte Partitionen zurückgewonnen werden können, und aktualisiert das Partitionsschema der Tabelle.

Dies ist ein Beispiel für eine CREATE TABLE-Anweisung für eine Tabelle, die nach Tagen partitioniert ist. Die Parameter start und end sind auf -7 bzw. 3 gesetzt. Dies bedeutet, dass Datenpartitionen für die nächsten 3 Tage vorab erstellt werden und die historischen Partitionen, die älter als 7 Tage sind, wiederhergestellt werden.

 CREATE TABLE `DAILY_TRADE_VALUE` ( `TRADE_DATE` datev2 NOT NULL COMMENT 'Trade date', `TRADE_ID` varchar(40) NOT NULL COMMENT 'Trade ID', ...... ) UNIQUE KEY(`TRADE_DATE`, `TRADE_ID`) PARTITION BY RANGE(`TRADE_DATE`) () DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10 PROPERTIES ( "dynamic_partition.enable" = "true", "dynamic_partition.time_unit" = "DAY", "dynamic_partition.start" = "-7", "dynamic_partition.end" = "3", "dynamic_partition.prefix" = "p", "dynamic_partition.buckets" = "10" );

Im Laufe der Zeit behält die Tabelle immer Partitionen im Bereich von [current date - 7, current date + 3] bei. Die dynamische Partitionierung ist besonders nützlich für Szenarien mit Echtzeitdatenerfassung, z. B. wenn die ODS-Schicht (Operational Data Store) Daten direkt aus externen Quellen wie Kafka empfängt.

Die start und end definieren einen festen Bereich für die Partitionen, sodass der Benutzer die Partitionen nur innerhalb dieses Bereichs verwalten kann. Wenn der Benutzer jedoch mehr historische Daten einbeziehen muss, muss er den start erhöhen, was zu unnötigem Metadaten-Overhead im Cluster führen kann.

Daher muss beim Anwenden einer dynamischen Partition ein Kompromiss zwischen Benutzerfreundlichkeit und Effizienz der Metadatenverwaltung eingegangen werden.

Worte der Entwickler

Mit zunehmender Komplexität des Geschäfts ist die dynamische Partitionierung aus folgenden Gründen unzureichend:

Es unterstützt nur die Partitionierung nach BEREICH, aber nicht nach LISTE.
Es kann nur auf die aktuellen realen Zeitstempel angewendet werden.
Es unterstützt nur einen einzigen kontinuierlichen Partitionsbereich und kann keine Partitionen außerhalb dieses Bereichs aufnehmen.

Angesichts dieser funktionalen Einschränkungen haben wir mit der Planung eines neuen Partitionierungsmechanismus begonnen, der sowohl die Partitionsverwaltung automatisieren als auch die Wartung der Datentabellen vereinfachen kann.

Wir haben herausgefunden, dass die ideale Partitionierungsimplementierung Folgendes umfassen sollte:

Sie müssen nach der Tabellenerstellung keine Partitionen mehr manuell erstellen.
Sie müssen in der Lage sein, alle aufgenommenen Daten in entsprechenden Partitionen unterzubringen.

Ersteres steht für Automatisierung und Letzteres für Flexibilität. Der Kern bei der Realisierung beider besteht darin, die Partitionserstellung mit den tatsächlichen Daten zu verknüpfen.

Dann begannen wir, über Folgendes nachzudenken: Was wäre, wenn wir mit der Erstellung von Partitionen warten würden, bis die Daten aufgenommen sind, anstatt dies während der Tabellenerstellung oder durch regelmäßige Abfragen zu tun? Anstatt die Partitionsverteilung vorab zu konstruieren, können wir die Zuordnungsregeln „Daten zu Partition“ definieren, sodass die Partitionen erstellt werden, nachdem die Daten eintreffen.

Im Vergleich zur manuellen Partitionierung läuft dieser gesamte Vorgang vollständig automatisiert ab, sodass keine menschliche Wartung erforderlich ist. Im Vergleich zur dynamischen Partitionierung werden ungenutzte Partitionen oder Partitionen vermieden, die zwar benötigt werden, aber nicht vorhanden sind.

Automatische Partitionierung

Mit Apache Doris 2.1.0 setzen wir den obigen Plan in die Tat um. Während der Datenaufnahme erstellt Doris Datenpartitionen basierend auf den konfigurierten Regeln. Die Doris-Backend-Knoten, die für die Datenverarbeitung und -verteilung verantwortlich sind, versuchen, für jede Datenzeile im DataSink-Operator des Ausführungsplans die entsprechende Partition zu finden. Es werden keine Daten mehr herausgefiltert, die in keine vorhandene Partition passen, oder in einer solchen Situation einen Fehler gemeldet, sondern automatisch Partitionen für alle aufgenommenen Daten generiert.

Automatische Partitionierung nach Bereich

Auto Partition by RANGE bietet eine optimierte Partitionierungslösung basierend auf der Zeitdimension. Es ist in Bezug auf die Parameterkonfiguration flexibler als Dynamic Partition. Die Syntax dafür lautet wie folgt:

 AUTO PARTITION BY RANGE (FUNC_CALL_EXPR) () FUNC_CALL_EXPR ::= DATE_TRUNC ( <partition_column>, '<interval>' )

Die obige <partition_column> ist die Partitionsspalte (also die Spalte, auf der die Partitionierung basiert). <interval> gibt die Partitionseinheit an, also die gewünschte Breite jeder Partition.

Wenn die Partitionsspalte beispielsweise k0 ist und Sie nach Monat partitionieren möchten, lautet die Partitionsanweisung AUTO PARTITION BY RANGE (DATE_TRUNC(k0, 'month')) . Für alle importierten Daten ruft das System DATE_TRUNC(k0, 'month') auf, um den linken Endpunkt der Partition zu berechnen, und dann den rechten Endpunkt durch Hinzufügen eines interval .

Jetzt können wir die automatische Partitionierung auf die Tabelle DAILY_TRADE_VALUE anwenden, die im vorherigen Abschnitt zur dynamischen Partitionierung vorgestellt wurde.

 CREATE TABLE DAILY_TRADE_VALUE ( `TRADE_DATE` DATEV2 NOT NULL COMMENT 'Trade Date', `TRADE_ID` VARCHAR(40) NOT NULL COMMENT 'Trade ID', ...... ) AUTO PARTITION BY RANGE (DATE_TRUNC(`TRADE_DATE`, 'month')) () DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10 PROPERTIES ( ...... );

Nach dem Importieren einiger Daten erhalten wir folgende Partitionen:

 mysql> show partitions from DAILY_TRADE_VALUE; Empty set (0.10 sec) mysql> insert into DAILY_TRADE_VALUE values ('2015-01-01', 1), ('2020-01-01', 2), ('2024-03-05', 10000), ('2024-03-06', 10001); Query OK, 4 rows affected (0.24 sec) {'label':'label_2a7353a3f991400e_ae731988fa2bc568', 'status':'VISIBLE', 'txnId':'85097'} mysql> show partitions from DAILY_TRADE_VALUE; +-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | PartitionId | PartitionName | VisibleVersion | VisibleVersionTime | State | PartitionKey | Range | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation | IsMutable | SyncWithBaseTables | UnsyncTables | +-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | 588395 | p20150101000000 | 2 | 2024-06-01 19:02:40 | NORMAL | TRADE_DATE | [types: [DATEV2]; keys: [2015-01-01]; ..types: [DATEV2]; keys: [2015-02-01]; ) | TRADE_DATE | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 588437 | p20200101000000 | 2 | 2024-06-01 19:02:40 | NORMAL | TRADE_DATE | [types: [DATEV2]; keys: [2020-01-01]; ..types: [DATEV2]; keys: [2020-02-01]; ) | TRADE_DATE | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 588416 | p20240301000000 | 2 | 2024-06-01 19:02:40 | NORMAL | TRADE_DATE | [types: [DATEV2]; keys: [2024-03-01]; ..types: [DATEV2]; keys: [2024-04-01]; ) | TRADE_DATE | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | +-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ 3 rows in set (0.09 sec)

Wie gezeigt, werden für die importierten Daten automatisch Partitionen erstellt. Dabei werden keine Partitionen erstellt, die über den Bereich der vorhandenen Daten hinausgehen.

Automatische Partitionierung nach LIST

Die automatische Partitionierung nach LIST dient zum Sharding von Daten auf der Grundlage nicht zeitbasierter Dimensionen wie region und department . Sie füllt diese Lücke für die dynamische Partitionierung, die die Datenpartitionierung nach LIST nicht unterstützt.

 AUTO PARTITION BY LIST (`partition_col`) ()

Dies ist ein Beispiel für die automatische Partitionierung nach LIST unter Verwendung der city als Partitionsspalte:

 mysql> CREATE TABLE `str_table` ( -> `city` VARCHAR NOT NULL, -> ...... -> ) -> DUPLICATE KEY(`city`) -> AUTO PARTITION BY LIST (`city`) -> () -> DISTRIBUTED BY HASH(`city`) BUCKETS 10 -> PROPERTIES ( -> ...... -> ); Query OK, 0 rows affected (0.09 sec) mysql> insert into str_table values ("Denver"), ("Boston"), ("Los_Angeles"); Query OK, 3 rows affected (0.25 sec) mysql> show partitions from str_table; +-------------+-----------------+----------------+---------------------+--------+--------------+-------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | PartitionId | PartitionName | VisibleVersion | VisibleVersionTime | State | PartitionKey | Range | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation | IsMutable | SyncWithBaseTables | UnsyncTables | +-------------+-----------------+----------------+---------------------+--------+--------------+-------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | 589685 | pDenver7 | 2 | 2024-06-01 20:12:37 | NORMAL | city | [types: [VARCHAR]; keys: [Denver]; ] | city | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 589643 | pLos5fAngeles11 | 2 | 2024-06-01 20:12:37 | NORMAL | city | [types: [VARCHAR]; keys: [Los_Angeles]; ] | city | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 589664 | pBoston8 | 2 | 2024-06-01 20:12:37 | NORMAL | city | [types: [VARCHAR]; keys: [Boston]; ] | city | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | +-------------+-----------------+----------------+---------------------+--------+--------------+-------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ 3 rows in set (0.10 sec)

Nach dem Einfügen der Daten für die Städte Denver, Boston und Los Angeles erstellte das System automatisch entsprechende Partitionen basierend auf den Städtenamen. Bisher konnte diese Art der benutzerdefinierten Partitionierung nur durch manuelle DDL-Anweisungen erreicht werden. So vereinfacht Auto Partition by LIST die Datenbankwartung.

Tipps & Hinweise

Manuelles Anpassen historischer Partitionen

Für Tabellen, die sowohl Echtzeitdaten als auch gelegentliche historische Updates erhalten, empfehlen wir zwei Optionen, da Auto Partition historische Partitionen nicht automatisch zurückfordert:

Verwenden Sie die automatische Partitionierung, die automatisch Partitionen für gelegentliche Aktualisierungen historischer Daten erstellt.
Verwenden Sie die automatische Partitionierung und erstellen Sie manuell eine LESS THAN Partition, um die historischen Aktualisierungen unterzubringen. Dies ermöglicht eine klarere Trennung von historischen und Echtzeitdaten und vereinfacht die Datenverwaltung.

 mysql> CREATE TABLE DAILY_TRADE_VALUE -> ( -> `TRADE_DATE` DATEV2 NOT NULL COMMENT 'Trade Date', -> `TRADE_ID` VARCHAR(40) NOT NULL COMMENT 'Trade ID' -> ) -> AUTO PARTITION BY RANGE (DATE_TRUNC(`TRADE_DATE`, 'DAY')) -> ( -> PARTITION `pHistory` VALUES LESS THAN ("2024-01-01") -> ) -> DISTRIBUTED BY HASH(`TRADE_DATE`) BUCKETS 10 -> PROPERTIES -> ( -> "replication_num" = "1" -> ); Query OK, 0 rows affected (0.11 sec) mysql> insert into DAILY_TRADE_VALUE values ('2015-01-01', 1), ('2020-01-01', 2), ('2024-03-05', 10000), ('2024-03-06', 10001); Query OK, 4 rows affected (0.25 sec) {'label':'label_96dc3d20c6974f4a_946bc1a674d24733', 'status':'VISIBLE', 'txnId':'85092'} mysql> show partitions from DAILY_TRADE_VALUE; +-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | PartitionId | PartitionName | VisibleVersion | VisibleVersionTime | State | PartitionKey | Range | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation | IsMutable | SyncWithBaseTables | UnsyncTables | +-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | 577871 | pHistory | 2 | 2024-06-01 08:53:49 | NORMAL | TRADE_DATE | [types: [DATEV2]; keys: [0000-01-01]; ..types: [DATEV2]; keys: [2024-01-01]; ) | TRADE_DATE | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 577940 | p20240305000000 | 2 | 2024-06-01 08:53:49 | NORMAL | TRADE_DATE | [types: [DATEV2]; keys: [2024-03-05]; ..types: [DATEV2]; keys: [2024-03-06]; ) | TRADE_DATE | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 577919 | p20240306000000 | 2 | 2024-06-01 08:53:49 | NORMAL | TRADE_DATE | [types: [DATEV2]; keys: [2024-03-06]; ..types: [DATEV2]; keys: [2024-03-07]; ) | TRADE_DATE | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | +-------------+-----------------+----------------+---------------------+--------+--------------+--------------------------------------------------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ 3 rows in set (0.10 sec)

NULL-Partition

Mit Auto Partition by LIST unterstützt Doris das Speichern von NULL-Werten in NULL-Partitionen. Beispiel:

 mysql> CREATE TABLE list_nullable -> ( -> `str` varchar NULL -> ) -> AUTO PARTITION BY LIST (`str`) -> () -> DISTRIBUTED BY HASH(`str`) BUCKETS auto -> PROPERTIES -> ( -> "replication_num" = "1" -> ); Query OK, 0 rows affected (0.10 sec) mysql> insert into list_nullable values ('123'), (''), (NULL); Query OK, 3 rows affected (0.24 sec) {'label':'label_f5489769c2f04f0d_bfb65510f9737fff', 'status':'VISIBLE', 'txnId':'85089'} mysql> show partitions from list_nullable; +-------------+---------------+----------------+---------------------+--------+--------------+------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | PartitionId | PartitionName | VisibleVersion | VisibleVersionTime | State | PartitionKey | Range | DistributionKey | Buckets | ReplicationNum | StorageMedium | CooldownTime | RemoteStoragePolicy | LastConsistencyCheckTime | DataSize | IsInMemory | ReplicaAllocation | IsMutable | SyncWithBaseTables | UnsyncTables | +-------------+---------------+----------------+---------------------+--------+--------------+------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ | 577297 | pX | 2 | 2024-06-01 08:19:21 | NORMAL | str | [types: [VARCHAR]; keys: [NULL]; ] | str | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 577276 | p0 | 2 | 2024-06-01 08:19:21 | NORMAL | str | [types: [VARCHAR]; keys: []; ] | str | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | | 577255 | p1233 | 2 | 2024-06-01 08:19:21 | NORMAL | str | [types: [VARCHAR]; keys: [123]; ] | str | 10 | 1 | HDD | 9999-12-31 23:59:59 | | NULL | 0.000 | false | tag.location.default: 1 | true | true | NULL | +-------------+---------------+----------------+---------------------+--------+--------------+------------------------------------+-----------------+---------+----------------+---------------+---------------------+---------------------+--------------------------+----------+------------+-------------------------+-----------+--------------------+--------------+ 3 rows in set (0.11 sec)

Auto Partition by RANGE unterstützt jedoch keine NULL-Partitionen, da die NULL-Werte in der kleinsten LESS THAN Partition gespeichert werden und es unmöglich ist, den entsprechenden Bereich dafür zuverlässig zu bestimmen. Wenn Auto Partition eine NULL-Partition mit einem Bereich von (-INFINITY, MIN_VALUE) erstellen würde, bestünde das Risiko, dass diese Partition in der Produktion versehentlich gelöscht wird, da die MIN_VALUE-Grenze möglicherweise nicht genau die beabsichtigte Geschäftslogik darstellt.

Zusammenfassung

Auto Partition deckt die meisten Anwendungsfälle von Dynamic Partition ab und bietet gleichzeitig den Vorteil der Definition von Partitionsregeln im Voraus. Sobald die Regeln definiert sind, wird der Großteil der Partitionserstellungsarbeit automatisch von Doris und nicht von einem DBA übernommen.

Bevor Sie Auto Partition verwenden, müssen Sie die entsprechenden Einschränkungen verstehen:

Auto Partition by LIST unterstützt Partitionierung basierend auf mehreren Spalten , aber jede automatisch erstellte Partition enthält nur einen einzigen Wert und der Partitionsname darf nicht länger als 50 Zeichen sein. Beachten Sie, dass die Partitionsnamen bestimmten Namenskonventionen folgen, die besondere Auswirkungen auf die Metadatenverwaltung haben. Das bedeutet, dass dem Benutzer nicht der gesamte 50-Zeichen-Platz zur Verfügung steht.
Die automatische Partitionierung nach RANGE unterstützt nur eine einzelne Partitionsspalte , die vom Typ DATE oder DATETIME sein muss.
Die automatische Partitionierung nach LIST unterstützt NULLABLE -Partitionsspalten und das Einfügen von NULL-Werten. Die automatische Partitionierung nach RANGE unterstützt keine NULLABLE-Partitionsspalten.
Es wird nicht empfohlen, Auto Partition in Verbindung mit Dynamic Partition nach Apache Doris 2.1.3 zu verwenden.

Leistungsvergleich

Die wichtigsten funktionalen Unterschiede zwischen Auto Partition und Dynamic Partition liegen im Erstellen und Löschen von Partitionen, in den unterstützten Partitionstypen und in deren Auswirkungen auf die Importleistung.

Dynamic Partition verwendet feste Threads, um Partitionen regelmäßig zu erstellen und freizugeben. Es unterstützt nur die Partitionierung nach BEREICH. Im Gegensatz dazu unterstützt Auto Partition sowohl die Partitionierung nach BEREICH als auch nach LISTE. Es erstellt automatisch Partitionen bei Bedarf basierend auf bestimmten Regeln während der Datenaufnahme und bietet so ein höheres Maß an Automatisierung und Flexibilität.

Die dynamische Partitionierung verlangsamt die Datenübertragungsgeschwindigkeit nicht, während die automatische Partitionierung einen gewissen Zeitaufwand verursacht, da sie zunächst nach vorhandenen Partitionen sucht und dann bei Bedarf neue erstellt. Wir präsentieren die Ergebnisse der Leistungstests.

Automatische Partitionierung: Aufnahme-Workflow

In diesem Teil geht es darum, wie die Datenaufnahme mit dem Auto Partition-Mechanismus implementiert wird. Als Beispiel verwenden wir Stream Load . Wenn Doris einen Datenimport initiiert, übernimmt einer der Doris Backend-Knoten die Rolle des Koordinators. Er ist für die anfängliche Datenverarbeitung verantwortlich und sendet die Daten dann zur Ausführung an die entsprechenden BE-Knoten, die sogenannten Executors.

Im letzten Datasink-Knoten der Ausführungspipeline des Koordinators müssen die Daten an die richtigen Partitionen, Buckets und Doris-Backend-Knotenstandorte weitergeleitet werden, bevor sie erfolgreich übertragen und gespeichert werden können.

Um diesen Datentransfer zu ermöglichen, richten die Koordinator- und Executor-Knoten Kommunikationskanäle ein:

Das sendende Ende wird als Knotenkanal bezeichnet.
Das Empfangsende wird als Tablets-Kanal bezeichnet.

So kommt Auto Partition beim Ermitteln der richtigen Partitionen für die Daten ins Spiel:

Wenn eine Tabelle nicht die erforderliche Partition hatte, war das Verhalten in Doris ohne Auto Partition bisher so, dass die BE-Knoten Fehler sammelten, bis ein DATA_QUALITY_ERROR gemeldet wurde. Jetzt, wo Auto Partition aktiviert ist, wird eine Anfrage an das Doris-Frontend gesendet, um die erforderliche Partition im laufenden Betrieb zu erstellen. Nachdem die Transaktion zur Partitionserstellung abgeschlossen ist, antwortet das Doris-Frontend dem Koordinator, der dann die entsprechenden Kommunikationskanäle (Knotenkanal und Tablet-Kanal) öffnet, um den Datenaufnahmeprozess fortzusetzen. Dies ist ein nahtloses Erlebnis für Benutzer.

In einer realen Clusterumgebung kann die Zeit, die der Koordinator damit verbringt, darauf zu warten, dass das Doris-Frontend die Partitionserstellung abschließt, zu einem hohen Mehraufwand führen. Dies liegt an der inhärenten Latenz von Thrift-RPC-Aufrufen sowie an Sperrkonflikten am Frontend unter Hochlastbedingungen.

Um die Effizienz der Datenaufnahme in Auto Partition zu verbessern, hat Doris Batching implementiert, um die Anzahl der an die FE gerichteten RPC-Aufrufe erheblich zu reduzieren. Dies führt zu einer deutlichen Leistungssteigerung bei Datenschreibvorgängen.

Beachten Sie, dass die neue Partition sofort sichtbar wird, wenn der FE-Master die Partitionserstellungstransaktion abschließt. Wenn der Importvorgang jedoch endgültig fehlschlägt oder abgebrochen wird, werden die erstellten Partitionen nicht automatisch wiederhergestellt.

Leistung der automatischen Partitionierung

Wir haben die Leistung und Stabilität von Auto Partition in Doris anhand verschiedener Anwendungsfälle getestet:

Fall 1 : 1 Frontend + 3 Backends; 6 zufällig generierte Datensätze mit jeweils 100 Millionen Zeilen und 2.000 Partitionen; die 6 Datensätze wurden gleichzeitig in 6 Tabellen aufgenommen

Ziel : Bewerten Sie die Leistung der automatischen Partitionierung unter hohem Druck und prüfen Sie, ob es zu Leistungseinbußen kommt.
Ergebnisse : Durch die automatische Partitionierung entsteht ein durchschnittlicher Leistungsverlust von weniger als 5 % , wobei alle Importtransaktionen stabil laufen.

Fall 2 : 1 Frontend + 3 Backends; Aufnahme von 100 Zeilen pro Sekunde von Flink durch Routine Load; Test mit 1, 10 bzw. 20 gleichzeitigen Transaktionen (Tabellen).

Ziel : Identifizieren Sie alle potenziellen Datenrückstandsprobleme, die bei der automatischen Partitionierung unter verschiedenen Parallelitätsstufen auftreten können.
Ergebnisse : Mit oder ohne aktivierter automatischer Partitionierung war die Datenaufnahme auf allen getesteten Parallelitätsebenen ohne Probleme mit Gegendruck erfolgreich, selbst bei 20 gleichzeitigen Transaktionen, als die CPU-Auslastung fast 100 % erreichte.

Um die Ergebnisse dieser Tests zusammenzufassen: Die Auswirkungen der Aktivierung der automatischen Partitionierung auf die Leistung der Datenübertragung sind minimal.

Fazit und Zukunftspläne

Auto Partition vereinfacht seit Apache Doris 2.1.0 die DDL- und Partitionsverwaltung. Es ist bei der Verarbeitung großer Datenmengen nützlich und erleichtert Benutzern die Migration von anderen Datenbanksystemen zu Apache Doris.

Darüber hinaus sind wir bestrebt, die Funktionen von Auto Partition zu erweitern, um komplexere Datentypen zu unterstützen.

Pläne für die automatische Partitionierung nach RANGE: