Add Redis DSL connector

docs/docs-cn/source/5.application-development/3.connector/11.redis.md

@@ -0,0 +1,92 @@
+# Redis Connector 介绍
+
+Redis Connector 支持在 DSL 中把 Redis 作为表 Source 和 Sink 使用。该连接器复用项目中
+Redis Store 已使用的 Jedis 客户端依赖，不引入新的 Redis 客户端。
+
+Sink 支持写入 Redis string 和 hash。Source 通过 Redis `SCAN` 命令按 key pattern
+扫描，并读取 string 或 hash 数据。
+
+## 语法示例
+
+### String Source/Sink
+
+```sql
+CREATE TABLE redis_string (
+  redis_key VARCHAR,
+  redis_value VARCHAR
+) WITH (
+  type = 'redis',
+  geaflow.dsl.redis.host = '127.0.0.1',
+  geaflow.dsl.redis.port = '6379',
+  geaflow.dsl.redis.data.type = 'string',
+  geaflow.dsl.redis.key.field = 'redis_key',
+  geaflow.dsl.redis.value.field = 'redis_value',
+  geaflow.dsl.redis.key.pattern = 'result:*'
+);
+```
+
+### Hash Source/Sink
+
+```sql
+CREATE TABLE redis_hash (
+  redis_key VARCHAR,
+  hash_field VARCHAR,
+  hash_value VARCHAR
+) WITH (
+  type = 'redis',
+  geaflow.dsl.redis.host = '127.0.0.1',
+  geaflow.dsl.redis.port = '6379',
+  geaflow.dsl.redis.data.type = 'hash',
+  geaflow.dsl.redis.key.field = 'redis_key',
+  geaflow.dsl.redis.hash.field.field = 'hash_field',
+  geaflow.dsl.redis.hash.value.field = 'hash_value',
+  geaflow.dsl.redis.key.pattern = 'dim:*'
+);
+```
+
+Hash Sink 有两种写入方式：配置 `geaflow.dsl.redis.hash.field.field` 和
+`geaflow.dsl.redis.hash.value.field` 时，每行写入一个 hash field；
+不配置时，连接器会把 key 字段以外的所有列写入同一个 Redis hash，
+列名作为 hash field 名。
+
+## 参数
+
+| 参数名 | 是否必填 | 默认值 | 描述 |
+| -------- | -------- | -------- | -------- |
+| geaflow.dsl.redis.host | 否 | 127.0.0.1 | Redis 服务地址。 |
+| geaflow.dsl.redis.port | 否 | 6379 | Redis 服务端口。 |
+| geaflow.dsl.redis.user | 否 | 空 | Redis 用户名。 |
+| geaflow.dsl.redis.password | 否 | 空 | Redis 密码。 |
+| geaflow.dsl.redis.connection.timeout | 否 | 5000 | Redis 连接超时时间，单位毫秒。 |
+| geaflow.dsl.redis.data.type | 否 | string | Redis 数据类型，支持 `string` 和 `hash`。 |
+| geaflow.dsl.redis.key.field | 否 | redis_key | 作为 Redis key 的表字段。 |
+| geaflow.dsl.redis.value.field | 否 | redis_value | 作为 Redis string value 的表字段。 |
+| geaflow.dsl.redis.hash.field.field | 否 | 空 | 作为 Redis hash field 名的表字段。 |
+| geaflow.dsl.redis.hash.value.field | 否 | 空 | 作为 Redis hash field value 的表字段。 |
+| geaflow.dsl.redis.key.pattern | 否 | * | Source 扫描使用的 key pattern。 |
+| geaflow.dsl.redis.scan.count | 否 | 100 | 每次 Redis `SCAN` 的 count 参数。 |
+
+## 示例
+
+```sql
+CREATE TABLE file_source (
+  redis_key VARCHAR,
+  redis_value VARCHAR
+) WITH (
+  type = 'file',
+  geaflow.dsl.file.path = '/path/to/file'
+);
+
+CREATE TABLE redis_sink (
+  redis_key VARCHAR,
+  redis_value VARCHAR
+) WITH (
+  type = 'redis',
+  geaflow.dsl.redis.host = '127.0.0.1',
+  geaflow.dsl.redis.port = '6379',
+  geaflow.dsl.redis.data.type = 'string'
+);
+
+INSERT INTO redis_sink
+SELECT redis_key, redis_value FROM file_source;
+```

