实时数仓Hologres - 实时计算Flink版

本文为您介绍如何使用实时数仓Hologres连接器。

背景信息

实时数仓Hologres是一站式实时数据仓库引擎，支持海量数据实时写入、实时更新、实时分析，支持标准SQL（兼容PostgreSQL协议），支持PB级数据多维分析（OLAP）与即席分析（Ad Hoc），支持高并发低延迟的在线数据服务（Serving），与MaxCompute、Flink、DataWorks深度融合，提供离在线一体化全栈数仓解决方案。Hologres连接器支持的信息如下。

类别	详情
支持类型	源表、维表和结果表
运行模式	流模式和批模式
数据格式	暂不支持
特有监控指标	源表： numRecordsIn numRecordsInPerSecond 结果表： numRecordsOut numRecordsOutPerSecond currentSendTime 说明指标含义详情，请参见监控指标说明。
API种类	Datastream和SQL
是否支持更新或删除结果表数据	是

特色功能

源表

功能

详情

实时消费Hologres

支持作为非Binlog Source读取Hologres全量数据。
支持实时消费Hologres的Binlog数据。
- 支持非CDC模式消费。
- 支持CDC模式消费。
- 支持全增量一体源表消费。

获取更多信息，详情请参见Flink/Blink实时消费Hologres Binlog。

结果表

功能	详情
流式语义	支持写入Changelog消息。
宽表Merge和局部更新功能	只更新修改部分的数据，而非整行更新。
作为CTAS和CDAS的目标端	支持实时同步单表、整库的数据以及相应的表结构变更到Hologres表中。
插入部分列说明仅实时计算引擎VVR 6.0.7及以上版本支持。	支持将Flink INSERT DML中指定的列名下推给连接器，从而仅更新指定的列。

前提条件

已创建Hologres表，详情请参见创建Hologres表。

使用限制

通用：
- 仅Flink计算引擎VVR 2.0.0及以上版本支持Hologres连接器。
- Hologres连接器不支持访问Hologres外部表。关于Hologres外部表详情请参见基于HoloWeb创建MaxCompute外部表。
- 连接器目前的已知缺陷以及版本功能发布记录详见Hologres Connector Release Note。
源表独有：
- Flink默认以批模式读取Hologres源表数据，即只扫描一次Hologres全表，扫描结束，消费结束，新到Hologres源表的数据不会被读取。从VVR 3.0.0版本开始，支持实时消费Hologres数据，详情请参见实时计算Flink版实时消费Hologres。从VVR 6.0.3版本开始，Hologres源表在批模式读取时支持filter下推，详见源表参数enable_filter_push_down。
- 实时计算引擎8.0以下版本Hologres CDC模式暂不支持定义Watermark。如果您需要进行窗口聚合，您可以采用非窗口聚合的方式，详情请参见MySQL/Hologres CDC源表不支持窗口函数，如何实现类似每分钟聚合统计的需求?。
结果表独有：无。
维表独有：
- 维表建议使用主键作为Join条件，对于此类主键点查的维表，创建Hologres表时建议选择行存模式，列存模式对于点查场景性能开销较大。选择行存模式创建维表时必须设置主键，并且将主键设置为Clustering Key才可以工作。详情请参见CREATE TABLE。
- 如果业务需要，无法使用主键作为Join条件，对于此类非主键点查的维表（即一对多的查询），创建Hologres表时建议选择列存模式，并合理设置分布键Distribution Key以及Event Time Column（Segment Key）以优化查询性能，详情请参见表存储格式：列存、行存、行列共存。
- VVR 4.0以下版本仅支持对维表主键点查的维表Join，VVR 4.0及以上版本，jdbc模式支持维表的非主键点查。

注意事项

使用了rpc模式时，VVR版本升级注意事项：
Hologres 2.0版本下线了rpc（用于维表与结果表）模式，全面转为jdbc相关模式（目前包括jdbc、jdbc_fixed和jdbc_copy等），rpc模式不会对同一批次中相同主键的数据做去重，如果业务场景需要保留完整的数据，切换为jdbc模式后，可以通过设置'jdbcWriteBatchSize'='1'防止去重，或者升级到VVR 8.0.5版本配置deduplication.enabled参数为false。
如果您作业中存在使用了rpc模式读写Hologres的情况，此时如果您需要将VVR 4.x升级到VVR 6.x或VVR 8.x，请按照以下情况进行处理：
- 升级到VVR 6.0.4～6.0.6版本，可能会抛出异常。推荐维表和结果表使用jdbc_fixed或jdbc模式。
- 升级到VVR 6.0.7及以上版本，无需您做任何处理，Flink系统会自动替换rpc为jdbc相关模式。
使用binlog源表且未指定jdbc模式时，VVR版本升级注意事项：
Hologres 2.0版本开始有限支持holohub（用于Binlog源表）模式，Hologres 2.1版本彻底下线了holohub模式，全面转为jdbc模式。
如果您作业中存在消费binlog源表的情况，而且binlog源表未通过sdkmode='jdbc'指定jdbc模式，默认会使用holohub模式。此时如果您需要将VVR 4.x升级到VVR 6.x或VVR 8.x，请按照以下情况进行处理：
- 如果Hologres版本是2.0。
  - 升级到VVR 6.0.7～VVR 8.0.3版本，仍然可以继续使用holohub模式。
  - 升级到VVR 8.0.4及以上版本，可能抛出权限不足的异常，需要特别配置用户权限，详情见实时计算Flink版实时消费Hologres。
- 如果Hologres版本是2.1。
  - 升级到VVR 6.0.7～VVR 8.0.4版本，可能无法正常消费Binlog，建议升级到VVR 8.0.5。
  - 升级到VVR 8.0.5及以上版本，无需您做任何处理，Flink系统会自动替换holohub模式为jdbc模式。

语法结构

CREATE TABLE hologres_table (
  name VARCHAR,
  age BIGINT,
  birthday BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector' = 'hologres',
  'dbname' = '<yourDBName>',
  'tablename' = '<yourTableName>',
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'endpoint' = '<yourEndpoint>',
  'sdkmode' = 'jdbc'
);

WITH参数

通用

