Skip to content

Commit

Permalink
[Feature][seatunnel-transforms] Add Uuid transform for spark and fix …
Browse files Browse the repository at this point in the history
…some error at replace doc

a

a
  • Loading branch information
xiaowu committed Apr 28, 2022
1 parent 0a611da commit 19dbcde
Show file tree
Hide file tree
Showing 7 changed files with 257 additions and 5 deletions.
10 changes: 5 additions & 5 deletions docs/en/transform/replace.md
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# Json
# Replace

## Description

Expand Down Expand Up @@ -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]

Expand Down
62 changes: 62 additions & 0 deletions docs/en/transform/uuid.md
Original file line number Diff line number Diff line change
@@ -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"
}
```
6 changes: 6 additions & 0 deletions seatunnel-core/seatunnel-core-spark/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
1 change: 1 addition & 0 deletions seatunnel-transforms/seatunnel-transforms-spark/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>

Expand Down
Original file line number Diff line number Diff line change
@@ -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>
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
/*
* 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 // secure & slow
val seed = for (_ <- 0 until 728) yield rand.nextInt
prng = new Well19937c(seed.toArray) // non-secure & fast
}
}

@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) {
this.prng = prng
}
}
Original file line number Diff line number Diff line change
@@ -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)
}
}

0 comments on commit 19dbcde

Please sign in to comment.