docs/docs-cn/source/5.application-development/3.connector/index.rst

@@ -16,4 +16,4 @@
    8.hudi.md
    9.pulsar.md
    10.udc.md
-
+   11.redis.md

docs/docs-en/source/5.application-development/3.connector/11.redis.md

@@ -0,0 +1,92 @@
+# Redis Connector Introduction
+
+The Redis connector supports using Redis as a DSL table source and sink. It reuses
+the Jedis client dependency already used by the Redis store module.
+
+The sink supports Redis string and hash writes. The source scans Redis keys by
+pattern with the Redis `SCAN` command and reads string or hash data.
+
+## Syntax
+
+### String sink/source
+
+```sql
+CREATE TABLE redis_string (
+  redis_key VARCHAR,
+  redis_value VARCHAR
+) WITH (
+  type = 'redis',
+  geaflow.dsl.redis.host = '127.0.0.1',
+  geaflow.dsl.redis.port = '6379',
+  geaflow.dsl.redis.data.type = 'string',
+  geaflow.dsl.redis.key.field = 'redis_key',
+  geaflow.dsl.redis.value.field = 'redis_value',
+  geaflow.dsl.redis.key.pattern = 'result:*'
+);
+```
+
+### Hash sink/source
+
+```sql
+CREATE TABLE redis_hash (
+  redis_key VARCHAR,
+  hash_field VARCHAR,
+  hash_value VARCHAR
+) WITH (
+  type = 'redis',
+  geaflow.dsl.redis.host = '127.0.0.1',
+  geaflow.dsl.redis.port = '6379',
+  geaflow.dsl.redis.data.type = 'hash',
+  geaflow.dsl.redis.key.field = 'redis_key',
+  geaflow.dsl.redis.hash.field.field = 'hash_field',
+  geaflow.dsl.redis.hash.value.field = 'hash_value',
+  geaflow.dsl.redis.key.pattern = 'dim:*'
+);
+```
+
+For hash sinks, if `geaflow.dsl.redis.hash.field.field` and
+`geaflow.dsl.redis.hash.value.field` are not configured, the connector writes all
+columns except the key column into the Redis hash, using column names as hash
+field names.
+
+## Options
+
+| Key | Required | Default | Description |
+| -------- | -------- | -------- | -------- |
+| geaflow.dsl.redis.host | No | 127.0.0.1 | Redis server host. |
+| geaflow.dsl.redis.port | No | 6379 | Redis server port. |
+| geaflow.dsl.redis.user | No | empty | Redis user name. |
+| geaflow.dsl.redis.password | No | empty | Redis password. |
+| geaflow.dsl.redis.connection.timeout | No | 5000 | Redis connection timeout in milliseconds. |
+| geaflow.dsl.redis.data.type | No | string | Redis data type, `string` or `hash`. |
+| geaflow.dsl.redis.key.field | No | redis_key | Table field used as the Redis key. |
+| geaflow.dsl.redis.value.field | No | redis_value | Table field used as the Redis string value. |
+| geaflow.dsl.redis.hash.field.field | No | empty | Table field used as the Redis hash field name. |
+| geaflow.dsl.redis.hash.value.field | No | empty | Table field used as the hash field value. |
+| geaflow.dsl.redis.key.pattern | No | * | Key pattern used by source scan. |
+| geaflow.dsl.redis.scan.count | No | 100 | Redis `SCAN` count per fetch. |
+
+## Example
+
+```sql
+CREATE TABLE file_source (
+  redis_key VARCHAR,
+  redis_value VARCHAR
+) WITH (
+  type = 'file',
+  geaflow.dsl.file.path = '/path/to/file'
+);
+
+CREATE TABLE redis_sink (
+  redis_key VARCHAR,
+  redis_value VARCHAR
+) WITH (
+  type = 'redis',
+  geaflow.dsl.redis.host = '127.0.0.1',
+  geaflow.dsl.redis.port = '6379',
+  geaflow.dsl.redis.data.type = 'string'
+);
+
+INSERT INTO redis_sink
+SELECT redis_key, redis_value FROM file_source;
+```