参数	说明	数据类型	是否必填	默认值	备注
connector	表类型。	String	是	无	固定值为`hologres`。
dbname	数据库名称。	String	是	无	Hologres V2.0版本推出了全新的弹性高可用实例形态，将计算资源分解为不同的计算组（Virtual Warehouse），更好的服务于高可用部署。不同的计算组使用相同的Endpoint，您可以通过在dbname参数后添加特定的后缀来指定连接某个计算组。例如某张维表希望连接特定的计算组read_warehouse，可以通过`'dbname' = 'db_test@read_warehouse'` 方式指定。说明仅JDBC相关模式支持使用计算组，详见源表、维表和结果表WITH参数中的sdkMode参数。
tablename	表名称。	String	是	无	如果Schema不为Public时，则tablename需要填写为`schema.tableName`。
username	用户名，请填写阿里云账号的AccessKey ID。	String	是	无	详情请参见如何查看AccessKey ID和AccessKey Secret信息？重要为了避免您的AK信息泄露，建议您使用变量的方式填写AccessKey取值，详情请参见项目变量。
password	密码，请填写阿里云账号的AccessKey Secret。	String	是	无
endpoint	Hologres服务地址。	String	是	无	详情请参见访问域名。
connection.ssl.mode	是否启用SSL（Secure Sockets Layer）传输加密，以及启用采用何种模式。	String	否	disable	参数取值如下： disable（默认值）：不启用传输加密。 require：启用SSL，只对数据链路加密。 verify-ca：启用SSL，加密数据链路，同时使用CA证书验证Hologres服务端的真实性。 verify-full：启用SSL，加密数据链路，使用CA证书验证Hologres服务端的真实性，同时比对证书内的CN或DNS与连接时配置的Hologres连接地址是否一致。说明 VVR 8.0.5及以上版本开始支持此参数。 Hologres自1.1版本起支持SSL传输加密的require模式，2.1版本起新增支持verify-ca和verify-full模式。详见传输加密。当配置为verify-ca或者verify-full时，需要同时配置connection.ssl.root-cert.location参数。
connection.ssl.root-cert.location	当传输加密模式需要证书时，配置证书的路径。	String	否	无	当connection.ssl.mode配置为verify-ca或者verify-full时，需要同时配置CA证书的路径。证书可以使用实时计算控制台的资源上传功能上传至平台，上传后文件存放在/flink/usrlib目录下。例如，需要使用的CA证书文件名为certificate.crt，则上传后参数取值应该为 `'/flink/usrlib/certificate.crt'`。说明 VVR 8.0.5及以上版本开始支持此参数。 CA证书获取方式见传输加密-下载CA证书。
jdbcRetryCount	当连接故障时，写入和查询的重试次数。	Integer	否	10	无。
jdbcRetrySleepInitMs	每次重试的固定等待时间。	Long	否	1000	实际重试的等待时间的计算公式为`jdbcRetrySleepInitMs+retry*jdbcRetrySleepStepMs`。单位为毫秒。
jdbcRetrySleepStepMs	每次重试的累加等待时间。	Long	否	5000	实际重试的等待时间的计算公式为`jdbcRetrySleepInitMs+retry*jdbcRetrySleepStepMs`。单位为毫秒。
jdbcConnectionMaxIdleMs	JDBC连接的空闲时间。	Long	否	60000	超过这个空闲时间，连接就会断开释放掉。单位为毫秒。
jdbcMetaCacheTTL	本地缓存TableSchema信息的过期时间。	Long	否	60000	单位为毫秒。
jdbcMetaAutoRefreshFactor	如果缓存的剩余时间小于触发时间，则系统会自动刷新缓存。	Integer	否	4	缓存的剩余时间计算方法：缓存的剩余时间=缓存的过期时间 - 缓存已经存活的时间。缓存自动刷新后，则从0开始重新计算缓存的存活时间。触发时间计算方法：jdbcMetaCacheTTL/jdbcMetaAutoRefreshFactor两个参数的比值。
type-mapping.timestamp-converting.legacy	Flink和Hologres之间是否进行时间类型的相互转换。	Boolean	否	true	参数取值如下： true：不进行相互转换。时区转换将采用运行环境中的JVM时区。 false（推荐）：进行相互转换。时区转换将使用Flink所配置的时区。说明仅实时计算引擎VVR 8.0.6及以上版本支持该参数。 Flink和Hologres的时区详情，请参见Flink与Hologres时区说明。 property-version=0时，默认值为true；property-version=1时，默认值为false。
property-version	Connector参数版本。	Integer	否	0	可填的值为0和1，默认值为0。说明仅VVR 8.0.6及以上版本支持配置该参数。在不同参数版本里，可用的参数集合和参数的默认值可能不同。如果存在区别，区别详情会在参数的说明部分描述。推荐使用参数版本1。

源表独有

参数	说明	数据类型	是否必填	默认值	备注
field_delimiter	导出数据时，不同行之间使用的分隔符。	String	否	"\u0002"	无。
binlog	是否消费Binlog数据。	Boolean	否	false	参数取值如下： true：消费Binlog数据。 false（默认值）：不消费Binlog数据。说明实时计算引擎VVR 8.0.6及以上版本支持多版本默认值。 property-version=0时，默认值为false。 property-version=1时，默认值为true。
sdkMode	SDK模式。	String	否	holohub	参数取值如下： holohub（默认值）：使用holohub模式消费binlog。 jdbc：使用jdbc模式消费binlog。 jdbc_fixed: 使用fixed jdbc模式消费binlog，与jdbc模式的区别在于不受连接数限制。目前此模式暂不支持消费开起数据脱敏功能的database的binlog，详情请参见数据脱敏。详情请参见Binlog Source表。版本推荐参数如下： VVR 6.0.3及以下版本：不支持配置该参数。 VVR 6.0.4～6.0.6版本：推荐使用默认配置，即holohub。 VVR 6.0.7及以上版本：推荐配置为jdbc。 VVR 8.0.4～8.0.5版本：系统会自动切换为jdbc。 VVR 8.0.6及以上版本：系统会根据Hologres版本进行自动切换。 2.1.27及以上版本，为jdbc_fixed模式。 2.1.0～2.1.26版本，为jdbc模式 Hologres实例为2.0版本，Flink系统会尝试使用jdbc模式，如果存在权限不足等异常，会自动选择holohub模式启动。说明 VVR 6.0.7及以上版本： Hologres实例为2.0以下版本，Flink系统采用您配置的参数值。 Hologres实例为2.0及以上版本，由于Hologres 2.0版本下线了holohub服务，此时如果您配置了holohub，Flink系统自动切换为jdbc。但是如果您配置为jdbc，Flink系统采用jdbc。 VVR 8.0.4～8.0.5版本： Hologres实例为2.0版本，Flink系统自动切换为jdbc。可能存在权限不足等问题，参考实时计算Flink版实时消费Hologres文档进行处理。
jdbcBinlogSlotName	JDBC模式的binlog源表的Slot名称。创建方法请参见JDBC模式Binlog源表。	String	否	无	仅在sdkMode配置为jdbc时生效，如果用户未配置，连接器会默认创建一个Slot来使用。详见JDBC模式Binlog源表。说明 Hologres实例2.1版本起，不再需要配置此参数，连接器也不会尝试自动创建。
binlogMaxRetryTimes	读取Binlog数据出错后的重试次数。	Integer	否	60	无。
binlogRetryIntervalMs	读取Binlog数据出错后的重试时间间隔。	Long	否	2000	单位为毫秒。
binlogBatchReadSize	批量读取Binlog的数据行数。	Integer	否	100	无。
cdcMode	是否采用CDC模式读取Binlog数据。	Boolean	否	false	参数取值如下： true：CDC模式读取Binlog数据。 false（默认值）：非CDC模式读取Binlog数据。说明实时计算引擎VVR 8.0.6及以上版本支持多版本默认值。 property-version=0时，默认值为false。 property-version=1时，默认值为true。
upsertSource	源表是否使用upsert类型的Changelog。	Boolean	否	false	仅在CDC模式下生效。参数取值如下： true：仅支持Upsert类型，包括INSERT、DELETE和UPDATE_AFTER。 false（默认值）：支持所有类型，包括INSERT、DELETE、UPDATE_BEFORE和UPDATE_AFTER。说明如果下游包含回撤算子（例如使用ROW_NUMBER OVER WINDOW去重），则需要设置upsertSource为true，此时源表会以Upsert方式从Hologres中读取数据。
binlogStartupMode	Binlog数据消费模式。	String	否	earliestOffset	参数取值如下： initial：先全量消费数据，再读取Binlog开始增量消费。 earliestOffset（默认值）：从最早的Binlog开始消费。 timestamp：从设置的startTime开始消费Binlog。说明如果设置了startTime或者在启动界面选择了启动时间，则binlogStartupMode强制使用timestamp模式，其他消费模式不生效，即startTime参数优先级更高。说明仅实时计算引擎VVR 4.0.13及以上版本支持该参数。实时计算引擎VVR 8.0.6及以上版本支持多版本默认值。 property-version=0时，默认值为false。 property-version=1时，默认值为true。
startTime	启动位点的时间。	String	否	无	格式为yyyy-MM-dd hh:mm:ss。如果没有设置该参数，且作业没有从State恢复，则从最早的Binlog开始消费Hologres数据。
jdbcScanFetchSize	扫描时攒批大小。	Integer	否	256	无。
jdbcScanTimeoutSeconds	扫描操作超时时间。	Integer	否	60	单位为秒。
jdbcScanTransactionSessionTimeoutSeconds	扫描操作所在事务的超时时间。	Integer	否	600秒（0表示不超时）	对应Hologres的GUC参数idle_in_transaction_session_timeout，详情请参见GUC参数。说明仅实时计算引擎Flink1.13-vvr-4.0.15及以上版本支持该参数。
enable_filter_push_down	全量读取阶段是否进行filter下推。	Boolean	否	false	参数取值如下： false（默认值）：不进行filter下推。 true：读取全量数据时，将支持的过滤条件下推到Hologres执行。包括非Binlog Source全量读取以及Binlog Source使用全增量一体化消费模式时的全量阶段。重要实时计算引擎Flink1.15-vvr-6.0.3到Flink1.15-vvr-6.0.5默认会进行filter下推，但如果作业使用了hologres维表，且写入的DML中包含有对维表非主键字段的过滤条件时，维表的filter会被错误的下推，可能导致维表join出现错误结果。因此推荐使用实时计算引擎Flink1.15-vvr-6.0.6及以上版本，并在源表增加此参数来开启过滤条件下推功能。

