Die systematischste Beherrschung der Echtzeitextraktion von Oracle-Daten aus der Flink CDC-Serie (Mining- und Tuning-Praktiken)

Dieser Artikel bietet Ihnen Echtzeit-Datenerfassung und Leistungsoptimierung von Oracle und teilt einige wichtige Details während des Testprozesses. Ich hoffe, dass er für alle hilfreich sein wird.

Flink CDC hat am 15. November 2021 die neueste Version 2.1 veröffentlicht, die durch die Einführung integrierter Debezium-Komponenten Unterstützung für Oracle hinzufügt. Der Autor hat diese Version sofort zur Testnutzung heruntergeladen und die Echtzeit-Datenerfassung und Leistungsoptimierung von Oracle erfolgreich implementiert. Jetzt werde ich einige wichtige Details während des Testprozesses mitteilen.

Testumgebung:

Oracle: 11.2.0.4.0 (RAC-Bereitstellung)

Flink: 1.13.1

Hadoop: 3.2.1

Bereitgestellt und verwendet über Flink on Yarn

1. Verbindung konnte nicht hergestellt werden zur Datenbank

Geben Sie gemäß der offiziellen Dokumentation die folgende Anweisung in die Flink SQL-CLI ein:

create table TEST (A string)
WITH ('connector'='oracle-cdc',
    'hostname'='10.230.179.125',
    'port'='1521',
    'username'='myname',
    'password'='***',
    'database-name'='MY_SERVICE_NAME',
    'schema-name'='MY_SCHEMA',
    'table-name'='TEST' );

Versuchen Sie dann, mithilfe von „select *“ von TEST zu beobachten, dass Sie keine normale Verbindung zu Oracle herstellen können. Der Fehler lautet wie folgt:

[ERROR] Could not execute SQL statement. Reason:
oracle.net.ns.NetException: Listener refused the connection with the following error:
ORA-12505, TNS:listener does not currently know of SID given in connect descriptor

Der Fehlermeldung nach zu urteilen, könnte es daran liegen, dass Flink CDC fälschlicherweise den in den Verbindungsinformationen angegebenen MY_SERVICE_NAME (den Dienstnamen von Oracle) als SID verwechselt hat. Also habe ich versucht, den Quellcode von Oracle Connector im Zusammenhang mit Flink CDC zu lesen, und festgestellt, dass in com.ververica.cdc.connectors.oracle.OracleValidator der Code für die Oracle-Verbindung wie folgt lautet:

public static Connection openConnection(Properties properties) throws SQLException {
    DriverManager.registerDriver(new oracle.jdbc.OracleDriver());
    String hostname = properties.getProperty("database.hostname");
    String port = properties.getProperty("database.port");
    String dbname = properties.getProperty("database.dbname");
    String userName = properties.getProperty("database.user");
    String userpwd = properties.getProperty("database.password");
    return DriverManager.getConnection(
            "jdbc:oracle:thin:@" + hostname + ":" + port + ":" + dbname, userName, userpwd);
}

Wie aus dem ersichtlich ist Oben gibt es in der aktuellen Version von Flink CDC keinen Unterschied zwischen den Verbindungsmethoden von SID und Dienstnamen, aber die Verbindungsmethode von SID wird direkt in den Code geschrieben (d. h. Port und Datenbankname werden durch „:“ getrennt). .

Ab Oracle 8i hat Oracle das Konzept des Servicenamens zur Unterstützung der Cluster-Bereitstellung (RAC) von Datenbanken eingeführt. Ein Servicename kann als logisches Konzept einer Datenbank verwendet werden, um Verbindungen zu verschiedenen SID-Instanzen der Datenbank zu vereinheitlichen. Auf dieser Grundlage können Sie die folgenden zwei Methoden in Betracht ziehen:

Ersetzen Sie in der Anweisung „Create Table“ von Flink CDC „database-name“ durch „Service Name“ durch eine der SIDs. Diese Methode kann das Verbindungsproblem lösen, kann sich jedoch nicht an das reale Szenario der Mainstream-Oracle-Clusterbereitstellung anpassen.

Ändern Sie den Quellcode. Insbesondere können Sie die Methode com.ververica.cdc.connectors.oracle.OracleValidator im neuen Projekt umschreiben und in die Verbindungsmethode des Dienstnamens ändern (d. h. verwenden Sie „/“, um den Port und den Datenbanknamen zu trennen). :

"jdbc :oracle:thin:@" + hostname + ":" + port + "/" + dbname, userName, userpwd);

Der Autor verwendet die zweite Methode, um eine normale Verbindung zur Datenbank herzustellen und gleichzeitig den Zugriff beizubehalten zur Oracle-Dienstnutzung des Namensattributs.

2. Die Oracle-Tabelle kann nicht gefunden werden

Befolgen Sie die obigen Schritte und beobachten Sie erneut, dass die Daten nicht normal abgerufen werden können:

