This topic provides the DDL syntax that is used to create an ApsaraDB for Redis dimension
table, describes the parameters in the WITH and CACHE clauses, and provides data type
mappings and sample code.
Note Flink can read data from self-managed Redis databases.
What is ApsaraDB for Redis?
ApsaraDB for Redis is a database service that is compatible with the protocols of the open source Redis
system. ApsaraDB for Redis supports a hybrid of memory and hard disks for storage.
ApsaraDB for Redis provides a hot standby architecture to ensure high availability.
ApsaraDB for Redis can meet the business requirements for high throughputs, low-latency
operations, and flexible configuration changes based on the scalable cluster architecture.
Limits
- Only Flink that uses Ververica Runtime (VVR) 2.0.0 or later versions supports ApsaraDB
for Redis connectors.
- ApsaraDB for Redis dimension tables can read only the data of the STRING type in ApsaraDB
for Redis databases.
DDL syntax
CREATE TABLE redis_dim (
id STRING,
name STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'redis',
'host' = '<yourHost>',
'port' = '<yourPort>',
'password' = '<yourPassword>',
'dbNum' = '<yourDbNum>'
);
Note
- You must declare only one primary key for each ApsaraDB for Redis dimension table.
- You can declare only two fields for an ApsaraDB for Redis dimension table. The data
type of the fields must be STRING.
- When you join a dimension table with another table, the ON condition must contain
equality conditions that include all primary keys.
- You must set the cache parameter to None or LRU for an ApsaraDB for Redis dimension
table.
Parameters in the WITH clause
Parameter |
Description |
Required |
Remarks |
connector |
The type of the dimension table. |
Yes |
Set the value to redis .
|
host |
The address that is used to access the ApsaraDB for Redis database. |
Yes |
N/A. |
port |
The port of the ApsaraDB for Redis database. |
No |
Default value: 6379. |
dbNum |
The sequence number of the ApsaraDB for Redis database. |
No |
Default value: 0. |
password |
The password that is used to access the ApsaraDB for Redis database. |
No |
By default, this parameter is empty. This indicates that permission verification is
not required.
|
clusterMode |
Specifies whether the ApsaraDB for Redis cluster is in cluster mode. |
No |
Valid values:
- True indicates that the cluster is in cluster mode.
- False indicates that the cluster is not in cluster mode. Default value: False.
|
hashName |
The hash key name in hash mode. |
No |
By default, this parameter is empty. In typical cases, the data type in the ApsaraDB
for Redis dimension table is STRING, which is represented as key-value pairs. If you configure the hashName parameter, the ApsaraDB for Redis dimension table stores data of the HASHMAP data
type in the format of key-{field-value} pairs.
- key is the value of the hashName parameter.
- field is the value of the key parameter that you specify in the CREATE TABLE statement.
- value is the value that is assigned to key. The value field returns the same results as value in
key-value for the STRING data type.
|
Parameters in the CACHE clause
Parameter |
Description |
Required |
Remarks |
cache |
The cache policy. |
No |
Valid values:
|
cacheSize |
The maximum number of data records that can be cached. |
No |
This parameter is available only if you set the cache parameter to LRU. Default value:
10000.
|
cacheTTLMs |
The cache timeout period. Unit: milliseconds. |
No |
The cacheTTLMs parameter applies configurations based on the value of the cache parameter.
- If you set the cache parameter to None, the cacheTTLMs parameter can be left empty. This indicates that cache entries do not expire.
- If you set the cache parameter to LRU, the cacheTTLMs parameter specifies the cache timeout period. By default, cache entries do not expire.
|
cacheEmpty |
Specifies whether to cache empty results. |
No |
Default value: true. |
Data type mapping
Data type of ApsaraDB for Redis |
Data type of Flink |
STRING |
VARCHAR |
Sample code
CREATE TEMPORARY TABLE datagen_source (
id STRING,
data STRING,
proctime as PROCTIME()
) WITH (
'connector' = 'datagen'
);
CREATE TEMPORARY TABLE redis_dim (
id STRING,
name STRING,
PRIMARY KEY (id) NOT ENFORCED -- The row key type in the ApsaraDB for Redis database.
) WITH (
'connector' = 'redis',
'host' = '<yourHost>',
'port' = '<yourPort>',
'password' = '<yourPassword>'
);
CREATE TEMPORARY TABLE blackhole_sink (
id STRING,
data STRING,
name STRING
) WITH (
'connector' = 'blackhole'
);
INSERT INTO blackhole_sink
SELECT e.*, w.*
FROM datagen_source AS e
JOIN redis_dim FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;