结果表独有

参数	说明	数据类型	是否必填	默认值	备注
sdkMode	SDK模式。	String	否	jdbc	参数取值如下： jdbc：默认值，表示使用jdbc模式进行写入。 jdbc_copy：是否使用fixed copy方式写入。 fixed copy是hologres1.3新增的能力，相比通过insert方法（jdbc模式）进行写入，fixed copy方式可以更高的吞吐（因为是流模式），更低的数据延时，更低的客户端内存消耗（因为不攒批）。但此模式暂不支持delete数据，也不支持写入分区父表，不支持ignoreNullWhenUpdate参数。 rpc：表示使用rpc模式进行写入，与useRpcMode参数一致，与jdbc模式的区别在于不占用连接数，并且不支持写入Hologres的JSONB、RoarinBitmap类型。 jdbc_fixed（beta功能）：表示使用fixed jdbc方式进行写入，需要Hologres引擎版本大于等于1.3，与jdbc模式的区别在于不占用连接数，不支持写入Hologres的Jsonb，RoarinBitmap类型。目前此模式暂不支持写入开启数据脱敏功能的database，详情请参见数据脱敏。版本推荐参数如下： VVR 6.0.3以下版本：不支持配置该参数。 VVR 6.0.4及以上版本：推荐配置为jdbc。说明 VVR 6.0.7～VVR 8.0.1版本：如果Hologres实例为2.0以下版本，Flink系统采用您配置的参数值。如果Hologres实例为2.0及以上版本，由于Hologres 2.0及以上版本下线了rpc服务，此时如果您将该参数配置为rpc，Flink系统将自动切换该参数值为jdbc_fixed，但是如果您配置为其他值，Flink系统将采用您配置的参数值。 rpc模式不会对同一批次中相同主键的数据做去重，如果业务场景需要保留完整的数据，切换为jdbc模式后，可以通过设置'jdbcWriteBatchSize'='1'防止去重。 VVR 8.0.3及以上版本自此版本开始，无论Hologres实例版本，都不再支持rpc模式，如果选择rpc模式，将自动切换该参数值为jdbc_fixed且设置'jdbcWriteBatchSize'='1'防止去重。
bulkload	是否采用bulkload写入。	Boolean	否	false	仅在sdkMode设置为jdbc_copy时生效。bulkload写入目前仅适用于无主键表或者主键保证不重复的有主键表（主键重复会抛出异常），相比默认的jdbc_copy，写入使用更少的Hologres资源。说明仅实时计算引擎VVR 8.0.5及以上版本且Hologres实例为2.1及以上版本支持该参数。
useRpcMode	是否通过RPC方式使用Hologres连接器。	Boolean	否	false	参数取值如下： true：使用RPC方式使用Hologres连接器。与sdkMode参数设置为rpc效果相同，通过RPC方式会降低SQL连接数。 false（默认值）：使用JDBC方式使用Hologres连接器。通过JDBC方式会占用SQL连接，导致JDBC连接数增加。说明 Hologres 2.0版本下线了rpc模式，推荐使用sdkMode参数来选择jdbc或者jdbc_fixed模式。实时计算引擎VVR 6.0.7及VVR 8.0.1版本在检测到Hologres实例是2.0及以上版本时，会自动切换rpc模式为jdbc_fixed模式。实时计算引擎VVR 8.0.3及以上版本会自动切换rpc模式为jdbc_fixed模式。 rpc模式不会对同一批次中相同主键的数据做去重，如果业务场景需要保留完整的数据变化记录，推荐使用实时计算引擎VVR 8.0.5及以上版本，jdbc模式可以配置deduplication.enabled参数为false不进行去重。 property-version=1时，该参数下线。
mutatetype	数据写入模式。	String	否	insertorignore	详情请参见流式语义。说明实时计算引擎VVR 8.0.6及以上版本支持多版本默认值。 property-version=0时，默认值为insertorignore。 property-version=1时，默认值为insertorupdate。
partitionrouter	是否写入分区表。	Boolean	否	false	无。
createparttable	当写入分区表时，是否根据分区值自动创建不存在的分区表。	Boolean	否	false	RPC模式下，如果分区值中存在短划线（-），暂不支持自动创建分区表。说明 Hologres实例1.3.22及以上版本开始支持使用Date类型做分区键。实时计算引擎VVR 8.0.3及以上版本，支持使用Date类型做分区键时自动创建分区表。请确保分区值不会出现脏数据，否则会创建错误的分区表导致Failover，建议慎用该参数。当sdk_mode设置为jdbc_copy时，不支持写入分区父表。
ignoredelete	是否忽略撤回消息。	Boolean	否	true	说明仅在使用流式语义时生效。实时计算引擎VVR 8.0.8及以上版本推荐使用`sink.delete-strategy`参数替换该参数。两个参数同时配置时，只有`sink.delete-strategy`参数生效。实时计算引擎VVR 8.0.6及以上版本支持多版本默认值。 property-version=0时，默认值为true。 property-version=1时，默认值为false。
sink.delete-strategy	撤回消息的处理策略。	String	否	无	参数取值如下： IGNORE_DELETE：忽略Update Before和Delete消息。适用于仅需插入或更新数据，而无需删除数据的场景。 NON_PK_FIELD_TO_NULL：忽略Update Before消息，并将Delete消息执行为将非主键字段更新为null。适用于希望在局部更新操作中执行删除操作而不影响其他列的场景。 DELETE_ROW_ON_PK：忽略Update Before消息，并将Delete消息执行为根据主键删除整行。适用于在局部更新过程中，希望执行删除整行操作，从而影响其他列的场景。 CHANGELOG_STANDARD：Flink框架按照 Flink SQL Changelog的工作原理运行，不忽略删除操作，并通过先删除数据再插入的方式执行更新操作，以确保数据准确性。适用于不涉及局部更新的场景。说明仅实时计算引擎VVR 8.0.8及以上版本支持该参数。启用NON_PK_FIELD_TO_NULL选项可能会导致记录中只有主键，其他所有列都为null。
connectionSize	单个Flink结果表任务所创建的JDBC连接池大小。	Integer	否	3	如果作业性能不足，建议您增加连接池大小。连接池大小和数据吞吐成正比。
jdbcWriteBatchSize	JDBC模式，Hologres Sink节点数据攒批条数（不是来一条数据处理一条，而是攒一批再处理）的最大值。	Integer	否	256	单位为数据行数。说明 jdbcWriteBatchSize、jdbcWriteBatchByteSize和jdbcWriteFlushInterval三者之间为或的关系。如果同时设置了这三个参数，则满足其中一个，就进行写入结果数据。
jdbcWriteBatchByteSize	JDBC模式，Hologres Sink节点数据攒批字节数（不是来一条数据处理一条，而是攒一批再处理）的最大值。	Long	否	2097152（210241024）字节，即2 MB	说明 jdbcWriteBatchSize、jdbcWriteBatchByteSize和jdbcWriteFlushInterval三者之间为或的关系。如果同时设置了这三个参数，则满足其中一个，就进行写入结果数据。
jdbcWriteFlushInterval	JDBC模式，Hologres Sink节点数据攒批写入Hologres的最长等待时间。	Long	否	10000	单位为毫秒。说明 jdbcWriteBatchSize、jdbcWriteBatchByteSize和jdbcWriteFlushInterval三者之间为或的关系。如果同时设置了这三个参数，则满足其中一个，就进行写入结果数据。
ignoreNullWhenUpdate	当mutatetype='insertOrUpdate'时，是否忽略更新写入数据中的Null值。	Boolean	否	false	取值说明如下： false（默认值）：将Null值写到Hologres结果表里。 true：忽略更新写入数据中的Null值。说明仅Flink计算引擎VVR 4.0及以上版本支持该参数。当sdk_mode设置为jdbc_copy时，不支持此参数。
connectionPoolName	连接池名称。同一个TaskManager中，配置相同名称的连接池的表可以共享连接池。	String	否	无	取值为非`'default'`的任意字符串。如果多个表设置相同的连接池，则这些使用相同连接池的表的connectionSize参数也需要相同。说明 VVR 4.0.12以下版本：不支持配置该参数。 VVR 4.0.12～VVR 8.0.3版本：默认不共享，每个表使用自己的连接池。 VVR 8.0.4以上版本：同一个作业中endpoint相同的表会默认共享连接池。作业中表数量较多时连接数可能相对不足影响性能，这种情况下推荐为不同的表设置不同的connectionPoolName。此参数可以按需配置，比如作业中有维表A，B以及结果表C，D，E五张hologres表，可以A表和B表使用'pool1'，C表和D表使用'pool2'，E表流量较大，单独使用'pool3'。
jdbcEnableDefaultForNotNullColumn	如果将Null值写入Hologres表中Not Null且无默认值的字段，是否允许连接器帮助填充一个默认值。	Boolean	否	true	参数取值如下： true（默认值）：允许连接器填充默认值并写入，规则如下。如果字段是String类型，则默认写为空（""）。如果字段是Number类型，则默认写为0。如果是Date、timestamp或timestamptz时间类型字段，则默认写为1970-01-01 00:00:00。 false：不填充默认值，写Null到Not Null字段时，会抛出异常。
remove-u0000-in-text.enabled	如果写入时字符串类型包含\u0000非法字符，是否允许连接器帮助去除。	Boolean	否	false	参数取值如下： false（默认值）：连接器不对数据进行操作，但碰到脏数据时写入可能抛出如下异常，`ERROR: invalid byte sequence for encoding "UTF8": 0x00` 此时需要在源表提前处理脏数据，或者在SQL中定义脏数据处理逻辑。 true：连接器会帮助去除字符串类型中的\u0000，防止写入抛出异常。重要实时计算引擎VVR 8.0.1及以上版本，仅`sdkMode='jdbc'`场景下支持该参数。实时计算引擎VVR 8.0.8及以上版本，仅`sdkMode='jdbc_copy'`或`sdkMode='jdbc'`场景下支持该参数。
partial-insert.enabled	是否只插入INSERT语句中定义的字段。	Boolean	否	false	参数取值如下： false（默认值）：无论INSERT语句中声明了哪些字段，都会更新结果表DDL中定义的所有字段，对于未在INSERT语句中声明的字段，会被更新为null。 true：将INSERT语句中定义的字段下推给连接器，从而可以只对声明的字段进行更新或插入。说明仅实时计算引擎VVR 6.0.7及以上版本支持该参数。此参数仅在mutatetype参数配置为InsertOrUpdate时生效。
deduplication.enabled	jdbc及jdbc_fixed模式写入攒批过程中，是否进行去重。	Boolean	否	true	参数取值如下： true（默认值）：如果一批数据中有主键相同的数据，默认进行去重，只保留最后一条到达的数据。以两个字段，其中第一个字段为主键的数据举例： `INSERT (1,'a')`和`INSERT (1,'b')`两条记录先后到达，去重之后只保留后到达的`(1,'b')`写入Hologres结果表中。 Hologres结果表中已经存在记录`(1,'a')`，此时`DELETE (1,'a')`和`INSERT (1,'b')`两条记录先后到达，只保留后到达的`(1,'b')`写入hologres中，表现为直接更新，而不是先删除再插入。 false：在攒批过程中不进行去重，如果发现新到的数据和目前攒批的数据中存在主键相同的情况，先将攒批数据写入，写入完成之后再继续写入新到的数据。说明仅实时计算引擎VVR 8.0.5及以上版本支持该参数。不允许攒批去重时，极端情况下（例如所有数据的主键都相同）写入会退化为不攒批的单条写入，对性能有一定影响。
check-and-put.column	启用条件更新能力，并指定检查的字段名。	String	否	无	参数取值必须设置为Hologres表存在的字段名。重要仅实时计算引擎VVR 8.0.11及以上版本支持该参数。仅`sdkMode`配置为`jdbc_fixed`或`jdbc`场景下支持该参数。结果表必须有主键， `mutateType`参数值必须是`Insertorupdate`或者`insertorreplace`。由于需要反查，建议结果表创建为行存表或者行列混存表。在数据重复较多的情况下，check-and-put操作会退化为单条写入，这将导致写入性能的降低。
check-and-put.operator	条件更新操作的比较操作符。	String	否	GREATER	比较新record的check字段与表中旧值，符合条件判断操作符时进行更新。目前支持配置为GREATER、GREATER_OR_EQUAL、EQUAL、NOT_EQUAL、LESS、LESS_OR_EQUAL、IS_NULL、IS_NOT_NULL。说明仅实时计算引擎VVR 8.0.11及以上版本支持该参数。
check-and-put.null-as	当条件更新时，如果旧数据为null，则将该null值视为此参数配置的有效值。	String	否	无	由于在PostgreSQL中，任何值与NULL进行比较的结果均为FALSE，因此当表中的原有数据为NULL时，进行更新操作时需要设置一个NULL-AS作为参数，相当于SQL中的COALESCE函数。说明仅实时计算引擎VVR 8.0.11及以上版本支持该参数。
aggressive.enabled	是否启用激进提交模式。	Boolean	否	false	设置为true时，即便攒批未达到预期条数，连接在空闲时将会被强制提交。在流量较小时，可以有效减少数据写入的延时。重要仅实时计算引擎VVR 8.0.11及以上版本支持该参数。仅`sdkMode`配置为`jdbc_fixed`或`jdbc`或`jdbc_copy`场景下支持该参数。