[ERROR] Could not execute SQL statement. Reason:
io.debezium.DebeziumException: Supplemental logging not configured for table MY_SERVICE_NAME.MY_SCHEMA.test.  Use command: ALTER TABLE MY_SCHEMA.test ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS

Beobachtet, dass die im Fehlerprotokoll erwähnte Tabelle MY_SERVICE_NAME .MY_SCHEMA.test ist. Warum sind der Datenbankname und der Schemaname in Großbuchstaben geschrieben, der Tabellenname jedoch in Kleinbuchstaben?

Beachten Sie, dass dieser Fehler vom io.debezium-Paket gemeldet wird (aus der pom.xml-Datei von Flink CDC, wir verwenden derzeit die Debezium-Version 1.5.4), können wir das in io sehen .debezium.relational In .Tables gibt es den folgenden Code:

private TableId toLowerCaseIfNeeded(TableId tableId) {
    return tableIdCaseInsensitive ? tableId.toLowercase() : tableId;
}

Es ist ersichtlich, dass die Entwickler von Debezium „Groß- und Kleinschreibung“ einheitlich als „Der Tabellenname muss in Kleinbuchstaben konvertiert werden“ definieren. Dies gilt für PostgreSQL, MySQL usw., die von Debezium unterstützt werden. Für die Oracle-Datenbank bedeutet „Unabhängigkeit der Groß-/Kleinschreibung“ jedoch, dass der Tabellenname beim Speichern interner Metainformationen in Großbuchstaben umgewandelt werden muss Beim Versuch, einen Tabellennamen in Kleinbuchstaben zu lesen, ist ein Fehler aufgetreten.

Da Debezium dieses Problem erst mit der neuesten stabilen Version 1.7.1 und der neuesten Entwicklungsversion 1.8.0 behoben hat, können wir dieses Problem mit den folgenden zwei Methoden umgehen:

Wenn Sie Oracle „ohne Berücksichtigung der Groß- und Kleinschreibung“ verwenden müssen Mit der Funktion „Sensibel“ können Sie den Quellcode direkt ändern und das oben genannte toLowercase in toUppercase ändern (dies ist auch die vom Autor gewählte Methode).

Wenn Sie den Quellcode nicht ändern möchten und nicht Oracles „ Wenn Sie die Funktion „Case insensitive“ verwenden, können Sie die create-Anweisung im Add „debezium.database.tablename.case.insensitive“=‘false‘ verwenden, wie im folgenden Beispiel:

create table TEST (A string)
WITH ('connector'='oracle-cdc',
    'hostname'='10.230.179.125',
    'port'='1521',
    'username'='myname',
    'password'='***',
    'database-name'='MY_SERVICE_NAME',
'schema-name'='MY_SCHEMA',
'table-name'='TEST',
'debezium.database.tablename.case.insensitive'='false' );

Der Nachteil dieser Methode besteht darin, dass sie Oracles „“ verliert. „case-insensitive“-Funktion, die in „table-name“ verwendet werden muss. Geben Sie Tabellennamen explizit in Großbuchstaben an.

Es ist zu beachten, dass Debezium den Parameter „database.tablename.case.insensitive“ derzeit nur für Oracle 11g standardmäßig auf „true“ und für andere Oracle-Versionen standardmäßig auf „false“ setzt. Wenn der Leser also nicht die Oracle 11g-Version verwendet, muss dieser Parameter nicht geändert werden, der Tabellenname in Großbuchstaben muss jedoch weiterhin explizit angegeben werden.

3. Die Datenverzögerung ist großDie Datenverzögerung ist groß, manchmal dauert es 3-5 Minuten, um Datenänderungen zu erfassen. Für dieses Problem wurde in den Flink CDC-FAQ eine klare Lösung angegeben: Fügen Sie der create-Anweisung die folgenden zwei Konfigurationselemente hinzu:

'debezium.log.mining.strategy'='online_catalog',
'debezium.log.mining.continuous.mine'='true'

那么为什么要这样做呢？我们依然可以通过分析源码和日志，结合 Oracle Logminer 的工作原理来加深对工具的理解。

对 Logminer 的抽取工作，主要在 Debezium 的 io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource 中 execute 方法进行。为节约篇幅，本文不列出实际的源码，仅提炼出关键过程绘于下面的流程图，有兴趣的读者可以对照该流程图，结合实际源码进行分析：

采用 redo_log_catalog 的方式，可以监控数据表的 DDL 信息，且由于 archive logs 被永久保存到磁盘上，可以在数据库宕机后依然正常获取到宕机前的所有 DDL 和 DML 操作。但由于涉及到比 online catalog 更多的信息监控，以及由此带来的频繁的日志切换和日志转储操作，其代价也是惊人的。

根据笔者实际测试情况，如果 debezium.log.mining.strategy 为默认配置 redo_log_catalog，则不仅需要多执行第 ① 步操作 (该操作耗时约为半分钟到 1 分钟之间)，在第 ④ 步，根据 archived logs 的数据量，耗时也会在 1 分钟到 4 分钟之间浮动；在第 ⑤ 步，实际查询 V$LOGMNR_CONTENTS 视图也常常需要十几秒才能完成。

