书接前文，在上一节中，我们将Mysql CDC数据实时接入了Impala Kudu表。完整的数据流向如下图所示：
图中MYSQL和SQLSERVER数据库的CDC数据采集，在本系列中已经讲解，本节给大家分享下PostgreSQL数据库的CDC数据采集和DDL监控。在正式进行实战操作之前，请先学习Mysql连接器的使用方法，因为很多的配置都是相同的。本文主要解释下PostgreSQL连接器特有的内容。再次提醒：要理解这套实时数仓架构的思路和落地方法，请通读本系列文章。
我们首先了解下PostgreSQL的Debezium连接器。
0. PostgreSQL的Debezium连接器
本部分内容译自官网，使用任何数据库的Debezium连接器之前，建议大家先阅读下相应连接器的官网说明，可以事半功倍。
Debezium PostgreSQL连接器捕获PostgreSQL数据库模式中的行级更改。有关与连接器兼容的PostgreSQL版本的信息，请参阅Debezium发布概述。
当第一次连接到PostgreSQL服务器或集群时，连接器获取所有模式的一致快照。快照完成后，连接器将持续捕获插入、更新和删除数据库内容以及提交到PostgreSQL数据库的行级更改。连接器生成数据更改事件记录，并将它们传输到Kafka主题。对于每个表，默认的行为是连接器将所有生成的事件流传输到该表的单独Kafka主题。应用程序和服务消费来自该主题的数据更改事件记录。
PostgreSQL的逻辑解码( logical decoding)特性是在9.4版引入的。它是一种机制，允许提取提交到事务日志的更改，并在输出插件的帮助下以用户友好的方式处理这些更改。输出插件允许客户端消费更改。
PostgreSQL连接器包含两个主要部分，它们一起工作来读取和处理数据库更改:
-
逻辑解码输出插件。需要安装选择使用的输出插件。在运行PostgreSQL服务器之前，必须配置使用所选输出插件的复制槽(replication slot)。该插件可以是以下其中之一:
-
decoderbufs基于Protobuf，由Debezium社区维护。
-
wal2json基于JSON，由wal2json社区维护(已弃用，计划在Debezium 2.0中删除)。
-
pgoutput是PostgreSQL 10+中的标准逻辑解码输出插件。它由PostgreSQL社区维护，并由PostgreSQL自己用于逻辑复制。该插件始终存在，因此不需要安装其他库。Debezium连接器将原始复制事件流直接解释为更改事件。注：本文示例使用这种方式。
-
-
读取所选逻辑解码输出插件产生的更改的Java代码(实际的Kafka Connect连接器)。它使用PostgreSQL的流复制协议，通过PostgreSQL JDBC驱动程序实现。
连接器为捕获的每一个行级插入、更新和删除操作生成一个更改事件，并为单独Kafka主题中的每个表发送更改事件记录。客户端应用程序读取与感兴趣的数据库表对应的Kafka主题，并可以对从这些主题接收到的每一个行级事件做出反应。
PostgreSQL通常在一段时间后清除预写日志(write-ahead log, WAL)段。这意味着连接器没有对数据库所做的所有更改的完整历史。因此，当PostgreSQL连接器第一次连接到一个特定的PostgreSQL数据库时，它首先对每个数据库模式执行一致的快照。在连接器完成快照之后，它将继续从创建快照的确切点开始流化更改。通过这种方式，连接器从所有数据的一致视图开始，并且不会忽略在创建快照时所做的任何更改。
这个连接器可容忍故障。当连接器读取更改并产生事件时，它会记录每个事件的WAL位置。如果连接器因任何原因(包括通信故障、网络问题或崩溃)而停止，那么在重新启动时，连接器将继续在它最后停止的位置读取WAL。这包括快照。如果连接器在快照期间停止，则在重新启动时连接器将开始一个新的快照。
重要提示
该连接器依赖PostgreSQL逻辑解码特性，该特性有以下局限性:
逻辑解码不支持DDL更改。这意味着连接器无法向消费者报告DDL更改事件。---本文提供了变通解决方法。
逻辑解码复制槽位仅在
primary服务器上支持。当存在一个PostgreSQL服务器集群时，连接器只能在活动的primary服务器上运行。它不能在hot或者warmstandby replicas上运行。如果primary服务器发生故障或降级，连接器将停止。在primary服务器恢复后，可以重新启动连接器。如果其他PostgreSQL服务器已升级为primary服务器，请在重新启动连接器之前调整连接器配置。
Debezium目前只支持UTF-8字符编码的数据库。使用单字节字符编码，不能正确处理包含扩展ASCII码字符的字符串。
对Debezium连接器的机制有个大概了解，现在可以动手配置了。
1. 前置条件
-
安装Postgresql数据库，本文使用pg12版本，安装过程参考官方文档，这里提供一个笔者使用的安装脚本：
sudo yum install -y https://download.postgresql.org/pub/repos/yum/reporpms/EL-7-x86_64/pgdg-redhat-repo-latest.noarch.rpmsudo yum install -y postgresql12-serversudo /usr/pgsql-12/bin/postgresql-12-setup initdbsudo systemctl enable postgresql-12sudo systemctl start postgresql-12
-
在postgresql.conf文件中指定以下内容：
# REPLICATION 重启生效wal_level = logical
-
配置用户权限。本文示例使用默认账户postgres，有最高权限。在生产环境中，可以为Debezium数据采集创建单独的用户并授权，相关内容参考官网：
https://debezium.io/documentation/reference/1.9/connectors/postgresql.html#postgresql-permissions-
配置PG，使之可以远程连接，建议由DBA操作。修改文件?
pg_hba.conf?：
#重启生效
host????all?????????????all?????????????0.0.0.0/0????????????md5-
下载Debezium pg连接器，解压到kafka connect配置文件指定目录中，本示例指定的目录为?connect_lib ，见下图：
2. 配置并创建连接器
{"name": "pg-test-connector","config": {"connector.class": "io.debezium.connector.postgresql.PostgresConnector","database.hostname": "xxx.xxx.xxx.xxx","database.port": "5432","database.user": "postgres","database.password": "postgres","database.dbname" : "postgres","database.server.name": "pg_test","name": "pg-test-connector","database.history.kafka.bootstrap.servers": "testcdh6:9092,testcdh7:9092,testcdh8:9092","table.include.list": "public.t_person" ,????"tombstones.on.delete":?"false",#同mysql连接器配置????"plugin.name":"pgoutput",#使用pg自带插件????"decimal.handling.mode":?"double"?#同mysql连接器配置}}
创建连接器步骤同Mysql连接器。在表t_person中增改删数据，查看topic中的消息格式：
{"before":{"id":9,"name":null,"age":null},"after":null,"source":{"version":"1.9.0.Final","connector":"postgresql","name":"pg_test","ts_ms":1663834544766,"snapshot":"false","db":"postgres","sequence":"[\"24283888\",\"24283944\"]","schema":"public","table":"t_person","txId":508,"lsn":24283944,"xmin":null},"op":"d","ts_ms":1663834544776,"transaction":null}
可以看出，消息格式和MySQL连接器的消息格式基本一致。上面是一个删除操作的消息格式，before对应的json对象中，只有主键id有值，其他字段值都是?null?。这是因为pg默认只会复制主键值，通过修改表的复制模式可以解决这个问题：???????
#解决除主键外 其他字段都是null的问题ALTER TABLE public.t_person REPLICA IDENTITY FULL
再次测试，消息内容如下：???????
{"before":{"id":8,"name":"7","age":90},"after":null,"source":{"version":"1.9.0.Final","connector":"postgresql","name":"pg_test","ts_ms":1663834899886,"snapshot":"false","db":"postgres","sequence":"[\"24304672\",\"24304672\"]","schema":"public","table":"t_person","txId":512,"lsn":24304672,"xmin":null},"op":"d","ts_ms":1663834900200,"transaction":null}
下一步就是将这些增删改消息接入impala kudu表，具体NIFI任务的创建过程，见前一节的详细描述。
3. PG的DDL监控
PG连接器无法监控DDL的变更。在实际项目中，DDL变更的监控可能非常重要，如何解决这个问题呢？这里给大家提供一种解决方式，供参考：创建触发器，监控DDL变更，将变更信息写入一张表中，监控这张表即可，如可以将此表配置到PG连接器中。
具体方式如下：???????
#创建表create table public.dfdms_ddl_audit(c_key bigserial primary key,c_time timestamp, -- DDL发生时间c_user varchar(64), -- DDL发生用户: current_userc_txn varchar(16), -- DDL发生事务: current transactionc_tag varchar(24), -- Either 'CREATE TABLE' or 'ALTER TABLE' or 'DROP TABLE'c_oid integer, -- 备用字段 - TG_OBJECTIDc_name varchar(64), -- 备用字段 - TG_OBJECTNAMEc_schema varchar(64), -- 备用字段 - TG_SCHEMANAME. For now - holds current_schemac_ddlqry text -- 与当前DDL事件关联的DDL查询)#创建函数CREATE OR REPLACE FUNCTION public.dfdms_intercept_ddl()RETURNS event_triggerLANGUAGE plpgsqlSECURITY DEFINERAS $$declare _qry text;BEGINif (tg_tag='CREATE TABLE' or tg_tag='ALTER TABLE' or tg_tag='DROP TABLE') thenSELECT current_query() into _qry;insert into public.dfdms_ddl_auditvalues(default,current_timestamp,current_user,cast(TXID_CURRENT()as varchar(16)),tg_tag,0,'',current_schema,_qry);end if;END;$$;#创建触发器CREATE EVENT TRIGGER dfdms_intercept_ddl ON ddl_command_endEXECUTE PROCEDURE public.dfdms_intercept_ddl();
通过添加字段、删除字段、修改字段类型，测试上述方式。
这种方式不好的一点是：具有侵入性。如果采集的是第三方PG数据库，需要征求同意，在对方库中创建触发器。
你是否有更好的方法？加我微信，一起讨论下。
登峰大数据
专注大数据技术分享；Spark技术实战；大数据英文技术书籍的中文版独发以术会友
219篇原创内容
公众号