维表独有

参数	说明	数据类型	是否必填	默认值	备注
sdkMode	SDK模式。	String	否	jdbc	参数取值如下： jdbc（默认值）：表示使用jdbc模式进行查询，支持主键点查和非主键的查询，但是非主键的查询对性能影响较大，查询较慢。 rpc：表示使用rpc模式进行点查，与useRpcMode参数一致，仅支持主键点查，即维表的主键字段必须与Flink Join On的字段完全匹配，与jdbc模式的区别在于不占用连接数，不支持读取Hologres的Jsonb，RoarinBitmap类型。 jdbc_fixed：（beta功能，需要hologres引擎版本大于等于1.3）表示使用fixed jdbc方式进行点查，与jdbc模式的区别在于不占用连接数，且不支持读取Hologres的Jsonb，RoarinBitmap类型。仅支持主键点查，即维表的主键字段必须与Flink Join On的字段完全匹配。目前此模式暂不支持查询开起数据脱敏功能的database，详情请参见数据脱敏。版本推荐参数如下： VVR 6.0.3以下版本：不支持配置该参数。 VVR 6.0.4及以上版本：推荐配置为jdbc。说明在VVR 6.0.6版本，SDK模式选择为jdbc时，如果维表字符串类型的查询结果中包含null值，可能抛出空指针异常，此时推荐您使用rpc模式绕过。 VVR 6.0.7～VVR 8.0.1版本：如果Hologres实例为2.0以下版本，Flink系统将采用您配置的参数值。如果Hologres实例为2.0及以上版本，由于Hologres 2.0及以上版本下线了rpc服务，此时如果您将该参数配置为了rpc，Flink系统自动将参数值切换为jdbc_fixed。但是如果您配置为其他值，Flink系统将采用您配置的参数值。 VVR 8.0.3及以上版本：自此版本开始，无论Hologres实例版本，都不再支持rpc模式，如果选择rpc模式，将自动切换该参数值为jdbc_fixed。
useRpcMode	是否通过RPC方式使用Hologres连接器。	Boolean	否	false	参数取值如下： true：使用RPC方式使用Hologres连接器。与sdkMode参数设置为rpc效果相同。通过RPC方式会降低SQL连接数。 false（默认值）：使用JDBC方式使用Hologres连接器。通过JDBC方式会占用SQL连接，导致JDBC链接数增加。说明 Hologres 2.0版本下线了rpc了服务，推荐使用sdkMode参数来选择jdbc或者jdbc_fixed模式。实时计算引擎VVR 6.0.7及VVR 8.0.1版本在检测到Hologres实例是2.0及以上版本时，会自动切换rpc模式为jdbc_fixed模式。实时计算引擎VVR 8.0.3及以上版本会自动切换rpc模式为jdbc_fixed模式。
connectionSize	单个Flink维表任务所创建的JDBC连接池大小。	Integer	否	3	如果作业性能不足，建议您增加连接池大小。连接池大小和数据吞吐成正比。
connectionPoolName	连接池名称。同一个TaskManager中，配置相同名称的连接池的表可以共享连接池。	String	否	无	取值为非`'default'`的任意字符串。如果多个表设置相同的连接池，则这些使用相同连接池的表的connectionSize参数也需要相同。您可以按需配置此参数，例如作业中有维表A，B以及结果表C，D，E五张hologres表，可以A表和B表使用pool1，C表和D表使用pool2，E表流量较大，单独使用pool3。说明 VVR 4.0.12以下版本：不支持配置该参数。 VVR 4.0.12～VVR 8.0.3版本：默认不共享，每个表使用自己的连接池。 VVR 8.0.4以上版本：同一个作业中Endpoint相同的表会默认共享连接池。作业中表数量较多时连接数可能相对不足影响性能，这种情况下推荐为不同的表设置不同的connectionPoolName。
jdbcReadBatchSize	点查Hologres维表时，攒批处理的最大条数。	Integer	否	128	无。
jdbcReadBatchQueueSize	维表点查请求缓冲队列大小。	Integer	否	256	无。
jdbcReadTimeoutMs	维表点查的超时时间。	Long	否	默认值为0，表示不会超时	仅vvr 4.0.15-flink 1.13及以上版本、vvr 6.0.2-flink 1.15及以上版本支持该参数。
jdbcReadRetryCount	维表点查超时时的重试次数。	Interger	否	VVR 8.0.5以下版本：1 VVR 8.0.5及以上版本：10	仅VVR 6.0.3以上版本支持该参数。本参数与jdbcRetryCount不同，后者是指连接发生异常时的重试次数。
jdbcScanFetchSize	在一对多join（即没有使用完整主键）时使用scan接口，scan攒批处理数据的条数。	Integer	否	256	无。
jdbcScanTimeoutSeconds	scan操作的超时时间。	Integer	否	60	单位为秒。
cache	缓存策略。	String	否	见备注列。	Hologres仅支持None和LRU两种缓存策略，取值详情请参见背景信息。说明 Cache默认值和VVR版本有关： VVR 4.x版本及以上版本，默认值为None。 VVR 4.x版本以下版本，默认值为LRU。
cacheSize	缓存大小。	Integer	否	10000	选择LRU缓存策略后，可以设置缓存大小。单位为条。
cacheTTLMs	缓存更新时间间隔。	Long	否	见备注列。	单位为毫秒。cacheTTLMs默认值和cache的配置有关：如果cache配置为LRU，则cacheTTLMs为缓存超时时间。默认不过期。如果cache配置为None，则cacheTTLMs可以不配置，表示缓存不超时。
cacheEmpty	是否缓存join结果为空的数据。	Boolean	否	true	true（默认值）：缓存join结果为空的数据。 false：不缓存join结果为空的数据。但当join语句中AND前面条件符合而后面条件不符合时，依然会缓存join结果为空的数据。代码示例如下。 `LEFT JOIN latest_emergency FOR SYSTEM_TIME AS OF PROCTIME() AS t2 ON t1.alarm_id = t2.alarm_id -- 如果发现是动态告警，则匹配时加入动态告警id，否则无需考虑动态告警id字段。 AND CASE WHEN alarm_type = 2 THEN t1.dynamic_id = t2.dynamic_alarm_id ELSE true END`
async	是否异步返回数据。	Boolean	否	false	true：表示异步返回数据。 false（默认值）：表示不进行异步返回数据。说明异步返回数据是无序的。