此外，由于 archive logs 在实际系统中增长速度较快，因此在实际使用中，常会配合进行定期删除或转储过期日志的操作。由于上述第 ④ 步的耗时较长，笔者观察到在第 ④ 步执行的过程中，在一定概率下会发生第 ② 步加入的a rchive logs 已过期而被删除转储的情况，于是在第 ⑤ 步查询的时候，会由于找不到第 ② 步加入的日志，而报下面的错误：

ORA-00308: cannot open archive log '/path/to/archive/log/...'
ORA-27037: unable to obtain file status

一般来说，Flink CDC 所需要监控的表，特别是对于业务系统有重大意义的表，一般不会进行 DDL 操作，仅需要捕捉 DML 操作即可，且对于数据库宕机等极特殊情况，也可使用在数据库恢复后进行全量数据更新的方式保障数据的一致性。因而，online_catalog 的方式足以满足我们的需要。

另外，无论使用 online_catalog，还是默认的 redo_log_catalog，都会存在第 ② 步找到的日志和第 ⑤ 步实际需要的日志不同步的问题，因此，加入 'debezium.log.mining.continuous.mine'='true' 参数，将实时搜集日志的工作交给 Oracle 自动完成，即可规避这一问题。

笔者按照这两个参数配置后，数据延迟一般可以从数分钟降至 5 秒钟左右。

四、调节参数继续降低数据延迟

上述流程图的第 ③ 步和第 ⑦ 步，提到了根据配置项来确定 LogMiner 监控时序范围，以及确定休眠时间。下面对该过程进行进一步分析，并对单个表的进一步调优给出一般性的方法论。

通过观察 io.debezium.connector.oracle.logminer.LogMinerHelper 类中的 getEndScn 方法，可了解到 debezium 对监控时序范围和休眠时间的调节原理。为便于读者理解，将该方法用流程图说明如下：

从上述的流程图中可以看出，debezium 给出 log.mining.batch.size.* 和 log.mining.sleep.time.* 两组参数，就是为了让每一次 logMiner 运行的步长能够尽可能和数据库自身 SCN 增加的步长一致。由此可见：

log.mining.batch.size.* 和 log.mining.sleep.time.* 参数的设定，和数据库整体的表现有关，和单个表的数据变化情况无关；

log.mining.batch.size.default 不仅仅是监控时序范围的起始值，还是监控时序范围变化的阈值。所以如果要实现更灵活的监控时序范围调整，可考虑适当减小该参数；

由于每一次确定监控时序范围时，都会根据 topScn 和 currentScn 的大小来调整 sleepTime，所以为了实现休眠时间更灵活的调整，可考虑适当增大 log.mining.sleep.time.increment.ms；

log.mining.batch.size.max 不能过小，否则会有监控时序范围永远无法追上数据库当前 SCN 的风险。为此，debezium 在 io.debezium.connector.oracle.OracleStreamingChangeEventSourceMetrics 中存在以下逻辑：

if (currentBatchSize == batchSizeMax) {
    LOGGER.info("LogMiner is now using the maximum batch size {}. This could be indicative of large SCN gaps", currentBatchSize);
}

如果当前的监控时序范围达到了 log.mining.batch.size.max，那么 debezium 会在日志中给出如上提示。在实际应用中，观察 Flink CDC 产生的 log 是否包含该提示，便可得知 log.mining.batch.size.max 的值是否合理。

五、Debezium Oracle Connector 的隐藏参数

Tatsächlich haben wir aus dem oben Gesagten etwas über zwei versteckte Parameter erfahren: debezium.database.tablename.case.insensitive (siehe den zweiten Abschnitt) und debezium.log.mining.continuous.mine (siehe den dritten Abschnitt). werden in der offiziellen Dokumentation von Debezium nicht wirklich beschrieben, können aber tatsächlich verwendet werden. Durch die Analyse des Quellcodes werden nun alle versteckten Parameter von Debezium Oracle Connector angegeben und ihre Beschreibungen lauten wie folgt:

Der Autor glaubt, dass zusätzlich zu den beiden Parametern, die wir oben verwendet haben, log.mining.history Es lohnt sich auch, auf den Parameter .recorder.class zu achten. Da dieser Parameter derzeit standardmäßig auf io.debezium.connector.oracle.logminer.NeverHistoryRecorder eingestellt ist, was eine leere Klasse ist, passen wir bei der Analyse des Flink CDC-Verhaltens eine Klasse an, die die Schnittstelle io.debezium.connector.oracle.logminer.HistoryRecorder implementiert. , die eine personalisierte Überwachung des Flink CDC-Verhaltens realisieren kann, ohne den Quellcode zu ändern.

Empfohlenes Tutorial: „Oracle Tutorial“

Das obige ist der detaillierte Inhalt vonDie systematischste Beherrschung der Echtzeitextraktion von Oracle-Daten aus der Flink CDC-Serie (Mining- und Tuning-Praktiken). Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!