[Feature][seatunnel-transforms] Add Uuid transform for spark

docs/en/transform/replace.md

@@ -1,4 +1,4 @@
-# Json
+# Replace
 
 ## Description
 
@@ -33,13 +33,13 @@ The name of the field to replaced.
 
 The string to match.
 
-### is_regex [string]
+### replacement [string]
 
-Whether or not to interpret the pattern as a regex (true) or string literal (false).
+The replacement pattern (is_regex is true) or string literal (is_regex is false).
 
-### replacement [boolean]
+### is_regex [boolean]
 
-The replacement pattern (is_regex is true) or string literal (is_regex is false).
+Whether or not to interpret the pattern as a regex (true) or string literal (false).
 
 ### replace_first [boolean]
 

docs/en/transform/uuid.md

@@ -0,0 +1,62 @@
+# Uuid
+
+## Description
+
+Generate a universally unique identifier on a specified field.
+
+:::tip
+
+This transform **ONLY** supported by Spark.
+
+:::
+
+## Options
+
+| name           | type   | required | default value |
+| -------------- | ------ | -------- | ------------- |
+| fields         | string | yes      | -             |
+| prefix         | string | no       | -             |
+| secure         | boolean| no       | false         |
+
+### field [string]
+
+The name of the field to generate.
+
+### prefix [string]
+
+The prefix string constant to prepend to each generated UUID.
+
+### secure [boolean]
+
+the cryptographically secure algorithm can be comparatively slow
+The nonSecure algorithm uses a secure random seed but is otherwise deterministic
+
+### common options [string]
+
+Transform plugin common parameters, please refer to [Transform Plugin](common-options.mdx) for details
+
+## Examples
+
+```bash
+  uuid {
+    fields = "u"
+    prefix = "uuid-"
+    secure = true
+  }
+}
+```
+
+Use `Uuid` as udf in sql.
+
+```bash
+  Uuid {
+    fields = "u"
+    prefix = "uuid-"
+    secure = true
+  }
+
+  # Use the uuid function (confirm that the fake table exists)
+  sql {
+    sql = "select * from (select raw_message, uuid() as info_row from fake) t1"
+  }
+```

seatunnel-core/seatunnel-core-spark/pom.xml

@@ -74,6 +74,12 @@
             <artifactId>seatunnel-transform-spark-replace</artifactId>
             <version>${project.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-transform-spark-uuid</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

seatunnel-transforms/seatunnel-transforms-spark/pom.xml

@@ -34,6 +34,7 @@
         <module>seatunnel-transform-spark-json</module>
         <module>seatunnel-transform-spark-split</module>
         <module>seatunnel-transform-spark-replace</module>
+        <module>seatunnel-transform-spark-uuid</module>
         <module>seatunnel-transform-spark-sql</module>
     </modules>
 

seatunnel-transforms/seatunnel-transforms-spark/seatunnel-transform-spark-uuid/pom.xml

@@ -0,0 +1,55 @@
+<?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.seatunnel</groupId>
+        <artifactId>seatunnel-transforms-spark</artifactId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>seatunnel-transform-spark-uuid</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api-spark</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_${scala.binary.version}</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_${scala.binary.version}</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-transforms/seatunnel-transforms-spark/seatunnel-transform-spark-uuid/src/main/scala/org/apache/seatunnel/spark/transform/Uuid.scala

@@ -0,0 +1,85 @@
+/*
+ * 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.spark.transform
+
+import java.security.SecureRandom
+import java.util.UUID
+
+import scala.collection.JavaConversions._
+
+import com.google.common.annotations.VisibleForTesting
+import org.apache.commons.math3.random.{RandomGenerator, Well19937c}
+import org.apache.seatunnel.common.Constants
+import org.apache.seatunnel.common.config.CheckConfigUtil.checkAllExists
+import org.apache.seatunnel.common.config.CheckResult
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory
+import org.apache.seatunnel.spark.{BaseSparkTransform, SparkEnvironment}
+import org.apache.spark.sql.{Dataset, Row}
+import org.apache.spark.sql.expressions.UserDefinedFunction
+import org.apache.spark.sql.functions.{col, udf}
+
+class Uuid extends BaseSparkTransform {
+  private var prng: RandomGenerator = _
+
+  override def process(df: Dataset[Row], env: SparkEnvironment): Dataset[Row] = {
+    val key = config.getString("fields")
+
+    val func: UserDefinedFunction = udf(() => {
+      generate(config.getString("prefix"))
+    })
+    var filterDf = df.withColumn(Constants.ROW_TMP, func())
+    filterDf = filterDf.withColumn(key, col(Constants.ROW_TMP))
+    val ds = filterDf.drop(Constants.ROW_TMP)
+    if (func != null) {
+      env.getSparkSession.udf.register("Uuid", func)
+    }
+    ds
+  }
+
+  override def checkConfig(): CheckResult = {
+    checkAllExists(config, "fields")
+  }
+
+  override def prepare(env: SparkEnvironment): Unit = {
+    val defaultConfig = ConfigFactory.parseMap(Map("prefix" -> "", "secure" -> false))
+    config = config.withFallback(defaultConfig)
+
+    /**
+     * The secure algorithm can be comparatively slow.
+     * The new nonSecure algorithm never blocks and is much faster.
+     * The nonSecure algorithm uses a secure random seed but is otherwise deterministic,
+     * though it is one of the strongest uniform pseudo-random number generators known so far.
+     * thanks for whoschek@cloudera.com
+     */
+    if (config.getBoolean("secure")) {
+      val rand = new SecureRandom
+      val seed = for (_ <- 0 until 728) yield rand.nextInt
+      prng = new Well19937c(seed.toArray)
+    }
+  }
+
+  @VisibleForTesting
+  def generate(prefix: String): String = {
+    val uuid = if (prng == null) UUID.randomUUID else new UUID(prng.nextLong, prng.nextLong)
+    prefix + uuid
+  }
+
+  @VisibleForTesting
+  def setPrng(prng: RandomGenerator): Unit = {
+    this.prng = prng
+  }
+}

seatunnel-transforms/seatunnel-transforms-spark/seatunnel-transform-spark-uuid/src/test/scala/org/apache/seatunnel/spark/transform/TestUuid.scala

@@ -0,0 +1,44 @@
+/*
+ * 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.spark.transform
+
+import java.security.SecureRandom
+
+import junit.framework.TestCase.assertEquals
+import org.apache.commons.math3.random.Well19937c
+import org.junit.Test
+
+class TestUuid {
+  @Test
+  def testUuid() {
+    val uuid = new Uuid
+    assertEquals(36, uuid.generate("").length)
+    assertEquals(37, uuid.generate("u").length)
+  }
+
+  @Test
+  def testSecureUuid() {
+    val rand = new SecureRandom
+    val seed = for (_ <- 0 until 728) yield rand.nextInt
+    val prng = new Well19937c(seed.toArray)
+
+    val uuid = new Uuid
+    uuid.setPrng(prng)
+    assertEquals(36, uuid.generate("").length)
+    assertEquals(37, uuid.generate("u").length)
+  }
+}