类型映射

Flink与Hologres的数据类型映射请参见Blink/Flink与Hologres的数据类型映射。

使用示例

源表示例

非Binlog Source表

CREATE TEMPORARY TABLE hologres_source (
  name varchar, 
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'field_delimiter'='|' --该参数可选。
  'sdkmode' = 'jdbc'
);

CREATE TEMPORARY TABLE blackhole_sink(
  name varchar,
  age BIGINT,
  birthday BIGINT 
) WITH (
  'connector'='blackhole'
);

INSERT INTO blackhole_sink 
SELECT name, age, birthday
from hologres_source;

Binlog Source表

Hologres连接器支持实时消费Hologres，即实时消费Hologres的Binlog数据。Flink实时消费Hologres详情请参见实时计算Flink版实时消费Hologres。

结果表示例

CREATE TEMPORARY TABLE datagen_source(
  name varchar, 
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE hologres_sink (
  name varchar, 
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector'='hologres',           
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>'
);

INSERT INTO hologres_sink SELECT * from datagen_source;

维表示例

CREATE TEMPORARY TABLE datagen_source (
  a INT,
  b BIGINT,
  c STRING,
  proctime AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE hologres_dim (
  a INT, 
  b VARCHAR, 
  c VARCHAR
) WITH (
  'connector' = 'hologres',
   ...
);

CREATE TEMPORARY TABLE blackhole_sink (
  a INT,
  b STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink SELECT T.a,H.b
FROM datagen_source AS T JOIN hologres_dim FOR SYSTEM_TIME AS OF T.proctime AS H ON T.a = H.a;

特色功能详解

流式语义

流处理，也称为流数据或流事件处理，即对一系列无界数据或事件连续处理。执行流数据或流事件处理的系统通常允许您指定一种可靠性模式或处理语义，保证整个系统处理数据的准确性，因为网络或设备故障等可能会导致数据丢失。

根据Hologres Sink的配置和Hologres表的属性，流式语义分为以下两种：

Exactly-once（仅一次）：即使在发生各种故障的情况下，系统只处理一次数据或事件。
At-least-once（至少一次）：如果在系统完全处理之前丢失了数据或事件，则从源头重新传输，因此可以多次处理数据或事件。如果第一次重试成功，则不必进行后续重试。

在Hologres结果表中使用流式语义，需要注意以下几点：

如果Hologres物理表未设置主键，则Hologres Sink使用At-least-once语义。
如果Hologres物理表已设置主键，则Hologres Sink通过主键确保Exactly-once语义。当同主键数据出现多次时，您需要设置mutatetype参数确定更新结果表的方式，mutatetype取值如下：
- insertorignore（默认值）：保留首次出现的数据，忽略后续所有数据。
- insertorreplace：后出现的数据整行替换已有数据。
- insertorupdate：更新已有数据的部分列。例如一张表有a、b、c和d四个字段，a是PK（Primary Key），写入Hologres时只写入a和b两个字段，在PK重复的情况下，系统只会更新b字段，c和d保持不变。
说明
- 当mutatetype设置为insertorreplace或insertorupdate时，系统根据主键更新数据。
- Flink定义的结果表中的数据列数不一定要和Hologres物理表的列数一致，您需要保证缺失的列没有非空约束，即列值可以为Null，否则会报错。
默认情况下，Hologres Sink只能向一张表导入数据。如果导入数据至分区表的父表，即使导入成功，也会查询数据失败。您可以设置参数partitionRouter为true，开启自动将数据路由到对应分区表的功能。注意事项如下：
- tablename参数需要填写为父表的表名。
- 如果没有提前创建分区表，需要在WITH参数中启用createparttable=true，从而支持自动创建分区表，否则会导入失败。

宽表Merge和局部更新功能

在把多个流的数据写到一张Hologres宽表的场景中，会涉及到宽表Merge和数据的局部更新。下面通过示例来介绍实现宽表merge的两种方式。

说明

本场景的两种实现方式均具有如下限制：

宽表必须有主键。
每个数据流的数据都必须包含完整的主键字段。
列存模式的宽表Merge场景在高RPS的情况下，CPU使用率会偏高，建议关闭表中字段的Dictionary Encoding功能。

方式一（推荐）

说明

仅实时计算引擎VVR 6.0.7及以上版本支持使用此方式。

假设有两个Flink数据流，一个数据流中包含a、b和c字段，另一个数据流中包含a、d和e字段，Hologres宽表WIDE_TABLE包含a、b、c、d和e字段，其中a字段为主键。具体操作如下：

使用Flink SQL创建一张Hologres结果表，并声明a、b、c、d、e五个字段，映射至宽表WIDE_TABLE。
结果表的属性设置：
- mutatetype设置为insertorupdate，可以根据主键更新数据。
- ignoredelete设置为true，防止回撤消息产生Delete请求。实时计算引擎VVR 8.0.8及以上版本推荐使用sink.delete-strategy配置局部更新时的各种策略。

将两个Flink数据流的数据分别INSERT至对应的结果表中。

// 已经定义的source1和source2
CREATE TEMPORARY TABLE hologres_sink ( -- 声明a,b,c,d,e五个字段
  a BIGINT, 
  b STRING,
  c STRING,
  d STRING,
  e STRING,
  primary key(a) not enforced
) WITH (
  'connector'='hologres',           
  'dbname'='<yourDbname>',
  'tablename'='<yourWideTablename>',  -- hologres宽表，包含a,b,c,d,e五个字段
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'mutatetype'='insertorupdate',    -- 根据主键更新数据
  'ignoredelete'='true',            -- 忽略回撤消息产生的Delete请求
  'partial-insert.enabled'='true'   -- 开启部分列更新参数，支持仅更新INSERT语句中生命的字段
);

BEGIN STATEMENT SET;
INSERT INTO hologres_sink(a,b,c) select * from source1;  -- 声明只插入a,b,c三个字段
INSERT INTO hologres_sink(a,d,e) select * from source2;  -- 声明只插入a,d,e三个字段
END;

方式二

说明

仅实时计算引擎VVR 6.0.6及以下版本支持使用此方式。

使用Flink SQL创建两张Hologres结果表，其中一张表只声明a、b和c字段，另一张表只声明a、d和e字段。这两张表都映射至宽表WIDE_TABLE。
两张结果表的属性设置：
- mutatetype设置为insertorupdate，可以根据主键更新数据。
- ignoredelete设置为true，防止回撤消息产生Delete请求。实时计算引擎VVR 8.0.8及以上版本推荐使用sink.delete-strategy配置局部更新时的各种策略。

将两个Flink数据流的数据分别INSERT至对应的结果表中。

// 已经定义的source1和source2

CREATE TEMPORARY TABLE hologres_sink_1 ( -- 只声明a,b,c三个字段
  a BIGINT, 
  b STRING,
  c STRING,
  primary key(a) not enforced
) WITH (
  'connector'='hologres',           
  'dbname'='<yourDbname>',
  'tablename'='<yourWideTablename>',  -- hologres宽表，包含a,b,c,d,e五个字段
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'mutatetype' = 'insertorupdate',    -- 根据主键更新数据
  'ignoredelete' = 'true'             -- 忽略回撤消息产生的Delete请求
);

CREATE TEMPORARY TABLE hologres_sink_2 ( -- 只声明a,d,e三个字段
  a BIGINT, 
  d STRING,
  e STRING,
  primary key(a) not enforced
) WITH (
  'connector'='hologres',           
  'dbname'='<yourDbname>',
  'tablename'='<yourWideTablename>',  -- hologres宽表，包含a,b,c,d,e五个字段
  'username' = '${secret_values.ak_id}',
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'mutatetype'='insertorupdate',    -- 根据主键更新数据
  'ignoredelete'='true'             -- 忽略回撤消息产生的Delete请求
);

BEGIN STATEMENT SET;
INSERT INTO hologres_sink_1 select * from source1;
INSERT INTO hologres_sink_2 select * from source2;
END;

作为CTAS和CDAS的目标端

Hologres支持实时同步单表或整库级别的数据，在同步过程之中如果上游的表结构发生了变更也会实时同步到Hologres表中。新数据流到Hologres表时，Flink会先触发Hologres修改相应的表结构，然后再将数据写入到相应的表中。以上过程全部由Flink自动完成，您无需关心实现细节，详情请参见CREATE TABLE AS（CTAS）语句和数据库实时入仓快速入门。

DataStream API

重要

通过DataStream的方式读写数据时，需要使用对应的DataStream连接器连接实时计算Flink版，DataStream连接器设置方法请参见DataStream连接器使用方法。Maven中央库中已经放置了Hologres DataStream连接器。VVR 6.0.7请使用其中的1.15-vvr-6.0.7-1版本的依赖。VVR 8.0.7的依赖请通过ververica-connector-hologres-1.17-vvr-8.0.7.jar下载，在本地调试时，需要使用相应的Uber JAR，详见本地运行和调试包含连接器的作业，VVR 8.0.7对应的Uber JAR为ververica-connector-hologres-1.17-vvr-8.0.7-uber.jar。

Hologres源表

VVR提供了RichInputFormat的实现类HologresBulkreadInputFormat来读取Hologres表数据。以下为构建Hologres Source读取表数据的示例。

VVR 4.0.15

// 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
TableSchema schema = TableSchema.builder()
  .field("a", DataTypes.INT())
  .build();
// Hologres的相关参数。
Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
config.setString(HologresConfigs.USERNAME, "yourUserName");
config.setString(HologresConfigs.PASSWORD, "yourPassword");
config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
config.setString(HologresConfigs.TABLE, "yourTableName");
// 构建JDBC Options。
JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
String query = JDBCUtils.getSimpleSelectFromStatement(
  jdbcOptions.getTable(), schema.getFieldNames());

// 构建HologresBulkreadInputFormat读取表数据。
HologresBulkreadInputFormat inputFormat = new HologresBulkreadInputFormat(jdbcOptions, schema, query);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TypeInformation<RowData> typeInfo = InternalTypeInfo.of(schema.toRowDataType().getLogicalType());
env.addSource(new InputFormatSourceFunction<>(inputFormat, typeInfo)).returns(typeInfo)
  .print();
env.execute();

VVR 6.0.7&VVR 8.0.7

// set up the Java DataStream API
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
TableSchema schema = TableSchema.builder()
  .field("a", DataTypes.INT())
  .build();

// Hologres的相关参数。
Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
config.setString(HologresConfigs.USERNAME, "yourUserName");
config.setString(HologresConfigs.PASSWORD, "yourPassword");
config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
config.setString(HologresConfigs.TABLE, "yourTableName");

// 构建JDBC Options。
JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);

HologresBulkreadInputFormat inputFormat = new HologresBulkreadInputFormat(new HologresConnectionParam(config), jdbcOptions,  schema, "", -1);
TypeInformation<RowData> typeInfo = InternalTypeInfo.of(schema.toRowDataType().getLogicalType());
env.addSource(new InputFormatSourceFunction<>(inputFormat, typeInfo)).returns(typeInfo)
  .print();
env.execute();

XML

Maven中央库中已经放置了Hologres DataStream连接器。

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-hologres</artifactId>
    <version>${vvr-version}</version>
</dependency>

Hologres Binlog源表

VVR提供了Source的实现类HologresBinlogSource来读取Hologres Binlog数据。以下为构建Hologres Binlog Source的示例。

VVR 4.0.15

// 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
TableSchema schema = TableSchema.builder()
  .field("a", DataTypes.INT())
  .build();

// Hologres的相关参数。
Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
config.setString(HologresConfigs.USERNAME, "yourUserName");
config.setString(HologresConfigs.PASSWORD, "yourPassword");
config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
config.setString(HologresConfigs.TABLE, "yourTableName");
config.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);

// 构建JDBC Options。
JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
jdbcOptions.setHolohubEndpoint(JDBCUtils.getHolohubEndpoint(jdbcOptions));
RowDataRecordConverter recordConverter = buildRecordConverter(schema, config, jdbcOptions);

// 构建Hologres Binlog Source。
long startTimeMs = 0;
HologresBinlogSource<RowData> source = new HologresBinlogSource<>(
  schema,
  config,
  jdbcOptions,
  recordConverter,
  startTimeMs);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
env.execute();

VVR 6.0.7

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
TableSchema schema = TableSchema.builder()
  .field("a", DataTypes.INT())
  .build();

// Hologres的相关参数。
Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
config.setString(HologresConfigs.USERNAME, "yourUserName");
config.setString(HologresConfigs.PASSWORD, "yourPassword");
config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
config.setString(HologresConfigs.TABLE, "yourTableName");
config.setString(HologresConfigs.SDK_MODE, "jdbc");
config.setBoolean(HologresBinlogConfigs.OPTIONAL_BINLOG, true);
config.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);
// 构建JDBC Options。
JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
// 设置或创建默认slotname
config.setString(HologresBinlogConfigs.JDBC_BINLOG_SLOT_NAME, HoloBinlogUtil.getOrCreateDefaultSlotForJDBCBinlog(jdbcOptions));

