This is an automated email from the ASF dual-hosted git repository.
zhangzp pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
The following commit(s) were added to refs/heads/master by this push:
new 6825291 [FLINK-29044] Add Transformer for DCT
6825291 is described below
commit 6825291e0a808474c89a0bf65ed43f0cf1034e79
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Wed Aug 24 21:15:51 2022 +0800
[FLINK-29044] Add Transformer for DCT
This closes #145.
---
.../flink/ml/examples/feature/DCTExample.java | 61 +++++++
flink-ml-lib/pom.xml | 27 +++
.../java/org/apache/flink/ml/feature/dct/DCT.java | 139 +++++++++++++++
.../org/apache/flink/ml/feature/dct/DCTParams.java | 45 +++++
.../java/org/apache/flink/ml/feature/DCTTest.java | 187 +++++++++++++++++++++
.../pyflink/examples/ml/feature/dct_example.py | 61 +++++++
flink-ml-python/pyflink/ml/lib/feature/dct.py | 74 ++++++++
.../pyflink/ml/lib/feature/tests/test_dct.py | 89 ++++++++++
flink-ml-uber/src/main/resources/META-INF/NOTICE | 6 +
.../META-INF/licenses/LICENSE.JLargeArrays | 23 +++
.../META-INF/licenses/LICENSE.JTransforms | 23 +++
11 files changed, 735 insertions(+)
diff --git
a/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/DCTExample.java
b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/DCTExample.java
new file mode 100644
index 0000000..2b3d683
--- /dev/null
+++
b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/DCTExample.java
@@ -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.
+ */
+
+package org.apache.flink.ml.examples.feature;
+
+import org.apache.flink.ml.feature.dct.DCT;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Arrays;
+import java.util.List;
+
+/** Simple program that creates a DCT instance and uses it for feature
engineering. */
+public class DCTExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input data.
+ List<Vector> inputData =
+ Arrays.asList(
+ Vectors.dense(1.0, 1.0, 1.0, 1.0), Vectors.dense(1.0,
0.0, -1.0, 0.0));
+ Table inputTable =
tEnv.fromDataStream(env.fromCollection(inputData)).as("input");
+
+ // Creates a DCT object and initializes its parameters.
+ DCT dct = new DCT();
+
+ // Uses the DCT object for feature transformations.
+ Table outputTable = dct.transform(inputTable)[0];
+
+ // Extracts and displays the results.
+ for (CloseableIterator<Row> it = outputTable.execute().collect();
it.hasNext(); ) {
+ Row row = it.next();
+
+ Vector inputValue = row.getFieldAs(dct.getInputCol());
+ Vector outputValue = row.getFieldAs(dct.getOutputCol());
+
+ System.out.printf("Input Value: %s\tOutput Value: %s\n",
inputValue, outputValue);
+ }
+ }
+}
diff --git a/flink-ml-lib/pom.xml b/flink-ml-lib/pom.xml
index 9a0fc0b..42eff45 100644
--- a/flink-ml-lib/pom.xml
+++ b/flink-ml-lib/pom.xml
@@ -58,6 +58,12 @@ under the License.
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>com.github.wendykierp</groupId>
+ <artifactId>JTransforms</artifactId>
+ <version>3.1</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-loader</artifactId>
@@ -101,6 +107,27 @@ under the License.
<build>
<plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <includes combine.children="append">
+ <include>com.github.wendykierp:JTransforms</include>
+ <include>pl.edu.icm:JLargeArrays</include>
+ </includes>
+ </artifactSet>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/dct/DCT.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/dct/DCT.java
new file mode 100644
index 0000000..c343cd7
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/dct/DCT.java
@@ -0,0 +1,139 @@
+/*
+ * 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.flink.ml.feature.dct;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.Transformer;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.jtransforms.dct.DoubleDCT_1D;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.BiConsumer;
+
+/**
+ * A Transformer that takes the 1D discrete cosine transform of a real vector.
No zero padding is
+ * performed on the input vector. It returns a real vector of the same length
representing the DCT.
+ * The return vector is scaled such that the transform matrix is unitary (aka
scaled DCT-II).
+ *
+ * <p>See
https://en.wikipedia.org/wiki/Discrete_cosine_transform#DCT-II(DCT-II in
Discrete cosine
+ * transform).
+ */
+public class DCT implements Transformer<DCT>, DCTParams<DCT> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public DCT() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl)
inputs[0]).getTableEnvironment();
+
+ RowTypeInfo inputTypeInfo =
TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(
+ inputTypeInfo.getFieldTypes(),
DenseVectorTypeInfo.INSTANCE),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(),
getOutputCol()));
+
+ DataStream<Row> stream =
+ tEnv.toDataStream(inputs[0])
+ .map(new DCTFunction(getInputCol(), getInverse()),
outputTypeInfo);
+
+ return new Table[] {tEnv.fromDataStream(stream)};
+ }
+
+ /**
+ * A {@link MapFunction} that contains the main logic to perform discrete
cosine transformation.
+ */
+ private static class DCTFunction implements MapFunction<Row, Row> {
+ private final String inputCol;
+
+ private final boolean isInverse;
+
+ private BiConsumer<double[], Boolean> dctTransformer;
+
+ private long previousVectorSize;
+
+ private DCTFunction(String inputCol, boolean isInverse) {
+ this.inputCol = inputCol;
+ this.isInverse = isInverse;
+ this.dctTransformer = null;
+ this.previousVectorSize = -1;
+ }
+
+ @Override
+ public Row map(Row row) throws Exception {
+ Vector vector = row.getFieldAs(inputCol);
+
+ if (previousVectorSize != vector.size()) {
+ if (isInverse) {
+ dctTransformer = new DoubleDCT_1D(vector.size())::inverse;
+ } else {
+ dctTransformer = new DoubleDCT_1D(vector.size())::forward;
+ }
+ previousVectorSize = vector.size();
+ }
+
+ double[] array = vector.toArray();
+ if (vector instanceof DenseVector) {
+ array = array.clone();
+ }
+
+ dctTransformer.accept(array, true);
+
+ return Row.join(row, Row.of(Vectors.dense(array)));
+ }
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static DCT load(StreamTableEnvironment tEnv, String path) throws
IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/dct/DCTParams.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/dct/DCTParams.java
new file mode 100644
index 0000000..4410883
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/dct/DCTParams.java
@@ -0,0 +1,45 @@
+/*
+ * 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.flink.ml.feature.dct;
+
+import org.apache.flink.ml.common.param.HasInputCol;
+import org.apache.flink.ml.common.param.HasOutputCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params for {@link DCT}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface DCTParams<T> extends HasInputCol<T>, HasOutputCol<T> {
+ Param<Boolean> INVERSE =
+ new BooleanParam(
+ "inverse",
+ "Whether to perform the inverse DCT (true) or forward DCT
(false).",
+ false);
+
+ default boolean getInverse() {
+ return get(INVERSE);
+ }
+
+ default T setInverse(boolean value) {
+ return set(INVERSE, value);
+ }
+}
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/DCTTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/DCTTest.java
new file mode 100644
index 0000000..e2146e2
--- /dev/null
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/DCTTest.java
@@ -0,0 +1,187 @@
+/*
+ * 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.flink.ml.feature;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.feature.dct.DCT;
+import org.apache.flink.ml.linalg.SparseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/** Tests the {@link DCT}. */
+public class DCTTest extends AbstractTestBase {
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ private static final List<Vector> inputData =
+ Arrays.asList(Vectors.dense(1.0, 1.0, 1.0, 1.0),
Vectors.dense(1.0, 0.0, -1.0, 0.0));
+
+ private static final List<Row> expectedForwardOutputData =
+ Arrays.asList(
+ Row.of(Vectors.dense(1.0, 1.0, 1.0, 1.0),
Vectors.dense(2.0, 0.0, 0.0, 0.0)),
+ Row.of(
+ Vectors.dense(1.0, 0.0, -1.0, 0.0),
+ Vectors.dense(0.0, 0.924, 1.0, -0.383)));
+
+ private static final List<Row> expectedInverseOutputData =
+ Arrays.asList(
+ Row.of(
+ Vectors.dense(1.0, 1.0, 1.0, 1.0),
+ Vectors.dense(1.924, -0.383, 0.383, 0.076)),
+ Row.of(Vectors.dense(1.0, 0.0, -1.0, 0.0),
Vectors.dense(0.0, 1.0, 1.0, 0.0)));
+
+ private Table inputTable;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+
config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH,
true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(4);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputTable =
tEnv.fromDataStream(env.fromCollection(inputData)).as("input");
+ }
+
+ @Test
+ public void testParam() {
+ DCT dct = new DCT();
+
+ assertEquals("input", dct.getInputCol());
+ assertEquals("output", dct.getOutputCol());
+ assertFalse(dct.getInverse());
+
+
dct.setInputCol("test_input").setOutputCol("test_output").setInverse(true);
+
+ assertEquals("test_input", dct.getInputCol());
+ assertEquals("test_output", dct.getOutputCol());
+ assertTrue(dct.getInverse());
+ }
+
+ @Test
+ public void testOutputSchema() {
+ Table inputTable =
+
tEnv.fromDataStream(env.fromElements(Row.of(Vectors.dense(0.0), "")))
+ .as("test_input", "dummy_input");
+
+ DCT dct = new
DCT().setInputCol("test_input").setOutputCol("test_output");
+
+ Table outputTable = dct.transform(inputTable)[0];
+
+ assertEquals(
+ Arrays.asList(dct.getInputCol(), "dummy_input",
dct.getOutputCol()),
+ outputTable.getResolvedSchema().getColumnNames());
+ }
+
+ @Test
+ public void testTransformForward() {
+ DCT dct = new DCT();
+ Table outputTable = dct.transform(inputTable)[0];
+
+ verifyTransformResult(
+ outputTable, expectedForwardOutputData, dct.getInputCol(),
dct.getOutputCol());
+ }
+
+ @Test
+ public void testTransformInverse() {
+ DCT dct = new DCT().setInverse(true);
+ Table outputTable = dct.transform(inputTable)[0];
+
+ verifyTransformResult(
+ outputTable, expectedInverseOutputData, dct.getInputCol(),
dct.getOutputCol());
+ }
+
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ inputTable = TestUtils.convertDataTypesToSparseInt(tEnv, inputTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
TestUtils.getColumnDataTypes(inputTable));
+
+ DCT dct = new DCT();
+ Table outputTable = dct.transform(inputTable)[0];
+
+ verifyTransformResult(
+ outputTable, expectedForwardOutputData, dct.getInputCol(),
dct.getOutputCol());
+ }
+
+ @Test
+ public void testSaveLoadAndTransform() throws Exception {
+ DCT dct = new DCT().setInverse(true);
+
+ DCT loadedDCT =
+ TestUtils.saveAndReload(tEnv, dct,
TEMPORARY_FOLDER.newFolder().getAbsolutePath());
+
+ Table outputTable = loadedDCT.transform(inputTable)[0];
+
+ verifyTransformResult(
+ outputTable,
+ expectedInverseOutputData,
+ loadedDCT.getInputCol(),
+ loadedDCT.getOutputCol());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void verifyTransformResult(
+ Table outputTable, List<Row> expectedOutputData, String inputCol,
String outputCol) {
+ List<Row> actualOutputData =
IteratorUtils.toList(outputTable.execute().collect());
+ actualOutputData.sort(
+ Comparator.comparingLong(
+ x ->
+ ((Vector)
Objects.requireNonNull(x.getField(inputCol)))
+ .toDense()
+ .hashCode()));
+
+ expectedOutputData.sort(
+ Comparator.comparingLong(
+ x ->
+ ((Vector)
Objects.requireNonNull(x.getField(0)))
+ .toDense()
+ .hashCode()));
+
+ assertEquals(actualOutputData.size(), expectedOutputData.size());
+ for (int i = 0; i < actualOutputData.size(); i++) {
+ Vector actualVector =
actualOutputData.get(i).getFieldAs(outputCol);
+ Vector expectedVector = expectedOutputData.get(i).getFieldAs(1);
+ assertArrayEquals(expectedVector.toArray(),
actualVector.toArray(), 1e-3);
+ }
+ }
+}
diff --git a/flink-ml-python/pyflink/examples/ml/feature/dct_example.py
b/flink-ml-python/pyflink/examples/ml/feature/dct_example.py
new file mode 100644
index 0000000..1ecff6d
--- /dev/null
+++ b/flink-ml-python/pyflink/examples/ml/feature/dct_example.py
@@ -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.
+################################################################################
+
+# Simple program that creates a Bucketizer instance and uses it for feature
+# engineering.
+#
+# Before executing this program, please make sure you have followed Flink ML's
+# quick start guideline to set up Flink ML and Flink environment. The guideline
+# can be found at
+#
+#
https://nightlies.apache.org/flink/flink-ml-docs-master/docs/try-flink-ml/quick-start/
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo
+from pyflink.ml.lib.feature.dct import DCT
+from pyflink.table import StreamTableEnvironment
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input data
+input_data = t_env.from_data_stream(
+ env.from_collection([
+ (Vectors.dense(1.0, 1.0, 1.0, 1.0),),
+ (Vectors.dense(1.0, 0.0, -1.0, 0.0),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['input'],
+ [DenseVectorTypeInfo()])))
+
+# create a DCT object and initialize its parameters
+dct = DCT()
+
+# use the dct for feature engineering
+output = dct.transform(input_data)[0]
+
+# extract and display the results
+field_names = output.get_schema().get_field_names()
+for result in t_env.to_data_stream(output).execute_and_collect():
+ input_value = result[field_names.index(dct.get_input_col())]
+ output_value = result[field_names.index(dct.get_output_col())]
+ print('Input Value: ' + str(input_value) + '\tOutput Value: ' +
str(output_value))
diff --git a/flink-ml-python/pyflink/ml/lib/feature/dct.py
b/flink-ml-python/pyflink/ml/lib/feature/dct.py
new file mode 100644
index 0000000..ce431fd
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/feature/dct.py
@@ -0,0 +1,74 @@
+################################################################################
+# 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.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, BooleanParam
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.feature.common import JavaFeatureTransformer
+from pyflink.ml.lib.param import HasInputCol, HasOutputCol
+
+
+class _DCTParams(
+ JavaWithParams,
+ HasInputCol,
+ HasOutputCol
+):
+ """
+ Params for :class:`DCT`.
+ """
+
+ INVERSE: Param[bool] = BooleanParam(
+ "inverse",
+ "Whether to perform the inverse DCT (true) or forward DCT (false).",
+ False)
+
+ def __init__(self, java_params):
+ super(_DCTParams, self).__init__(java_params)
+
+ def set_inverse(self, value: bool):
+ return typing.cast(_DCTParams, self.set(self.INVERSE, value))
+
+ def get_inverse(self) -> bool:
+ return self.get(self.INVERSE)
+
+ @property
+ def inverse(self):
+ return self.get_inverse()
+
+
+class DCT(JavaFeatureTransformer, _DCTParams):
+ """
+ A Transformer that takes the 1D discrete cosine transform of a real vector.
+ No zero padding is performed on the input vector. It returns a real vector
+ of the same length representing the DCT. The return vector is scaled such
+ that the transform matrix is unitary (aka scaled DCT-II).
+
+ See https://en.wikipedia.org/wiki/Discrete_cosine_transform#DCT-II (DCT-II
+ in Discrete cosine transform).
+ """
+
+ def __init__(self, java_model=None):
+ super(DCT, self).__init__(java_model)
+
+ @classmethod
+ def _java_transformer_package_name(cls) -> str:
+ return "dct"
+
+ @classmethod
+ def _java_transformer_class_name(cls) -> str:
+ return "DCT"
diff --git a/flink-ml-python/pyflink/ml/lib/feature/tests/test_dct.py
b/flink-ml-python/pyflink/ml/lib/feature/tests/test_dct.py
new file mode 100644
index 0000000..08c2b0c
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/feature/tests/test_dct.py
@@ -0,0 +1,89 @@
+################################################################################
+# 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.
+################################################################################
+import os
+from typing import List
+
+from pyflink.common import Row, Types
+from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo
+from pyflink.ml.lib.feature.dct import DCT
+from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
+from pyflink.table import Table
+
+
+class DCTTest(PyFlinkMLTestCase):
+ def setUp(self):
+ super(DCTTest, self).setUp()
+ self.input = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (Vectors.dense(1.0, 1.0, 1.0, 1.0),),
+ (Vectors.dense(1.0, 0.0, -1.0, 0.0),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['input'],
+ [DenseVectorTypeInfo()])))
+
+ self.expected_output = [
+ Row(Vectors.dense(1.0, 1.0, 1.0, 1.0), Vectors.dense(2.0, 0.0,
0.0, 0.0)),
+ Row(Vectors.dense(1.0, 0.0, -1.0, 0.0), Vectors.dense(0.0, 0.924,
1.0, -0.383))]
+
+ def test_param(self):
+ dct = DCT()
+
+ self.assertEqual('input', dct.get_input_col())
+ self.assertEqual('output', dct.get_output_col())
+ self.assertFalse(dct.get_inverse())
+
+ dct.set_input_col('test_input') \
+ .set_output_col('test_output') \
+ .set_inverse(True)
+
+ self.assertEqual('test_input', dct.get_input_col())
+ self.assertEqual('test_output', dct.get_output_col())
+ self.assertTrue(dct.get_inverse())
+
+ def test_output_schema(self):
+ temp_table = self.input.alias('test_input')
+ dct = DCT().set_input_col('test_input').set_output_col('test_output')
+ output = dct.transform(temp_table)[0]
+ self.assertEqual(['test_input', 'test_output'],
output.get_schema().get_field_names())
+
+ def test_transform(self):
+ dct = DCT()
+ output = dct.transform(self.input)[0]
+ self.verify_output_result(output, self.expected_output)
+
+ def test_save_load_transform(self):
+ dct = DCT()
+ path = os.path.join(self.temp_dir, 'test_save_load_transform_dct')
+ dct.save(path)
+ dct = DCT.load(self.t_env, path)
+ output = dct.transform(self.input)[0]
+ self.verify_output_result(output, self.expected_output)
+
+ def verify_output_result(
+ self,
+ output: Table,
+ expected_result: List[Row]):
+ actual_result = [result for result in
+
self.t_env.to_data_stream(output).execute_and_collect()]
+ actual_result.sort(key=lambda x: hash(x[0]))
+ expected_result.sort(key=lambda x: hash(x[0]))
+
+ for item1, item2 in zip(actual_result, expected_result):
+ for i, j in zip(item1[1]._values, item2[1]._values):
+ self.assertAlmostEqual(i, j, delta=1e-3)
diff --git a/flink-ml-uber/src/main/resources/META-INF/NOTICE
b/flink-ml-uber/src/main/resources/META-INF/NOTICE
index ee0576d..2a47205 100644
--- a/flink-ml-uber/src/main/resources/META-INF/NOTICE
+++ b/flink-ml-uber/src/main/resources/META-INF/NOTICE
@@ -8,3 +8,9 @@ This project bundles the following dependencies under the MIT
license.
See bundled license files for details.
- dev.ludovic.netlib:blas:2.2.0
+
+This project bundles the following dependencies under the BSD license.
+See bundled license files for details.
+
+- com.github.wendykierp:JTransforms:3.1
+- pl.edu.icm:JLargeArrays:1.5
diff --git
a/flink-ml-uber/src/main/resources/META-INF/licenses/LICENSE.JLargeArrays
b/flink-ml-uber/src/main/resources/META-INF/licenses/LICENSE.JLargeArrays
new file mode 100644
index 0000000..a0bae8f
--- /dev/null
+++ b/flink-ml-uber/src/main/resources/META-INF/licenses/LICENSE.JLargeArrays
@@ -0,0 +1,23 @@
+JLargeArrays
+Copyright (C) 2013 onward University of Warsaw, ICM
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+ list of conditions and the following disclaimer.
+2. Redistributions in binary form must reproduce the above copyright notice,
+ this list of conditions and the following disclaimer in the documentation
+ and/or other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git
a/flink-ml-uber/src/main/resources/META-INF/licenses/LICENSE.JTransforms
b/flink-ml-uber/src/main/resources/META-INF/licenses/LICENSE.JTransforms
new file mode 100644
index 0000000..a03bcb7
--- /dev/null
+++ b/flink-ml-uber/src/main/resources/META-INF/licenses/LICENSE.JTransforms
@@ -0,0 +1,23 @@
+JTransforms
+Copyright (c) 2007 onward, Piotr Wendykier
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+ list of conditions and the following disclaimer.
+2. Redistributions in binary form must reproduce the above copyright notice,
+ this list of conditions and the following disclaimer in the documentation
+ and/or other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
