diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.redis.md b/docs/docs-cn/source/5.application-development/3.connector/11.redis.md new file mode 100644 index 000000000..4c47e970e --- /dev/null +++ b/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; +``` diff --git a/docs/docs-cn/source/5.application-development/3.connector/index.rst b/docs/docs-cn/source/5.application-development/3.connector/index.rst index 5ca142767..8a8b7a663 100644 --- a/docs/docs-cn/source/5.application-development/3.connector/index.rst +++ b/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 diff --git a/docs/docs-en/source/5.application-development/3.connector/11.redis.md b/docs/docs-en/source/5.application-development/3.connector/11.redis.md new file mode 100644 index 000000000..0bb835bdd --- /dev/null +++ b/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; +``` diff --git a/docs/docs-en/source/5.application-development/3.connector/index.rst b/docs/docs-en/source/5.application-development/3.connector/index.rst index 7e696edc4..a88ec1e17 100644 --- a/docs/docs-en/source/5.application-development/3.connector/index.rst +++ b/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 \ No newline at end of file + 10.udc.md + 11.redis.md diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/pom.xml new file mode 100644 index 000000000..f7b8cc50c --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/pom.xml @@ -0,0 +1,59 @@ + + + + + + org.apache.geaflow + geaflow-dsl-connector + 0.8.0-SNAPSHOT + + 4.0.0 + + geaflow-dsl-connector-redis + geaflow-dsl-connector-redis + + + + org.apache.geaflow + geaflow-dsl-common + + + org.apache.geaflow + geaflow-dsl-connector-api + + + redis.clients + jedis + + + com.github.fppt + jedis-mock + test + + + org.testng + testng + ${testng.version} + test + + + diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisClient.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisClient.java new file mode 100644 index 000000000..67c22ddac --- /dev/null +++ b/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); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisConfigKeys.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisConfigKeys.java new file mode 100644 index 000000000..85131a11f --- /dev/null +++ b/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."); +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisDataType.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisDataType.java new file mode 100644 index 000000000..c394edf4b --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisDataType.java @@ -0,0 +1,36 @@ +/* + * 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 java.util.Locale; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; + +public enum RedisDataType { + STRING, + HASH; + + public static RedisDataType of(String value) { + try { + return RedisDataType.valueOf(value.toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + throw new GeaFlowDSLException("Unsupported redis data type: " + value, e); + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisRowConverter.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisRowConverter.java new file mode 100644 index 000000000..a3bdc1431 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisRowConverter.java @@ -0,0 +1,70 @@ +/* + * 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 java.math.BigDecimal; +import java.sql.Date; +import java.sql.Timestamp; +import org.apache.geaflow.common.binary.BinaryString; +import org.apache.geaflow.common.type.IType; +import org.apache.geaflow.common.type.Types; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; + +public class RedisRowConverter { + + public static String toRedisString(Object value) { + return value == null ? null : value.toString(); + } + + public static Object fromRedisString(String value, IType type) { + if (value == null) { + return null; + } + String typeName = type.getName(); + switch (typeName) { + case Types.TYPE_NAME_STRING: + return value; + case Types.TYPE_NAME_BINARY_STRING: + return BinaryString.fromString(value); + case Types.TYPE_NAME_BOOLEAN: + return Boolean.valueOf(value); + case Types.TYPE_NAME_BYTE: + return Byte.valueOf(value); + case Types.TYPE_NAME_SHORT: + return Short.valueOf(value); + case Types.TYPE_NAME_INTEGER: + return Integer.valueOf(value); + case Types.TYPE_NAME_LONG: + return Long.valueOf(value); + case Types.TYPE_NAME_FLOAT: + return Float.valueOf(value); + case Types.TYPE_NAME_DOUBLE: + return Double.valueOf(value); + case Types.TYPE_NAME_DECIMAL: + return new BigDecimal(value); + case Types.TYPE_NAME_TIMESTAMP: + return Timestamp.valueOf(value); + case Types.TYPE_NAME_DATE: + return Date.valueOf(value); + default: + throw new GeaFlowDSLException("Redis connector does not support type: " + typeName); + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableConnector.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableConnector.java new file mode 100644 index 000000000..b602b4c07 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableConnector.java @@ -0,0 +1,46 @@ +/* + * 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.Configuration; +import org.apache.geaflow.dsl.connector.api.TableReadableConnector; +import org.apache.geaflow.dsl.connector.api.TableSink; +import org.apache.geaflow.dsl.connector.api.TableSource; +import org.apache.geaflow.dsl.connector.api.TableWritableConnector; + +public class RedisTableConnector implements TableReadableConnector, TableWritableConnector { + + public static final String TYPE = "REDIS"; + + @Override + public String getType() { + return TYPE; + } + + @Override + public TableSource createSource(Configuration conf) { + return new RedisTableSource(); + } + + @Override + public TableSink createSink(Configuration conf) { + return new RedisTableSink(); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableSink.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableSink.java new file mode 100644 index 000000000..69460ac35 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableSink.java @@ -0,0 +1,135 @@ +/* + * 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 java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import org.apache.geaflow.api.context.RuntimeContext; +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.connector.api.TableSink; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; + +public class RedisTableSink implements TableSink { + + private static final Logger LOGGER = LoggerFactory.getLogger(RedisTableSink.class); + + private Configuration conf; + private StructType schema; + private RedisDataType dataType; + private String keyField; + private String valueField; + private String hashFieldField; + private String hashValueField; + private transient JedisPool jedisPool; + + @Override + public void init(Configuration tableConf, StructType schema) { + LOGGER.info("Prepare redis sink with config: {}, schema: {}", tableConf, schema); + this.conf = tableConf; + this.schema = schema; + this.dataType = RedisDataType.of( + tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_DATA_TYPE)); + this.keyField = tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_KEY_FIELD); + this.valueField = tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_VALUE_FIELD); + this.hashFieldField = tableConf.getString( + RedisConfigKeys.GEAFLOW_DSL_REDIS_HASH_FIELD_FIELD); + this.hashValueField = tableConf.getString( + RedisConfigKeys.GEAFLOW_DSL_REDIS_HASH_VALUE_FIELD); + requireField(keyField); + if (dataType == RedisDataType.STRING) { + requireField(valueField); + } else if (!hashFieldField.isEmpty() || !hashValueField.isEmpty()) { + requireField(hashFieldField); + requireField(hashValueField); + } + } + + @Override + public void open(RuntimeContext context) { + this.jedisPool = RedisClient.createJedisPool(conf); + } + + @Override + public void write(Row row) throws IOException { + try (Jedis jedis = jedisPool.getResource()) { + String key = fieldAsString(row, keyField); + if (key == null) { + throw new GeaFlowDSLException("Redis key field can not be null: " + keyField); + } + if (dataType == RedisDataType.STRING) { + jedis.set(key, fieldAsString(row, valueField)); + } else { + writeHash(jedis, key, row); + } + } catch (Exception e) { + throw new IOException("Failed to write row to Redis", e); + } + } + + @Override + public void finish() { + } + + @Override + public void close() { + if (Objects.nonNull(jedisPool)) { + jedisPool.close(); + } + } + + private void writeHash(Jedis jedis, String key, Row row) { + if (!hashFieldField.isEmpty()) { + String field = fieldAsString(row, hashFieldField); + String value = fieldAsString(row, hashValueField); + jedis.hset(key, field, value); + return; + } + Map values = new HashMap<>(); + for (int i = 0; i < schema.size(); i++) { + String fieldName = schema.getField(i).getName(); + if (!fieldName.equalsIgnoreCase(keyField)) { + values.put(fieldName, RedisRowConverter.toRedisString( + row.getField(i, schema.getType(i)))); + } + } + if (!values.isEmpty()) { + jedis.hset(key, values); + } + } + + private String fieldAsString(Row row, String fieldName) { + int index = schema.indexOf(fieldName); + return RedisRowConverter.toRedisString(row.getField(index, schema.getType(index))); + } + + private void requireField(String fieldName) { + if (fieldName == null || fieldName.isEmpty() || schema.indexOf(fieldName) < 0) { + throw new GeaFlowDSLException("Redis connector requires field: " + fieldName); + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableSource.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableSource.java new file mode 100644 index 000000000..5548dc131 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableSource.java @@ -0,0 +1,274 @@ +/* + * 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 java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import org.apache.geaflow.api.context.RuntimeContext; +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableSchema; +import org.apache.geaflow.dsl.connector.api.FetchData; +import org.apache.geaflow.dsl.connector.api.Offset; +import org.apache.geaflow.dsl.connector.api.Partition; +import org.apache.geaflow.dsl.connector.api.TableSource; +import org.apache.geaflow.dsl.connector.api.serde.TableDeserializer; +import org.apache.geaflow.dsl.connector.api.window.FetchWindow; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; +import redis.clients.jedis.ScanParams; +import redis.clients.jedis.ScanResult; + +public class RedisTableSource implements TableSource { + + private Configuration conf; + private StructType schema; + private RedisDataType dataType; + private String keyPattern; + private int scanCount; + private String keyField; + private String valueField; + private String hashFieldField; + private String hashValueField; + private transient JedisPool jedisPool; + + @Override + public void init(Configuration tableConf, TableSchema tableSchema) { + this.conf = tableConf; + this.schema = tableSchema; + this.dataType = RedisDataType.of( + tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_DATA_TYPE)); + this.keyPattern = tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_KEY_PATTERN); + this.scanCount = tableConf.getInteger(RedisConfigKeys.GEAFLOW_DSL_REDIS_SCAN_COUNT); + this.keyField = tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_KEY_FIELD); + this.valueField = tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_VALUE_FIELD); + this.hashFieldField = tableConf.getString( + RedisConfigKeys.GEAFLOW_DSL_REDIS_HASH_FIELD_FIELD); + this.hashValueField = tableConf.getString( + RedisConfigKeys.GEAFLOW_DSL_REDIS_HASH_VALUE_FIELD); + } + + @Override + public void open(RuntimeContext context) { + this.jedisPool = RedisClient.createJedisPool(conf); + } + + @Override + public List listPartitions() { + return Collections.singletonList(new RedisPartition()); + } + + @Override + public List listPartitions(int parallelism) { + List partitions = new ArrayList<>(); + for (int i = 0; i < parallelism; i++) { + partitions.add(new RedisPartition(i, parallelism)); + } + return partitions; + } + + @Override + public TableDeserializer getDeserializer(Configuration conf) { + return (TableDeserializer) new RedisTableDeserializer(); + } + + @Override + public FetchData fetch(Partition partition, Optional startOffset, + FetchWindow windowInfo) throws IOException { + String cursor = startOffset.map(offset -> ((RedisOffset) offset).getCursor()) + .orElse(ScanParams.SCAN_POINTER_START); + ScanParams scanParams = new ScanParams().match(keyPattern).count(scanCount); + List records = new ArrayList<>(); + String nextCursor; + try (Jedis jedis = jedisPool.getResource()) { + ScanResult scanResult = jedis.scan(cursor, scanParams); + nextCursor = scanResult.getCursor(); + RedisPartition redisPartition = (RedisPartition) partition; + for (String key : scanResult.getResult()) { + if (redisPartition.accept(key)) { + records.addAll(readKey(jedis, key)); + } + } + } catch (Exception e) { + throw new IOException("Failed to fetch data from Redis", e); + } + boolean finish = ScanParams.SCAN_POINTER_START.equals(nextCursor); + return (FetchData) FetchData.createStreamFetch((List) records, + new RedisOffset(nextCursor), finish); + } + + @Override + public void close() { + if (Objects.nonNull(jedisPool)) { + jedisPool.close(); + } + } + + private List readKey(Jedis jedis, String key) { + if (dataType == RedisDataType.STRING) { + String value = jedis.get(key); + return Collections.singletonList( + new RedisRecord(key, null, value, RedisDataType.STRING)); + } + List records = new ArrayList<>(); + for (java.util.Map.Entry entry : jedis.hgetAll(key).entrySet()) { + records.add(new RedisRecord(key, entry.getKey(), entry.getValue(), RedisDataType.HASH)); + } + return records; + } + + private class RedisTableDeserializer implements TableDeserializer { + + @Override + public void init(Configuration conf, StructType schema) { + } + + @Override + public List deserialize(RedisRecord record) { + Object[] values = new Object[schema.size()]; + for (int i = 0; i < schema.size(); i++) { + String fieldName = schema.getField(i).getName(); + String value = fieldValue(record, fieldName); + values[i] = RedisRowConverter.fromRedisString(value, schema.getType(i)); + } + return Collections.singletonList(ObjectRow.create(values)); + } + } + + private String fieldValue(RedisRecord record, String fieldName) { + if (fieldName.equalsIgnoreCase(keyField)) { + return record.getKey(); + } + if (dataType == RedisDataType.STRING && fieldName.equalsIgnoreCase(valueField)) { + return record.getValue(); + } + if (dataType == RedisDataType.HASH) { + if (!hashFieldField.isEmpty() && fieldName.equalsIgnoreCase(hashFieldField)) { + return record.getField(); + } + if (!hashValueField.isEmpty() && fieldName.equalsIgnoreCase(hashValueField)) { + return record.getValue(); + } + if (fieldName.equalsIgnoreCase(record.getField())) { + return record.getValue(); + } + } + if ("redis_type".equalsIgnoreCase(fieldName)) { + return record.getDataType().name().toLowerCase(); + } + throw new GeaFlowDSLException("Cannot map redis record to field: " + fieldName); + } + + public static class RedisRecord { + + private final String key; + private final String field; + private final String value; + private final RedisDataType dataType; + + public RedisRecord(String key, String field, String value, RedisDataType dataType) { + this.key = key; + this.field = field; + this.value = value; + this.dataType = dataType; + } + + public String getKey() { + return key; + } + + public String getField() { + return field; + } + + public String getValue() { + return value; + } + + public RedisDataType getDataType() { + return dataType; + } + } + + public static class RedisPartition implements Partition { + + private int index; + private int parallel; + + public RedisPartition() { + this(0, 1); + } + + public RedisPartition(int index, int parallel) { + this.index = index; + this.parallel = parallel; + } + + @Override + public String getName() { + return "redis-" + index + "-" + parallel; + } + + @Override + public void setIndex(int index, int parallel) { + this.index = index; + this.parallel = parallel; + } + + public boolean accept(String key) { + return Math.floorMod(key.hashCode(), parallel) == index; + } + } + + public static class RedisOffset implements Offset { + + private final String cursor; + + public RedisOffset(String cursor) { + this.cursor = cursor; + } + + public String getCursor() { + return cursor; + } + + @Override + public String humanReadable() { + return "RedisOffset{cursor='" + cursor + "'}"; + } + + @Override + public long getOffset() { + return Long.parseLong(cursor); + } + + @Override + public boolean isTimestamp() { + return false; + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector new file mode 100644 index 000000000..1518b5d6e --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector @@ -0,0 +1,18 @@ +# 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. + +org.apache.geaflow.dsl.connector.redis.RedisTableConnector diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/test/java/org/apache/geaflow/dsl/connector/redis/RedisTableConnectorTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/test/java/org/apache/geaflow/dsl/connector/redis/RedisTableConnectorTest.java new file mode 100644 index 000000000..0d0e777dc --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/test/java/org/apache/geaflow/dsl/connector/redis/RedisTableConnectorTest.java @@ -0,0 +1,159 @@ +/* + * 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 static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + +import com.github.fppt.jedismock.RedisServer; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.common.type.Types; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.types.TableField; +import org.apache.geaflow.dsl.common.types.TableSchema; +import org.apache.geaflow.dsl.connector.api.FetchData; +import org.apache.geaflow.dsl.connector.api.Partition; +import org.apache.geaflow.dsl.connector.api.serde.TableDeserializer; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import redis.clients.jedis.Jedis; + +public class RedisTableConnectorTest { + + private RedisServer redisServer; + private Configuration conf; + + @BeforeMethod + public void setUp() throws Exception { + redisServer = RedisServer.newRedisServer().start(); + conf = new Configuration(); + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_HOST, redisServer.getHost()); + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_PORT, String.valueOf(redisServer.getBindPort())); + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_KEY_PATTERN, "test:*"); + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_SCAN_COUNT, "100"); + } + + @AfterMethod + public void tearDown() throws Exception { + redisServer.stop(); + } + + @Test + public void testStringSinkAndSource() throws Exception { + TableSchema schema = new TableSchema( + new TableField("redis_key", Types.STRING), + new TableField("redis_value", Types.STRING)); + + RedisTableSink sink = new RedisTableSink(); + sink.init(conf, schema); + sink.open(null); + sink.write(ObjectRow.create("test:string:1", "hello")); + sink.close(); + + try (Jedis jedis = new Jedis(redisServer.getHost(), redisServer.getBindPort())) { + assertEquals(jedis.get("test:string:1"), "hello"); + } + + RedisTableSource source = new RedisTableSource(); + source.init(conf, schema); + source.open(null); + List rows = readRows(source, schema); + source.close(); + + assertEquals(rows.size(), 1); + assertEquals(rows.get(0).getField(0, Types.STRING), "test:string:1"); + assertEquals(rows.get(0).getField(1, Types.STRING), "hello"); + } + + @Test + public void testHashEntrySinkAndSource() throws Exception { + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_DATA_TYPE, "hash"); + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_HASH_FIELD_FIELD, "hash_field"); + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_HASH_VALUE_FIELD, "hash_value"); + TableSchema schema = new TableSchema( + new TableField("redis_key", Types.STRING), + new TableField("hash_field", Types.STRING), + new TableField("hash_value", Types.STRING)); + + RedisTableSink sink = new RedisTableSink(); + sink.init(conf, schema); + sink.open(null); + sink.write(ObjectRow.create("test:hash:1", "name", "geaflow")); + sink.write(ObjectRow.create("test:hash:1", "version", "0.8.0")); + sink.close(); + + try (Jedis jedis = new Jedis(redisServer.getHost(), redisServer.getBindPort())) { + assertEquals(jedis.hget("test:hash:1", "name"), "geaflow"); + assertEquals(jedis.hget("test:hash:1", "version"), "0.8.0"); + } + + RedisTableSource source = new RedisTableSource(); + source.init(conf, schema); + source.open(null); + List rows = readRows(source, schema); + source.close(); + + assertEquals(rows.size(), 2); + assertTrue(rows.stream().anyMatch(row -> "name".equals(row.getField(1, Types.STRING)))); + assertTrue(rows.stream().anyMatch(row -> "version".equals(row.getField(1, Types.STRING)))); + } + + @Test + public void testHashColumnSink() throws Exception { + conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_DATA_TYPE, "hash"); + TableSchema schema = new TableSchema( + new TableField("redis_key", Types.STRING), + new TableField("name", Types.STRING), + new TableField("age", Types.INTEGER)); + + RedisTableSink sink = new RedisTableSink(); + sink.init(conf, schema); + sink.open(null); + sink.write(ObjectRow.create("test:hash:2", "geaflow", 10)); + sink.close(); + + try (Jedis jedis = new Jedis(redisServer.getHost(), redisServer.getBindPort())) { + assertEquals(jedis.hget("test:hash:2", "name"), "geaflow"); + assertEquals(jedis.hget("test:hash:2", "age"), "10"); + } + } + + private List readRows(RedisTableSource source, TableSchema schema) throws Exception { + List partitions = source.listPartitions(); + FetchData fetchData = source.fetch(partitions.get(0), + java.util.Optional.empty(), null); + assertTrue(fetchData.isFinish()); + assertFalse(fetchData.getDataSize() < 0); + TableDeserializer deserializer = source.getDeserializer(conf); + deserializer.init(conf, schema); + List rows = new ArrayList<>(); + Iterator iterator = fetchData.getDataIterator(); + while (iterator.hasNext()) { + rows.addAll(deserializer.deserialize(iterator.next())); + } + return rows; + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml index e1f262b9d..b295968d6 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml @@ -49,6 +49,7 @@ geaflow-dsl-connector-paimon geaflow-dsl-connector-neo4j geaflow-dsl-connector-elasticsearch + geaflow-dsl-connector-redis