boolean cdcMode = config.get(HologresBinlogConfigs.BINLOG_CDC_MODE)
  && config.get(HologresBinlogConfigs.OPTIONAL_BINLOG);
// 构建Binlog Record Converter。
JDBCBinlogRecordConverter recordConverter = new JDBCBinlogRecordConverter(
  jdbcOptions.getTable(),
  schema,
  new HologresConnectionParam(config),
  cdcMode,
  Collections.emptySet());

// 构建Hologres Binlog Source。
long startTimeMs = 0;
HologresJDBCBinlogSource source = new HologresJDBCBinlogSource(
  new HologresConnectionParam(config),
  schema,
  config,
  jdbcOptions,
  startTimeMs,
  StartupMode.TIMESTAMP,
  recordConverter,
  "",
  -1);

env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
env.execute();

VVR 8.0.7

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
TableSchema schema = TableSchema.builder()
  .field("a", DataTypes.INT())
  .build();

// Hologres的相关参数。
Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
config.setString(HologresConfigs.USERNAME, "yourUserName");
config.setString(HologresConfigs.PASSWORD, "yourPassword");
config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
config.setString(HologresConfigs.TABLE, "yourTableName");
config.setString(HologresConfigs.SDK_MODE, "jdbc");
config.setBoolean(HologresBinlogConfigs.OPTIONAL_BINLOG, true);
config.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);
// 构建JDBC Options。
JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
// 构建Hologres Binlog Source。
long startTimeMs = 0;
HologresBinlogSource source = new HologresBinlogSource(
  new HologresConnectionParam(config),
  schema,
  config,
  jdbcOptions,
  startTimeMs,
  StartupMode.INITIAL,
  "",
  "",
  -1,
  Collections.emptySet());