docs/docs-en/source/5.application-development/3.connector/index.rst

@@ -15,4 +15,5 @@ Connector
    7.hbase.md
    8.hudi.md
    9.pulsar.md
-   10.udc.md
+   10.udc.md
+   11.redis.md

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/pom.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you under the Apache License, Version 2.0 (the
+  ~ "License"); you may not use this file except in compliance
+  ~ with the License.  You may obtain a copy of the License at
+  ~
+  ~   http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing,
+  ~ software distributed under the License is distributed on an
+  ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  ~ KIND, either express or implied.  See the License for the
+  ~ specific language governing permissions and limitations
+  ~ under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.geaflow</groupId>
+        <artifactId>geaflow-dsl-connector</artifactId>
+        <version>0.8.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>geaflow-dsl-connector-redis</artifactId>
+    <name>geaflow-dsl-connector-redis</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-connector-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.github.fppt</groupId>
+            <artifactId>jedis-mock</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>${testng.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisClient.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.redis;
+
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.apache.geaflow.common.config.Configuration;
+import redis.clients.jedis.JedisPool;
+
+public final class RedisClient {
+
+    private RedisClient() {
+    }
+
+    public static JedisPool createJedisPool(Configuration conf) {
+        String host = conf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_HOST);
+        int port = conf.getInteger(RedisConfigKeys.GEAFLOW_DSL_REDIS_PORT);
+        String user = conf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_USER);
+        String password = conf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_PASSWORD);
+        int timeout = conf.getInteger(RedisConfigKeys.GEAFLOW_DSL_REDIS_CONNECTION_TIMEOUT);
+        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
+        return new JedisPool(poolConfig, host, port, timeout, user, password);
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisConfigKeys.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.redis;
+
+import org.apache.geaflow.common.config.ConfigKey;
+import org.apache.geaflow.common.config.ConfigKeys;
+
+public class RedisConfigKeys {
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_HOST = ConfigKeys
+        .key("geaflow.dsl.redis.host")
+        .defaultValue("127.0.0.1")
+        .description("Redis server host.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_PORT = ConfigKeys
+        .key("geaflow.dsl.redis.port")
+        .defaultValue(6379)
+        .description("Redis server port.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_USER = ConfigKeys
+        .key("geaflow.dsl.redis.user")
+        .defaultValue("")
+        .description("Redis user name.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_PASSWORD = ConfigKeys
+        .key("geaflow.dsl.redis.password")
+        .defaultValue("")
+        .description("Redis password.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_CONNECTION_TIMEOUT = ConfigKeys
+        .key("geaflow.dsl.redis.connection.timeout")
+        .defaultValue(5000)
+        .description("Redis connection timeout in milliseconds.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_DATA_TYPE = ConfigKeys
+        .key("geaflow.dsl.redis.data.type")
+        .defaultValue("string")
+        .description("Redis data type. Supported values are string and hash.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_KEY_FIELD = ConfigKeys
+        .key("geaflow.dsl.redis.key.field")
+        .defaultValue("redis_key")
+        .description("Table field used as Redis key.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_VALUE_FIELD = ConfigKeys
+        .key("geaflow.dsl.redis.value.field")
+        .defaultValue("redis_value")
+        .description("Table field used as Redis string value.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_HASH_FIELD_FIELD = ConfigKeys
+        .key("geaflow.dsl.redis.hash.field.field")
+        .defaultValue("")
+        .description("Table field used as Redis hash field name.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_HASH_VALUE_FIELD = ConfigKeys
+        .key("geaflow.dsl.redis.hash.value.field")
+        .defaultValue("")
+        .description("Table field used as Redis hash field value.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_KEY_PATTERN = ConfigKeys
+        .key("geaflow.dsl.redis.key.pattern")
+        .defaultValue("*")
+        .description("Redis key pattern used by source scan.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_SCAN_COUNT = ConfigKeys
+        .key("geaflow.dsl.redis.scan.count")
+        .defaultValue(100)
+        .description("Redis SCAN count per fetch.");
+}