[Feature-2582][Connector] Add Hashes && Lists in Redis Sink

docs/en/connector/sink/Redis.md

@@ -54,7 +54,7 @@ Redis timeout
 
 ### data_type [string]
 
-Redis data type eg: `KV HASH LIST SET ZSET`
+Redis data type eg: `KV HASH LIST SET ZSET HASHES LISTS`
 
 ### hash_name [string]
 

seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-redis/src/main/scala/org/apache/seatunnel/spark/redis/common/RedisDataType.scala

@@ -22,7 +22,9 @@ object RedisDataType extends Enumeration {
 
   val KV: Value = Value("KV")
   val HASH: Value = Value("HASH")
+  val HASHES: Value = Value("HASHES")
   val LIST: Value = Value("LIST")
+  val LISTS: Value = Value("LISTS")
   val SET: Value = Value("SET")
   val ZSET: Value = Value("ZSET")
 }

seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-redis/src/main/scala/org/apache/seatunnel/spark/redis/sink/Redis.scala

@@ -48,16 +48,18 @@ class Redis extends SparkBatchSink with Logging {
       case RedisDataType.SET => dealWithSet(data, config.getString(SET_NAME), config.getInt(TTL))(sc = sc, redisConfig = redisConfigs)
       case RedisDataType.ZSET => dealWithZSet(data, config.getString(ZSET_NAME), config.getInt(TTL))(sc = sc, redisConfig = redisConfigs)
       case RedisDataType.LIST => dealWithList(data, config.getString(LIST_NAME), config.getInt(TTL))(sc = sc, redisConfig = redisConfigs)
+      case RedisDataType.HASHES => dealWithHASHes(data, config.getInt(TTL))(sc = sc, redisConfig = redisConfigs)
+      case RedisDataType.LISTS => dealWithLists(data, config.getInt(TTL))(sc = sc, redisConfig = redisConfigs)
     }
   }
 
   override def checkConfig(): CheckResult = {
     if (config.hasPath(DATA_TYPE)) {
       val dataType = config.getString(DATA_TYPE)
-      val dataTypeList = List("KV", "HASH", "SET", "ZSET", "LIST")
+      val dataTypeList = List("KV", "HASH", "SET", "ZSET", "LIST", "HASHES", "LISTS")
       val bool = dataTypeList.contains(dataType.toUpperCase)
       if (!bool) {
-        CheckResult.error("Unknown redis config. data_type must be in [KV HASH SET ZSET LIST]")
+        CheckResult.error("Unknown redis config. data_type must be in [KV HASH SET ZSET LIST HASHES LISTS]")
       } else {
         CheckResult.success()
       }
@@ -106,5 +108,15 @@ class Redis extends SparkBatchSink with Logging {
     sc.toRedisHASH(value, hashName, ttl)(redisConfig = redisConfig)
   }
 
+  def dealWithHASHes(data: Dataset[Row], ttl: Int)(implicit sc: SparkContext, redisConfig: RedisConfig): Unit = {
+    val value = data.rdd.map(x => (x.getString(0), Map(x.getString(1) -> x.getString(2))))
+    sc.toRedisHASHes(value, ttl)(redisConfig = redisConfig)
+  }
+
+  def dealWithLists(data: Dataset[Row], ttl: Int)(implicit sc: SparkContext, redisConfig: RedisConfig): Unit = {
+    val value = data.rdd.map(x => (x.getString(0), scala.Seq(x.getString(1))))
+    sc.toRedisLISTs(value, ttl)(redisConfig = redisConfig)
+  }
+
   override def getPluginName: String = "Redis"
 }

seatunnel-e2e/seatunnel-spark-e2e/pom.xml

@@ -49,6 +49,11 @@
             <artifactId>postgresql</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

seatunnel-e2e/seatunnel-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/redis/FakeSourceToRedisIT.java

@@ -0,0 +1,80 @@
+/*
+ * 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.seatunnel.e2e.spark.redis;
+
+import static org.testcontainers.shaded.org.awaitility.Awaitility.given;
+
+import org.apache.seatunnel.e2e.spark.SparkContainer;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.shaded.com.google.common.collect.Lists;
+import redis.clients.jedis.Jedis;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+@Slf4j
+public class FakeSourceToRedisIT extends SparkContainer {
+    private static final String REDIS_IMAGE = "redis:latest";
+    private static final String REDIS_CONTAINER_HOST = "spark_e2e_redis";
+    private static final String REDIS_HOST = "localhost";
+    private static final int REDIS_PORT = 6379;
+    private GenericContainer<?> redisContainer;
+    private Jedis jedis;
+
+    @BeforeEach
+    public void startRedisContainer() {
+        redisContainer = new GenericContainer<>(REDIS_IMAGE)
+                .withNetwork(NETWORK)
+                .withNetworkAliases(REDIS_CONTAINER_HOST)
+                .withLogConsumer(new Slf4jLogConsumer(log));
+        redisContainer.setPortBindings(Lists.newArrayList(String.format("%s:%s", REDIS_PORT, REDIS_PORT)));
+        Startables.deepStart(Stream.of(redisContainer)).join();
+        log.info("Redis container started");
+        given().ignoreExceptions()
+                .await()
+                .atMost(180, TimeUnit.SECONDS)
+                .untilAsserted(this::initJedis);
+    }
+
+    private void initJedis() {
+        jedis = new Jedis(REDIS_HOST, REDIS_PORT);
+    }
+
+    @Test
+    public void testFakeSourceToRedisSink() throws IOException, InterruptedException {
+        Container.ExecResult execResult = executeSeaTunnelSparkJob("/jdbc/fakesource_to_redis.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+    }
+
+    @AfterEach
+    public void close() {
+        super.close();
+        jedis.close();
+        redisContainer.close();
+    }
+}

seatunnel-e2e/seatunnel-spark-e2e/src/test/resources/redis/fakesource_to_redis.conf

@@ -0,0 +1,61 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  spark.app.name = "SeaTunnel"
+  spark.executor.instances = 2
+  spark.executor.cores = 1
+  spark.executor.memory = "1g"
+  spark.master = local
+}
+
+source {
+  # This is a example source plugin **only for test and demonstrate the feature source plugin**
+    Fake {
+      result_table_name = "my_dataset"
+      schema = {
+        fields {
+          key = "string"
+          hash_key = "string"
+          item = "string"
+        }
+      }
+    }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+  redis {
+      host = "localhost"
+      port = 6379
+      data_type = "HASHES"
+      db_num = 0
+    }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+}