env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
env.execute();

说明

方法buildRecordConverter不在VVR Connector依赖中，是示例代码中提供的方法。
Hologres Binlog注意事项和实现原理等详情，请参见Binlog Source表。

Hologres结果表

VVR提供了OutputFormatSinkFunction的实现类HologresSinkFunction来写入数据。以下为构建Hologres Sink的示例。

VVR 4.0.15

// 初始化读取的表的Schema。
TableSchema schema = TableSchema.builder()
  .field("a", DataTypes.INT())
  .field("b", DataTypes.STRING())
  .build();

// Hologres的相关参数。
Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
config.setString(HologresConfigs.USERNAME, "yourUserName");
config.setString(HologresConfigs.PASSWORD, "yourPassword");
config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
config.setString(HologresConfigs.TABLE, "yourTableName");
config.setBoolean(HologresConfigs.USE_RPC_MODE, true);
HologresConnectionParam hologresConnectionParam = new HologresConnectionParam(config);

// 构建Hologres Writer，以RowData的方式写入数据。
AbstractHologresWriter<RowData> hologresWriter =
  buildHologresWriter(schema, config, hologresConnectionParam);

// 构建Hologres Sink。
HologresSinkFunction sinkFunction = new HologresSinkFunction(hologresConnectionParam, hologresWriter);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TypeInformation<RowData> typeInfo = InternalTypeInfo.of(schema.toRowDataType().getLogicalType());
int offset = (int) (System.currentTimeMillis() % Integer.MAX_VALUE);
env.fromElements((RowData)GenericRowData.of(2 + offset, StringData.fromString("2")), GenericRowData.of(3 + offset, StringData.fromString("3"))).returns(typeInfo)
  .addSink(sinkFunction);
env.execute();

VVR 6.0.7&VVR 8.0.7

// set up the Java DataStream API
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
TableSchema schema = TableSchema.builder()
  .field("a", DataTypes.INT())
  .field("b", DataTypes.STRING())
  .build();

// Hologres的相关参数。
Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
config.setString(HologresConfigs.USERNAME, "yourUserName");
config.setString(HologresConfigs.PASSWORD, "yourPassword");
config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
config.setString(HologresConfigs.TABLE, "yourTableName");
config.setString(HologresConfigs.SDK_MODE, "jdbc");

HologresConnectionParam hologresConnectionParam = new HologresConnectionParam(config);
// 构建Hologres Writer，以RowData的方式写入数据。
AbstractHologresWriter<RowData> hologresWriter = HologresJDBCWriter.createRowDataWriter(
  hologresConnectionParam, schema, HologresTableSchema.get(hologresConnectionParam), new Integer[0]);
// 构建Hologres Sink。
HologresSinkFunction sinkFunction = new HologresSinkFunction(hologresConnectionParam, hologresWriter);
TypeInformation<RowData> typeInfo = InternalTypeInfo.of(schema.toRowDataType().getLogicalType());
int offset = (int) (System.currentTimeMillis() % Integer.MAX_VALUE);
env.fromElements((RowData)GenericRowData.of(2 + offset, StringData.fromString("2")), GenericRowData.of(3 + offset, StringData.fromString("3"))).returns(typeInfo)
  .addSink(sinkFunction);

env.execute();

说明

方法buildHologresWriter不在VVR Connector依赖中，是示例代码中提供的方法。

Flink与Hologres时区说明

时间类型

产品	类型	说明
Flink	Flink TIMESTAMP	表示没有时区信息的日期和时间，描述年、月、日、小时、分钟、秒和小数秒对应的时间戳。可以通过一个字符串来指定，例如`1970-01-01 00:00:04.001`。
Flink	Flink TIMESTAMP_LTZ	用于描述时间线上的绝对时间点，使用long保存从epoch至今的毫秒数，使用int保存毫秒中的纳秒数。epoch时间是从Java的标准epoch时间开始计算。在计算和可视化时，每个TIMESTAMP_LTZ类型的数据都使用Session （会话）中配置的时区。可以用于跨时区的计算，因为它是一个基于epoch的绝对时间点，代表的就是不同时区的同一个绝对时间点。相同的TIMESTAMP_LTZ值，在不同的时区可能会反映出不同的本地TIMESTAMP，例如：如果一个TIMESTAMP_LTZ值为`2024-03-19T04:00:00Z`，在上海时区（UTC+8）的本地时间戳会表示为`2024-03-19T12:00:00`，而在格林威治时区（UTC+0）则表示为`2024-03-19T04:00:00`。
Hologres	TIMESTAMP	类似于Flink的`TIMESTAMP`类型，表示没有时区信息的日期和时间。存储数据时不会改变，即使客户端的时区发生变化，存储的值也保持不变。例如：`2022-01-01 01:01:01.123456`。
Hologres	TIMESTAMP WITH TIME ZONE (TIMESTAMPTZ)	类似于Flink的`TIMESTAMP_LTZ`类型，它带有时区信息。Hologres存储`TIMESTAMPTZ`数据时，会将其转换为UTC时区的值。当查询数据时，Hologres会根据客户端的时区配置参数将UTC时区的值转换为客户端时区的值。例如北京（UTC+8）时区的时间戳`2022-02-01 10:33:20.125+08`。在Hologres中存储为TIMESTAMPTZ类型时，其值会是`2022-02-01 10:33:20.125+08`。

时间类型映射

实时计算引擎VVR 8.0.6及以上版本且type-mapping.timestamp-converting.legacy=false时，支持所有时间类型间的相互转换。

Flink	Hologres	详情
TIMESTAMP	TIMESTAMP	之间相互转换是直接的，不涉及时区转换。因此推荐采用该数据映射。
TIMESTAMP LTZ	TIMESTAMPTZ	之间相互转换是直接的，不涉及时区转换。因此推荐采用该数据映射。
TIMESTAMP	TIMESTAMPTZ	之间的转换涉及时区转换。为了在转换中保持准确性，需要通过配置项参数`table.local-time-zone`设置Flink时区，配置项参数设置方法请参见如何配置自定义的作业运行参数？。例如当设置`'table.local-time-zone': 'Asia/Shanghai'`时，表示Flink时区为上海（+8时区）时，Flink TIMESTAMP类型的数据为2022-01-01 01:01:01.123456，写入Hologres TIMESTAMP TZ的数值为2022-01-01 01:01:01.123456+8。
TIMESTAMP LTZ	TIMESTAMP

实时计算引擎VVR8.0.6及以上版本且type-mapping.timestamp-converting.legacy=true时或者VVR 8.0.5及以下版本，除TIMESTAMP间转化，其他类型相互转化可能会出现数据偏差问题。

Flink	Hologres	备注
TIMESTAMP	TIMESTAMP	之间相互转换是直接的，不涉及时区转换。因此推荐采用该数据映射。
TIMESTAMP LTZ	TIMESTAMPTZ	读写Hologres数据时都当作无时区时间进行处理，可能会存在数据偏差。例如，Flink TIMESTAMP_LTZ类型的数值为2024-03-19T04:00:00Z，在上海（+8时区）对应的实际无时区时间为2024-03-19T12:00:00，但是写入时将2024-03-19T04:00:00当作无时区时间，写入Hologres TIMESTAMPTZ的数值为2024-03-19T04:00:00+08，数值偏差8小时。
TIMESTAMP	TIMESTAMPTZ	时区转换默认采用的是运行环境的JVM时区，而不是Flink时区，这与Flink内部计算的时区转换格式不同。当Flink时区与机器的JVM时区不一致时，会导致数据存在偏差，建议采用Flink时区进行Hologres数据的读写。
TIMESTAMP LTZ	TIMESTAMP

实时计算Flink版：实时数仓Hologres

背景信息

特色功能

源表

结果表

前提条件

使用限制

注意事项

语法结构

WITH参数

类型映射

使用示例

源表示例

非Binlog Source表

Binlog Source表

结果表示例

维表示例

特色功能详解

流式语义

宽表Merge和局部更新功能

方式一（推荐）

方式二

作为CTAS和CDAS的目标端

DataStream API

Hologres源表

VVR 4.0.15

VVR 6.0.7&VVR 8.0.7

XML

Hologres Binlog源表

VVR 4.0.15

VVR 6.0.7

VVR 8.0.7

Hologres结果表

VVR 4.0.15

VVR 6.0.7&VVR 8.0.7

Flink与Hologres时区说明

时间类型

时间类型映射

相关文档